en/NlpInPracticeCourse/LanguageResourcesFromWeb: plagiarism_simple.py

#!/usr/bin/python3

"""
IA161, autumn 2023, homework on Building Language Resources from the Web.
This is a basic script for simple plagiarism detection that implements
relative frequency of words. You can add e.g. the following to improve it:
- change function doc_similarity,
- make it work with sentences rather than whole documents,
- add another method of duplicate detection,
- use multiple methods to decide if a sample is a duplicate.
Input: vertical (one token per line),
- structure <doc/> with attributes author, id, class, source,
- 3 columns: word, tag, lemma+POS.
Output:
- for each plagiarism document: id, id of detected source, id of the actual source,
- evaluation of precision, recall, F1-measure.
Usage (processing pipeline on asteria04):
ssh aurora.fi.muni.cz
ssh asteria04
cat *.txt | /opt/majka_pipe/majka-czech_v2.sh | cut -f1-3 | python3 plagiarism_simple.py
cat *.txt | /opt/treetagger_pipe/tt-english_v2.1.sh | python3 plagiarism_simple.py
"""

import sys, re
DOC_HEADER_RE = re.compile(
    '<doc author="([^"]+)" id="(\d+)" class="(plagiarism|original)" source="(\d+)"')

"""
All documents are read into the memory. There is a small number of data so it's ok.
A list of words with their counts is created for each document.
This is how the documents are represented -- with regards to further calculations.
You may use this or implement another representation more suitable to your solution.
"""
doc_sets = {} #sets of documents, each set of documents is by a single author
doc = {}
for line in sys.stdin:
    if line.startswith('<doc'):
        #document start
        #a data structure for document content and metadata is created here
        author, id_, class_, source_id = DOC_HEADER_RE.match(line).groups()
        doc = {
            'author': author,
            'id': id_,
            'class': class_,
            'source_id': source_id,
            'wordlist': {},
        }
    elif line.startswith('</doc'):
        #document end
        #the document is added to the set of its author to original or suspicious documents
        if not doc['author'] in doc_sets:
            doc_sets[doc['author']] = {'original': [], 'suspicious': []}
        if doc['class'] == 'original':
            doc_sets[doc['author']]['original'].append(doc)
        else:
            doc_sets[doc['author']]['suspicious'].append(doc)
    elif not line.startswith('<'):
        #token line
        #the occurrence of this word is added into the wordlist of the current document
        #you may want to e.g. try using the lemma or the tag instead of the wordform
        wordform, lemma, tag = line.rstrip().split('\t')[:3]
        doc['wordlist'][wordform] = doc['wordlist'].get(wordform, 0) + 1

"""
You can improve the following similarity function or implement it in a different way.
The function compares a pair of documents, e.g. a suspicious document with an original.
By default, the documents are represented by word-frequency vectors.
Their similarity is calculated as 1.0 minus the distance of their vector representation.
Each dimension of the vector corresponds to a word and the value is the frequency of the word.
This way, only the relative word count is important to represent a document.
Therefore e.g. the order of words is irrelevant in this method.
Output: Document similarity score between 0.0 and 1.0 where 1.0 = identical documents.
"""
DOC_SIMILARITY_THRESHOLD = 0.5
from scipy import spatial
def doc_similarity(doc1, doc2):
    vector1, vector2 = [], []
    all_words = list(doc1['wordlist'].keys()) + list(doc2['wordlist'].keys())
    doc1_len = float(sum(doc1['wordlist'].values()))
    doc2_len = float(sum(doc2['wordlist'].values()))
    for word in all_words:
        vector1.append(doc1['wordlist'].get(word, 0) / doc1_len)
        vector2.append(doc2['wordlist'].get(word, 0) / doc2_len)
    cosine_similarity = 1.0 - spatial.distance.cosine(vector1, vector2)
    return cosine_similarity

"""
Suspicious documents are compared to originals from the same set of documents
using the pair similarity function.
At the same moment, the performance of the method is evaluated by calculating
true positives (tp), true negatives (tn), false positives (fp) and false negatives (fn),
precision, recall and F-1 measure. See https://en.wikipedia.org/wiki/F-score.
"""
stats = {'tp': 0, 'fp': 0, 'tn': 0, 'fn': 0}
for author, doc_set in doc_sets.items():
    sys.stdout.write(u'Doc set by %s\n' % author)
    set_stats = {'tp': 0, 'fp': 0, 'tn': 0, 'fn': 0}
    for doc in doc_set['suspicious']:
        #a suspicious document is compared to all original documents from the same set
        #the ID of the most similar original document is stored
        most_similar_doc_id = None
        highest_similarity_score = 0.0
        plagiarism = False
        for orig_doc in doc_set['original']:
            similarity_score = doc_similarity(doc, orig_doc)
            if similarity_score >= DOC_SIMILARITY_THRESHOLD \
                    and similarity_score > highest_similarity_score:
                most_similar_doc_id = orig_doc['id']
                highest_similarity_score = similarity_score
                plagiarism = True
        sys.stdout.write(u'%s\t%s\t%s\n' % (doc['id'], most_similar_doc_id, doc['source_id']))
        #evaluation -- the current document
        if most_similar_doc_id == doc['source_id']:
            if doc['class'] == 'plagiarism':
                set_stats['tp'] += 1
            else:
                set_stats['tn'] += 1
        else:
            if doc['class'] == 'plagiarism':
                set_stats['fp'] += 1
            else:
                set_stats['fn'] += 1
    #evaluation -- the current set of documents by a single author
    try:
        precision = set_stats['tp'] / float(set_stats['tp'] + set_stats['fp'])
    except ZeroDivisionError:
        precision = 0.0
    try:
        recall = set_stats['tp'] / float(set_stats['tp'] + set_stats['fn'])
    except ZeroDivisionError:
        recall = 0.0
    try:
        f1_measure = 2 * precision * recall / (precision + recall)
    except ZeroDivisionError:
        f1_measure = 0.0
    sys.stdout.write(u'Set precision: %.2f, recall: %.2f, F1: %.2f\n\n' %
        (precision, recall, f1_measure))
    stats['tp'] += set_stats['tp']
    stats['fp'] += set_stats['fp']
    stats['tn'] += set_stats['tn']
    stats['fn'] += set_stats['fn']
#evaluation -- overall performance
try:
    precision = stats['tp'] / float(stats['tp'] + stats['fp'])
except ZeroDivisionError:
    precision = 0.0
try:
    recall = stats['tp'] / float(stats['tp'] + stats['fn'])
except ZeroDivisionError:
    recall = 0.0
try:
    f1_measure = 2 * precision * recall / (precision + recall)
except ZeroDivisionError:
    f1_measure = 0.0
sys.stdout.write(u'Overall precision: %.2f, recall: %.2f, F1: %.2f\n' %
    (precision, recall, f1_measure))