Marcell / marcell-scripts

import argparse
import os
import re

import gcld3


def main():
    args = parse_arguments()
    if not args.input or not args.output:
        print('Error: Input and output must be selected!')
    copy_by_year(args.input, args.output)


def parse_arguments():
    parser = argparse.ArgumentParser(description='Clean conllup corpora texts from strange sentences.')
    parser.add_argument('-o', '--output', help='output directory')
    required_arguments = parser.add_argument_group('required arguments')
    required_arguments.add_argument('-i', '--input', help='corpora root directory', required=True)
    return parser.parse_args()


def copy_by_year(root_directory, out_corpora_directory):
    lang_detector = gcld3.NNetLanguageIdentifier(min_num_bytes=0, max_num_bytes=1000)
    for root, dirs, files in os.walk(root_directory):
        for filename in files:
            if filename.endswith('.conllup') or filename.endswith('.conllu'):
                src = os.path.join(root, filename)
                year = get_year(src)
                year_path = os.path.join(out_corpora_directory, year)
                os.makedirs(year_path, exist_ok=True)
                dst = os.path.join(year_path, filename)
                clean_and_save_full(src, dst, year, lang_detector)


def get_year(filepath):
    with open(filepath, 'r') as conllup_file:
        for line in conllup_file:
            line = line.strip()
            if is_segment(line):
                continue
            elif is_metadata(line):
                name, value = get_metadata(line)
                if name == 'date':
                    return value.split('-')[0]
    return 0


def is_segment(line):
    if line and line[0].isdigit():
        return True
    return False


def is_metadata(line):
    if line.startswith('#'):
        return True
    return False


def get_metadata(line):
    name_value_pair = line.split('=', 1)
    name = name_value_pair[0].lstrip('#').strip()
    value = name_value_pair[1].strip()
    return name, value


def is_full_text(filepath):
    with open(filepath, 'r') as conllup_file:
        for line in conllup_file:
            line = line.strip()
            if is_segment(line):
                continue
            elif is_metadata(line):
                name, value = get_metadata(line)
                if name == 'content_type':
                    if value.strip() == 'full_text':
                        return True
                    else:
                        return False
    return False


def get_first_sentence(filepath):
    with open(filepath, 'r') as conllup_file:
        for line in conllup_file:
            line = line.strip()
            if is_segment(line):
                continue
            elif is_metadata(line):
                name, value = get_metadata(line)
                if name == 'text':
                    return value.strip()
    return ''


def get_domain(filepath):
    doc_id = os.path.basename(filepath).split('.')[0]

    if doc_id in ['pl-bn-617666',
                  'pl-bn-609242',
                  'pl-bn-609273',
                  'pl-bn-617396',
                  'pl-bn-617282',
                  'pl-bn-617402',
                  'pl-bn-466202',
                  'pl-bn-617598']:
        return 'Health'

    with open(filepath, 'r') as conllup_file:
        for line in conllup_file:
            line = line.strip()
            if is_segment(line):
                continue
            elif is_metadata(line):
                name, value = get_metadata(line)
                if name == 'disciplines':
                    return map_discipline(value.strip())
    return 'Other'


def map_discipline(disciplines_val):
    disciplines = disciplines_val.split(' | ')
    if ('biomedical engineering' in disciplines or 'health sciences' in disciplines or
            'medical sciences' in disciplines or 'pharmaceutical sciences' in disciplines or
            'physical culture sciences' in disciplines):
        return 'Health'
    elif 'education' in disciplines:
        return 'Education'
    elif ('agriculture and gardening' in disciplines or
          'forestry sciences' in disciplines or 'veterinary medicine' in disciplines or
          'biological sciences' in disciplines or 'zootechnics and fishing' in disciplines):
        return 'Nature'
    elif 'politics and administration' in disciplines:
        return 'Politics'
    elif 'economics and finance' in disciplines or 'socio-economic geography and spatial economy' in disciplines:
        return 'Economy'
    elif ('art sciences' in disciplines or 'fine arts and conservation of works of art' in disciplines or
          'science about culture and religion' in disciplines or 'literature' in disciplines or
          'philosophy' in disciplines or 'archeology' in disciplines or 'history' in disciplines or
          'linguistics' in disciplines):
        return 'Culture'
    elif ('legal sciences' in disciplines or 'management and quality' in disciplines or
          'security sciences' in disciplines or 'social communication and media' in disciplines or
          'sociological sciences' in disciplines or 'psychology' in disciplines or 'the canonic law' in disciplines):
        return 'Social issues'
    elif ('civil engineering and transport' in disciplines or
          'environmental engineering, mining and energy' in disciplines or
          'automation, electronics and electrical engineering' in disciplines or
          'mechanical engineering' in disciplines or
          'material engineering' in disciplines or
          'architecture and urban planning' in disciplines or
          'informatics' in disciplines or
          'chemical engineering' in disciplines or
          'chemical sciences' in disciplines or
          'mathematics' in disciplines or
          'physical sciences' in disciplines or
          'technical IT and telecommunications' in disciplines or
          'food and nutrition technology' in disciplines or
          'Earth and the environment sciences' in disciplines):
        return 'Science'
    return disciplines_val


def clean_and_save_full(src, dst, year, lang_detector):
    cleaned_lines = []
    paragraph = []
    sentence = []

    publishing_company = get_publishing_company(src)
    journal = get_journal(src)
    domain = get_domain(src)
    title = get_title(src)

    first_sentence_text = get_first_sentence(src)
    full_text = is_full_text(src)

    article_id = os.path.basename(src).split('.')[0]

    if domain == 'Other':
        pass
    elif first_sentence_text in ['Artykuł nie zawiera streszczenia', 'nie dotyczy', 'Metryczka wydawnicza numeru.',
                               'Materiały bibliograficzne', 'Brak abstraktu w języku polskim'] and \
            not full_text:
        pass
    elif publishing_company == 'Uniwersytet Jana Długosza w Częstochowie. Wydawnictwo Uniwersytetu Jana Długosza w Częstochowie' and \
            journal == 'Edukacja Muzyczna' and year in ['2010', '2011', '2012', '2013', '2014'] and full_text:
        pass
    elif publishing_company == 'Uniwersytet Jana Długosza w Częstochowie. Wydawnictwo Uniwersytetu Jana Długosza w Częstochowie' and \
            journal == 'Prace Naukowe Akademii im. Jana Długosza w Częstochowie. Technika, Informatyka, Inżynieria Bezpieczeństwa' and \
            year in ['2013', '2014', '2015', '2016'] and full_text:
        pass
    elif publishing_company == 'Stowarzyszenie Geodetów Polskich' and \
            journal == 'Archiwum Fotogrametrii, Kartografii i Teledetekcji' and \
            year in ['2000', '2002', '2004'] and full_text:
        pass
    elif publishing_company == 'Uniwersytet Śląski' and \
            journal == 'Studia Politicae Universitatis Silesiensis' and \
            year in ['2006'] and full_text:
        pass
    elif publishing_company == 'Polskie Towarzystwo Rusycystyczne' and \
            journal == 'Przegląd Rusycystyczny' and \
            year in ['2010'] and full_text:
        pass
    elif publishing_company == 'Uniwersytet im. Adama Mickiewicza w Poznaniu' and \
            journal == 'Rocznik Integracji Europejskiej' and \
            year in ['2010'] and full_text:
        pass
    elif not get_authors(src):
        print('Missing authors:', article_id)
        pass
    elif not journal:
        print('Missing journal:', article_id)
        pass
    elif not title:
        print('Missing title:', article_id)
        pass
    else:
        with open(src, 'r') as conllup_file:
            first_sentence = True
            for line in conllup_file:
                line = line.strip()
                if is_segment(line):
                    sentence.append(line)
                elif is_metadata(line):
                    name, value = get_metadata(line)

                    if value == '[UNKNOWN]':
                        pass
                    elif name == 'newpar id':
                        if sentence and sentence_is_fine(sentence, lang_detector, first_sentence, full_text, title):
                            sentence.append('')
                            paragraph.extend(sentence)
                            first_sentence = False
                        sentence = []
                        if len(paragraph) > 1:
                            cleaned_lines.extend(paragraph)
                        paragraph = [line]
                    elif name == 'sent_id':
                        if sentence and sentence_is_fine(sentence, lang_detector, first_sentence, full_text, title):
                            sentence.append('')
                            paragraph.extend(sentence)
                            first_sentence = False
                        sentence = [line]
                    elif name == 'text':
                        sentence.append(line)
                    elif name == 'type':
                        cleaned_lines.append(line)
                        cleaned_lines.append(f'# domain = {domain}')
                    else:
                        cleaned_lines.append(line)

            if sentence and sentence_is_fine(sentence, lang_detector, first_sentence, full_text, title):
                sentence.append('')
                paragraph.extend(sentence)
            if len(paragraph) > 1:
                cleaned_lines.extend(paragraph)

            if not cleaned_lines[-1]:
                cleaned_lines.append('')

        contains_polish_signs = False
        for line in cleaned_lines:
            if is_metadata(line):
                name, value = get_metadata(line)
                if name == 'text' and ('ą' in value or 'ł' in value):
                    contains_polish_signs = True
                    break

        if not cleaned_lines[-1].strip():
            cleaned_lines.pop()

        if len(cleaned_lines) >= 100 and contains_polish_signs:
            with open(dst, 'w') as dst_file:
                dst_file.write('\n'.join(cleaned_lines))


def get_publishing_company(filepath):
    with open(filepath, 'r') as conllup_file:
        for line in conllup_file:
            line = line.strip()
            if is_segment(line):
                continue
            elif is_metadata(line):
                name, value = get_metadata(line)
                if name == 'publishing_company':
                    return value.strip()
    return ''


def get_authors(filepath):
    with open(filepath, 'r') as conllup_file:
        for line in conllup_file:
            line = line.strip()
            if is_segment(line):
                continue
            elif is_metadata(line):
                name, value = get_metadata(line)
                if name == 'authors':
                    return value.strip()
    return ''


def get_journal(filepath):
    with open(filepath, 'r') as conllup_file:
        for line in conllup_file:
            line = line.strip()
            if is_segment(line):
                continue
            elif is_metadata(line):
                name, value = get_metadata(line)
                if name == 'journal':
                    return value.strip()
    return ''


def get_title(filepath):
    with open(filepath, 'r') as conllup_file:
        for line in conllup_file:
            line = line.strip()
            if is_segment(line):
                continue
            elif is_metadata(line):
                name, value = get_metadata(line)
                if name == 'title':
                    return value.strip()
    return ''


def get_type(filepath):
    with open(filepath, 'r') as conllup_file:
        for line in conllup_file:
            line = line.strip()
            if is_segment(line):
                continue
            elif is_metadata(line):
                name, value = get_metadata(line)
                if name == 'type':
                    return value.strip()
    return ''


def get_id(filepath):
    with open(filepath, 'r') as conllup_file:
        for line in conllup_file:
            line = line.strip()
            if is_segment(line):
                continue
            elif is_metadata(line):
                name, value = get_metadata(line)
                if name == 'newdoc id':
                    return value.strip()
    return ''


def is_bib_start(text):
    if re.search('^\d*\s*(Bibliografia|Literatura|LITERATURA)', text):
        return True
    return False


def sentence_is_fine(sentence_lines, lang_detector, first_sentence, full_text, title):
    first_segment = sentence_lines[2]
    _, text = get_metadata(sentence_lines[1])

    if not full_text and first_sentence and title.startswith(text):
        return True

    if not all_parentheses_closed(text):
        return False
    if text[0].islower():
        return False
    if text[0].isdigit():
        return False
    if text[-1] not in ['!', '?', '.', '…']:
        return False
    if first_segment.split()[3] == 'PUNCT' or first_segment.split()[4] == 'interp':
        return False
    if not correct_root(sentence_lines):
        return False
    if wrong_start(text):
        return False
    if unwanted_word(text):
        return False
    if not contains_verb(sentence_lines):
        return False
    if to_much_short_words(text):
        return False
    if to_much_upper(text):
        return False
    if lang_detector.FindLanguage(text).language != 'pl':
        return False
    for sign in ['∏', 'ê', 'Ê', '´', 'ç', 'ƒ', '˝', 'à', '¹', 'æ', '¿', '³', 'ñ', 'Ŝ', 'œ', 'œ', '�', 'Œ']:
        if sign in text:
            return False
    return True


def all_parentheses_closed(sequence):
    stack = []
    opening = set('([{')
    closing = set(')]}')
    pair = {')': '(', ']': '[', '}': '{'}
    for i in sequence:
        if i in opening:
            stack.append(i)
        if i in closing:
            if not stack:
                return False
            elif stack.pop() != pair[i] :
                return False
            else:
                continue
    if not stack :
        return True
    else:
        return False


def correct_root(sentence_lines):
    root_count = 0
    for line in sentence_lines:
        if line.strip() and line[0] != '#':
            cols = line.split('\t')
            if 'root' == cols[7]:
                root_count += 1
                if cols[6] != '0':
                    return False
    if root_count == 1:
        return True
    return False


def to_much_upper(text):
    upper_count = 0
    for sign in text:
        if sign.isupper():
            upper_count += 1
        if float(upper_count)/float(len(text)) >= 0.2:
            return True
    return False


def to_much_short_words(text):
    short_count = 0
    words = text.split()
    for word in words:
        if len(word) == 1:
            short_count += 1
        if float(short_count)/float(len(words)) >= 0.5:
            return True
    return False


def unwanted_word(text):
    words = text.split()
    for word in words:
        if len(word) > 1 and (word.startswith('-') or word.endswith('-') or word.startswith('‑') or word.endswith('‑')):
            return True
        elif word in ['(red.)', 'Wydawnictwo', 'wyd.', 'grantu', 'issue', 'volume', 'Journal', 'journal',
                      'STRESZCZENIE:', 'KLUCZOWE:', '(bud.', 'Druk.']:
            return True
        elif re.search('^\d\d\d\d:', word):
            return True
    return False


def wrong_start(text):
    if re.search('^([Rr]y[sc]\.?\s+\d|Rysunek\s+\d\.|Tabela\s+\d|Tab\.?\s+\d|Recenzował:|Słowa\s+kluczowe:|Źródło:|'
                 'STRESZCZENIE:|Vol\.\s+\d|Streszczenie:|Dane\s+autorów:|WSTĘP|Streszczenie\s+[A-ZĄĆĘŁŃÓŚŹŻ]|'
                 'Key\s+words:|Keywords:|Explanation:|Pobrano\s+z:|E-mail:|[XIV]+\.|Zeszyt\s+|Roczniki\s+|Literatura:|'
                 'Wstęp\s+|Lp\.\s+|A b s tr a k t\s+|Adres\s+do\s+korespondencji:|Zesz\.\s+|Przegl\.|Zagłęb\.|'
                 'WęWęW\s+giel|Retyk\s+i\s+lias|Górny\s+dogger|Pobrane\s+z\s+czasopisma\s+|ORCID:|'
                 'Autor\s+korespondujący:|Materiały\s+konf\.)',
                 text):
        return True
    return False


def contains_verb(sentence_lines):
    for line in sentence_lines:
        if line.strip() and line[0] != '#':
            cols = line.split('\t')
            if cols[3] in ['VERB', 'AUX']:
                return True
    return False


def text_hard_error(sentence_lines):
    _, text = get_metadata(sentence_lines[1])
    for sign in ['∏', 'ê', 'Ê', '´', 'ç', 'ƒ', '˝', 'à', '¹', 'æ', '¿', '³', 'ñ', 'Ŝ', 'œ', 'œ']:
        if sign in text:
            return True
    return False


if __name__ == '__main__':
    main()