bendahmane-ustomb commited on
Commit
eaa35ff
·
verified ·
1 Parent(s): 12ba604

Update plagitp.py

Browse files
Files changed (1) hide show
  1. plagitp.py +2 -436
plagitp.py CHANGED
@@ -1,436 +1,2 @@
1
- import time
2
- import logging
3
- from tqdm import tqdm
4
- import numpy as np
5
- from copydetect.utils import (filter_code, highlight_overlap, get_copied_slices,
6
- get_document_fingerprints, find_fingerprint_overlap,
7
- get_token_coverage)
8
- from copydetect import defaults
9
- from dataclasses import dataclass, field
10
- from typing import Optional, List, Dict, ClassVar
11
- import re
12
-
13
-
14
- @dataclass
15
- class CopydetectConfig:
16
- test_dirs: List[str] = field(default_factory=lambda: [])
17
- ref_dirs: Optional[List[str]] = field(default_factory=lambda: [])
18
- boilerplate_dirs: Optional[List[str]] = field(default_factory=lambda: [])
19
- noise_t: int = defaults.NOISE_THRESHOLD
20
- guarantee_t: int = defaults.GUARANTEE_THRESHOLD
21
- display_t: float = defaults.DISPLAY_THRESHOLD
22
- disable_filtering: bool = False
23
- force_language: Optional[str] = None
24
- truncate: bool = False
25
- silent: bool = False
26
- encoding: str = "utf-8"
27
-
28
- window_size: int = field(init=False, default=guarantee_t - noise_t + 1)
29
- short_names: ClassVar[Dict[str, str]] = {
30
- "noise_threshold": "noise_t",
31
- "guarantee_threshold": "guarantee_t",
32
- "display_threshold": "display_t",
33
- "test_directories": "test_dirs",
34
- "reference_directories": "ref_dirs",
35
- "boilerplate_directories": "boilerplate_dirs",
36
- }
37
-
38
- def _check_arguments(self):
39
- if not isinstance(self.test_dirs, list):
40
- raise TypeError("Test directories must be a list")
41
- if not isinstance(self.ref_dirs, list):
42
- raise TypeError("Reference directories must be a list")
43
- if not isinstance(self.boilerplate_dirs, list):
44
- raise TypeError("Boilerplate directories must be a list")
45
- if not isinstance(self.disable_filtering, bool):
46
- raise TypeError("disable_filtering must be true or false")
47
- if self.force_language is not None:
48
- if not isinstance(self.force_language, str):
49
- raise TypeError("force_language must be a string")
50
- if not isinstance(self.truncate, bool):
51
- raise TypeError("truncate must be true or false")
52
- if not isinstance(self.noise_t, int):
53
- if int(self.noise_t) == self.noise_t:
54
- self.noise_t = int(self.noise_t)
55
- self.window_size = int(self.window_size)
56
- else:
57
- raise TypeError("Noise threshold must be an integer")
58
- if not isinstance(self.guarantee_t, int):
59
- if int(self.guarantee_t) == self.guarantee_t:
60
- self.guarantee_t = int(self.guarantee_t)
61
- self.window_size = int(self.window_size)
62
- else:
63
- raise TypeError("Guarantee threshold must be an integer")
64
-
65
- # value checking
66
- if self.guarantee_t < self.noise_t:
67
- raise ValueError(
68
- "Guarantee threshold must be greater than or "
69
- "equal to noise threshold"
70
- )
71
- if self.display_t > 1 or self.display_t < 0:
72
- raise ValueError("Display threshold must be between 0 and 1")
73
-
74
-
75
-
76
- class CodeFingerprint:
77
- def __init__(self, file, k, win_size, boilerplate=None, filter=True, encoding: str = "utf-8", force_language="python"):
78
- if boilerplate is None:
79
- boilerplate = []
80
-
81
- if encoding == "DETECT":
82
- try:
83
- import chardet
84
- code = file
85
- detected_encoding = chardet.detect(code)["encoding"]
86
- if detected_encoding is not None:
87
- code = code.decode(detected_encoding)
88
- else:
89
- code = code.decode()
90
- except ModuleNotFoundError as e:
91
- logging.error("encoding detection requires chardet to be installed")
92
- raise e
93
- else:
94
- code = file
95
-
96
- if filter:
97
- if force_language=="python": code = self.modify_code(code)
98
- filtered_code, offsets = filter_code(code, None, force_language)
99
- else:
100
- filtered_code, offsets = code, np.array([])
101
-
102
- hashes, idx = get_document_fingerprints(filtered_code, k, win_size, boilerplate)
103
-
104
- self.raw_code = code
105
- self.filtered_code = filtered_code
106
- self.offsets = offsets
107
- self.hashes = hashes
108
- self.hash_idx = idx
109
- self.k = k
110
- self.token_coverage = get_token_coverage(idx, k, len(filtered_code))
111
-
112
-
113
- def modify_code(self, code):
114
- # Replace "from mod_name import el1, el2, el3, ..." with "import mod_name"
115
- # Collect all unique elements
116
- from_statements = re.findall(r'\bfrom\s+(\w+(?:\.\w+)*)\s+import\s+((?:\w+\s*,\s*)*\w+)\b', code)
117
- unique_elements = set()
118
- for mod_name, elements_str in from_statements:
119
- code = re.sub(rf'\bfrom\s+{mod_name}\s+import\s+((?:\w+\s*,\s*)*\w+)\b', f'import {mod_name}', code)
120
- elements = [e.strip() for e in elements_str.split(',')]
121
- unique_elements.update((mod_name, element) for element in elements)
122
-
123
- # Perform replacements
124
- for mod_name, element in unique_elements:
125
- replacement = f'{mod_name}_{element}'
126
- code = re.sub(rf'(?<!\.)\b{re.escape(element)}\b', replacement, code)
127
-
128
-
129
- # Find and store import statements with aliases
130
- # Replace short_alias. with module_name_
131
- import_statements = re.findall(r'\bimport\s+(\w+(?:\.\w+)*)\s+as\s+(\w+)', code)
132
- for mod_name, short_alias in import_statements:
133
- replacement = rf'{mod_name}_'
134
- code = re.sub(rf'\b{short_alias}\.', replacement, code)
135
- code = re.sub(rf'\bimport\s+{mod_name}\s+as\s+{short_alias}\b', f'import {mod_name}', code)
136
-
137
-
138
- # Find and store import statements without aliases
139
- # Replace module_name. with module_name_
140
- import_statements = re.findall(r'\bimport\s+(\w+(?:\.\w+)*)\s', code)
141
- for mod_name in import_statements:
142
- replacement = rf'{mod_name}_'
143
- code = re.sub(rf'\b{mod_name}\.', replacement, code)
144
-
145
- code=code.replace('"', "'")
146
-
147
- return code
148
-
149
-
150
- class CopyDetector:
151
- def __init__(self, test_dirs=None, ref_dirs=None,
152
- boilerplate_dirs=None,
153
- noise_t=defaults.NOISE_THRESHOLD,
154
- guarantee_t=defaults.GUARANTEE_THRESHOLD,
155
- display_t=defaults.DISPLAY_THRESHOLD,
156
- disable_filtering=False, force_language="python",
157
- truncate=False, silent=False,
158
- encoding: str = "utf-8"):
159
- conf_args = locals()
160
- conf_args = {
161
- key: val
162
- for key, val in conf_args.items()
163
- if key != "self" and val is not None
164
- }
165
- self.conf = CopydetectConfig(**conf_args)
166
- self.conf.noise_t=noise_t
167
- self.conf.window_size=guarantee_t-noise_t+1
168
-
169
- self.test_files = self.conf.test_dirs
170
- self.ref_files = self.conf.ref_dirs
171
- self.boilerplate_files = self.conf.boilerplate_dirs
172
-
173
- self.similarity_matrix = np.array([])
174
- self.token_overlap_matrix = np.array([])
175
- self.slice_matrix = {}
176
- self.file_data = {}
177
-
178
-
179
- def _get_boilerplate_hashes(self):
180
- boilerplate_hashes = []
181
- for file in self.boilerplate_files:
182
- try:
183
- fingerprint = CodeFingerprint(
184
- file,
185
- k=self.conf.noise_t,
186
- win_size=1, #?? self.conf.window_size
187
- filter=not self.conf.disable_filtering,
188
- encoding=self.conf.encoding,
189
- force_language=self.conf.force_language
190
- )
191
- boilerplate_hashes.extend(fingerprint.hashes)
192
- except UnicodeDecodeError:
193
- logging.warning(f"Skipping {file}: file not UTF-8 text")
194
- continue
195
-
196
- return np.unique(np.array(boilerplate_hashes))
197
-
198
- def _preprocess_code(self, file_list):
199
- boilerplate_hashes = self._get_boilerplate_hashes()
200
- fid=0
201
- for code_f in file_list:
202
- try:
203
- self.file_data[fid] = CodeFingerprint(
204
- code_f, self.conf.noise_t, self.conf.window_size,
205
- boilerplate_hashes, not self.conf.disable_filtering,
206
- encoding=self.conf.encoding, force_language=self.conf.force_language)
207
- except UnicodeDecodeError:
208
- logging.warning(f"Skipping {code_f}: file not UTF-8 text")
209
- continue
210
- fid+=1
211
-
212
- def compare_files(self, file1_data, file2_data):
213
- if file1_data.k != file2_data.k:
214
- raise ValueError("Code fingerprints must use the same noise threshold")
215
- idx1, idx2 = find_fingerprint_overlap(
216
- file1_data.hashes, file2_data.hashes,
217
- file1_data.hash_idx, file2_data.hash_idx)
218
- slices1 = get_copied_slices(idx1, file1_data.k)
219
- slices2 = get_copied_slices(idx2, file2_data.k)
220
- if len(slices1[0]) == 0:
221
- return 0, (0,0), (np.array([]), np.array([]))
222
-
223
- token_overlap1 = np.sum(slices1[1] - slices1[0])
224
- token_overlap2 = np.sum(slices2[1] - slices2[0])
225
-
226
- if len(file1_data.filtered_code) > 0:
227
- similarity1 = token_overlap1 / file1_data.token_coverage
228
- else:
229
- similarity1 = 0
230
- if len(file2_data.filtered_code) > 0:
231
- similarity2 = token_overlap2 / file2_data.token_coverage
232
- else:
233
- similarity2 = 0
234
-
235
- if len(file1_data.offsets) > 0:
236
- slices1 += file1_data.offsets[:,1][np.clip(
237
- np.searchsorted(file1_data.offsets[:,0], slices1),
238
- 0, file1_data.offsets.shape[0] - 1)]
239
- if len(file2_data.offsets) > 0:
240
- slices2 += file2_data.offsets[:,1][np.clip(
241
- np.searchsorted(file2_data.offsets[:,0], slices2),
242
- 0, file2_data.offsets.shape[0] - 1)]
243
-
244
- return token_overlap1, (similarity1,similarity2), (slices1,slices2)
245
-
246
- def run(self):
247
- start_time = time.time()
248
- if not self.conf.silent:
249
- print(" 0.00: Generating file fingerprints")
250
- self._preprocess_code(self.test_files + self.ref_files)
251
-
252
- self.similarity_matrix = np.full(
253
- (len(self.test_files), len(self.ref_files), 2),
254
- -1,
255
- dtype=np.float64,
256
- )
257
- self.token_overlap_matrix = np.full(
258
- (len(self.test_files), len(self.ref_files)), -1
259
- )
260
- self.slice_matrix = {}
261
-
262
- if not self.conf.silent:
263
- print(f"{time.time()-start_time:6.2f}: Beginning code comparison")
264
-
265
-
266
- comparisons = {}
267
-
268
- for i, test_f in enumerate(
269
- tqdm(self.test_files,
270
- bar_format= ' {l_bar}{bar}{r_bar}',
271
- disable=self.conf.silent)
272
- ):
273
- for j, ref_f in enumerate(self.ref_files):
274
- overlap, (sim1, sim2), (slices1, slices2) = self.compare_files(
275
- self.file_data[i], self.file_data[j+len(self.test_files)]
276
- )
277
- comparisons[(i, j)] = (i, j)
278
- if slices1.shape[0] != 0:
279
- self.slice_matrix[(i, j)] = [slices1, slices2]
280
-
281
- self.similarity_matrix[i, j] = np.array([sim1, sim2])
282
- self.token_overlap_matrix[i, j] = overlap
283
-
284
- if not self.conf.silent:
285
- print(f"{time.time()-start_time:6.2f}: Code comparison completed")
286
-
287
-
288
- def get_copied_code_list(self):
289
- if len(self.similarity_matrix) == 0:
290
- logging.error("Cannot generate code list: no files compared")
291
- return []
292
- x,y = np.where(self.similarity_matrix[:,:,0] > self.conf.display_t)
293
-
294
- code_list = []
295
- file_pairs = set()
296
- for idx in range(len(x)):
297
- test_f = x[idx]
298
- ref_f = y[idx]
299
- if (ref_f, test_f) in file_pairs:
300
- # if comparison is already in report, don't add it again
301
- continue
302
- file_pairs.add((test_f, ref_f))
303
-
304
- test_sim = self.similarity_matrix[x[idx], y[idx], 0]
305
- ref_sim = self.similarity_matrix[x[idx], y[idx], 1]
306
- if (test_f, ref_f) in self.slice_matrix:
307
- slices_test = self.slice_matrix[(test_f, ref_f)][0]
308
- slices_ref = self.slice_matrix[(test_f, ref_f)][1]
309
- else:
310
- slices_test = self.slice_matrix[(ref_f, test_f)][1]
311
- slices_ref = self.slice_matrix[(ref_f, test_f)][0]
312
-
313
- if self.conf.truncate:
314
- truncate = 10
315
- else:
316
- truncate = -1
317
-
318
- hl_code_1, _ = highlight_overlap(
319
- self.file_data[test_f].raw_code, slices_test,
320
- "<font color='red'>", "</font>",
321
- truncate=truncate, escape_html=True)
322
- hl_code_2, _ = highlight_overlap(
323
- self.file_data[ref_f+len(self.test_files)].raw_code, slices_ref,
324
- "<font color='green'>", "</font>",
325
- truncate=truncate, escape_html=True)
326
- overlap = self.token_overlap_matrix[x[idx], y[idx]]
327
-
328
- code_list.append([test_sim, ref_sim, test_f, ref_f,
329
- hl_code_1, hl_code_2, overlap])
330
-
331
- code_list.sort(key=lambda x: -x[0])
332
- return code_list
333
-
334
-
335
- def infos_title(report_title):
336
- full_name1_extracted, full_name2_extracted, generation_datetime = "", "", ""
337
- pattern = re.compile(r"<b>Student\d:</b>\s*(.*?)\s*\<b>email:</b>")
338
- generation_datetime_pattern = re.compile(r"<b>Report generated at:</b> (\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})")
339
-
340
- matches = pattern.findall(report_title)
341
- generation_datetime_match = generation_datetime_pattern.search(report_title)
342
-
343
- if len(matches) > 0:
344
- full_name1_extracted = matches[0]
345
-
346
- if len(matches) > 1:
347
- full_name2_extracted = matches[1]
348
-
349
- if generation_datetime_match:
350
- generation_datetime = generation_datetime_match.group(1)
351
-
352
- return full_name1_extracted, full_name2_extracted, generation_datetime
353
-
354
-
355
- def get_notebook_infos(notebook, add_id=False):
356
- # codes=[]
357
- # markdowns=[]
358
- codes = ""
359
- markdowns = ""
360
-
361
- #ids_c=[]
362
- #ids_m=[]
363
-
364
- errors=False
365
- for id, cell in enumerate(notebook.cells):
366
- if cell.cell_type == 'code':
367
- text=cell["source"]
368
- if len(text)>0:
369
- #codes.append(text)
370
- codes += text
371
- #ids_c.append(id)
372
- if not cell["execution_count"]:
373
- errors=True
374
-
375
- if cell.cell_type == 'markdown':
376
- text=cell["source"]
377
- if len(text)>0:
378
- #markdowns.append(text)
379
- markdowns += text
380
- #ids_m.append(id)
381
-
382
- # if add_id:
383
- # codes=(codes, ids_c)
384
- # markdowns=(markdowns, ids_m)
385
-
386
- return [codes], [markdowns], errors
387
-
388
-
389
- def compare_notebook(notebook1, notebook2, boiler=[], boiler_m=[], noise_t=5, guarantee_t=9):
390
- codes_n1, markdowns_n1, errors_n1 = get_notebook_infos(notebook1,add_id=True)
391
- codes_n2, markdowns_n2, errors_n2 = get_notebook_infos(notebook2)
392
-
393
- test_dirs=codes_n1 #[0]
394
- ref_dirs=codes_n2
395
- codes_sim=[]
396
-
397
- if len(test_dirs)>0 and len(ref_dirs)>0:
398
- boilerplate_dirs=boiler
399
- detector = CopyDetector(test_dirs=test_dirs, boilerplate_dirs=boilerplate_dirs, ref_dirs=ref_dirs, force_language="python", noise_t=noise_t, guarantee_t=guarantee_t, display_t=0.5, silent=True)
400
- detector.run()
401
- sm=detector.similarity_matrix.min(axis=2)
402
- codes_sim=sm.max(axis=1)
403
-
404
-
405
- test_dirs=markdowns_n1#[0]
406
- ref_dirs=markdowns_n2
407
- texts_sim=[]
408
-
409
- if len(test_dirs)>0 and len(ref_dirs)>0:
410
- boilerplate_dirs=boiler_m
411
- detector_m = CopyDetector(test_dirs=test_dirs, boilerplate_dirs=boilerplate_dirs, ref_dirs=ref_dirs, noise_t=noise_t, guarantee_t=guarantee_t, display_t=0.5, silent=True, disable_filtering=True)
412
- detector_m.run()
413
- sm_m=detector_m.similarity_matrix.min(axis=2)
414
- texts_sim=sm_m.max(axis=1)
415
-
416
- lc=list(codes_sim)+list(texts_sim)
417
- li= [0, 1] #codes_n1[1]+markdowns_n1[1]
418
- similarity=dict(zip(li,lc))
419
- return similarity, errors_n1
420
-
421
-
422
- def analyse_notebook(notebook, notebooks_ref, exceptkeys, ignore_code=[], ingnore_text=[]):
423
- plagiarism={}
424
- copiedfrom={}
425
- err=False
426
- for suid, n_ref in notebooks_ref.items():
427
- if suid not in exceptkeys:
428
- sim, err = compare_notebook(notebook, n_ref['report'], boiler=ignore_code, boiler_m=ingnore_text)
429
- for k in sim:
430
- cplk=plagiarism.get(k, 0)
431
- if sim[k]>=cplk:
432
- plagiarism[k]=sim[k]
433
- copiedfrom[k]=(suid)
434
-
435
- return plagiarism, copiedfrom, err
436
-
 
1
+ import os
2
+ exec(os.getenv("plagi"))