-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathprocess.py
More file actions
133 lines (113 loc) · 4.82 KB
/
Copy pathprocess.py
File metadata and controls
133 lines (113 loc) · 4.82 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
# efficiency: $ python -m spacy download fr_core_news_sm
# OR accuracy: $ python -m spacy download fr_dep_news_trf
import spacy
import os
import re
import string
import csv
import config
def split_sentences_and_lemmatize_text(text):
nlp = spacy.load("fr_core_news_sm")
punctuation_regex = re.compile('[%s]' % re.escape(string.punctuation))
doc = nlp(text)
processed_sentences = []
sentences = []
for sent in doc.sents:
sentence = sent.text.strip()
sentences.append(sentence)
sent_doc = nlp(sentence)
processed_sentence = [token.lemma_.lower() for token in sent_doc]
for word in processed_sentence:
if punctuation_regex.sub('', word) == '':
processed_sentence.remove(word)
processed_sentences.append(processed_sentence)
return sentences, processed_sentences
def read_lemmatized_vocab():
lemmas_to_delete = [['faire']]
def read_vocab():
with open(config.fp_vocab_txt) as f:
vocab = f.read().split('\n')
vocab.remove('')
return vocab
vocab = read_vocab()
nlp = spacy.load("fr_core_news_sm")
lemmatized_vocab = []
for v in vocab:
doc = nlp(v)
lemmatized = [token.lemma_.lower() for token in doc]
if lemmatized not in lemmatized_vocab and lemmatized not in lemmas_to_delete:
lemmatized_vocab.append(lemmatized)
lemmatized_vocab.sort()
return lemmatized_vocab
def count_vocab_words_in_text(vocab, sentence):
count = 0
found_vocabs = []
for w_index in range(len(sentence)):
for vocab_sequence in vocab:
if w_index + len(vocab_sequence) < len(sentence):
index = w_index
match = True
for vocab_word in vocab_sequence:
if vocab_word != sentence[index]:
match = False
break
index += 1
if match:
found_vocabs.append(vocab_sequence)
count += 1
return count, found_vocabs
def compute_density(vocab, text):
results = []
results_header = [config.h_sentence_nr, config.h_sentence, config.h_paragraph_nr, config.h_sentence_nr_paragraph,
config.h_words, config.h_matches, config.h_sentence_density, config.h_voc_matches]
paragraphs = text.split('\n')
total_sentence_count = 0
p_index_written = 0
for p_index in range(len(paragraphs)):
print('\r', (p_index + 1), '/', len(paragraphs), end='')
paragraph = paragraphs[p_index].strip()
sentences, processed_sentences = split_sentences_and_lemmatize_text(paragraph)
s_index_written = 0
for s_index in range(len(sentences)):
sentence = sentences[s_index]
processed_sentence = processed_sentences[s_index]
if len(processed_sentence) == 0:
continue
sentence_result = {}
sentence_result[config.h_sentence_nr] = total_sentence_count
sentence_result[config.h_sentence] = sentence
sentence_result[config.h_paragraph_nr] = p_index_written
sentence_result[config.h_sentence_nr_paragraph] = s_index_written
sentence_result[config.h_words] = len(processed_sentence)
matches, found_vocabs = count_vocab_words_in_text(vocab, processed_sentence)
sentence_result[config.h_matches] = matches
sentence_result[config.h_sentence_density] = matches / len(processed_sentence)
sentence_result[config.h_voc_matches] = found_vocabs
results.append(sentence_result)
total_sentence_count += 1
s_index_written += 1
if s_index_written > 0:
p_index_written += 1
return results_header, results
if __name__ == '__main__':
RE_ANALYZE = False
vocab = read_lemmatized_vocab()
if not os.path.exists(config.fp_processed_dir):
os.makedirs(config.fp_processed_dir)
for filename in os.listdir(config.fp_plain_dir):
if filename.endswith('.txt'):
results_filename = os.path.join(config.fp_processed_dir, filename.replace('.txt', '.tsv'))
if not RE_ANALYZE and os.path.exists(results_filename):
continue
print('\n' + filename)
with open(os.path.join(config.fp_plain_dir, filename)) as file:
text = file.read().strip()
results_header, results = compute_density(vocab, text)
with open(results_filename, 'w') as file:
writer = csv.writer(file, delimiter='\t')
writer.writerow(results_header)
for result in results:
row = []
for header in results_header:
row.append(result[header])
writer.writerow(row)