Hackathon - Document Search Solution¶

metadata:
  entity: Company Sample A
  language: VIE
  number: "No. 120"
  issued_date: 2021-11-04
  effective_date: 2022-01-01
  source: Business Development
  subject: Strategic Opportunities Template
  title: Strategic Opportunities Template
  type: Internal Regulation
  is_internal: true

keyword:
  string: Ha Noi
  related: Ho Chi Minh|Hai Phong|Ninh Binh|Vinh Phuc

most_search:
  1: Salary
  2: License
  3: Tax

.
data/ # Contain dataset
src/ # Contain `src` component
|____ document/
|____ file/
|____ util/
__init__.py
.gitignore
config.py # Configuration file
extract.py # Extract Document Information
metadata.py # Parse metadata
README.md # Project indtroduction
Makefile # Automation with target
requirements.txt # Dependencies

seaborn==0.11.2
thefuzz==0.19.0
pandas>=1.1.0
nltk>=3.5
pyarrow>=7.0.0
matplotlib==3.5.2

# Global
import sys
import os

# Path Append
sys.path.append(os.path.abspath(os.curdir))

# Declare

# Temporary folder, which contains a log file
TEMP_DIR = os.path.join('temp')

# Source of files, which has been stored raw file from uploaded document.
SOURCE_DIR = os.path.join('data', 'source')

# Source of processed file and data extraction.
OUTPUT_DIR = os.path.join(TEMP_DIR, 'document')

# File support
SUPPORTED_FILE_TYPE: list[str] = ["*.doc", "*.docx", "*.doc", "*.txt"]

# Truncated legth threshold
TRUNCATED_LENGTH: int = 500

# Metadata selekon
METADATA_SKLETON: dict[str, dict] = {
    'metadata': {
        'language': None,
        'title': None,
        'subject': None,
        'entity': None,
        'source': None,
        'number': None,
        'type': None,
        'issued_date': None,
        'effective_date': None,
        'is_internal': False
    }
}

python document/metadata.py --source data/source/

python document/extract.py --source data/source/ --destination temp/document

#!/bin/bash

# Default Shell
SHELL := /usr/bin/bash

# Folder
SOURCE_DIR = data/source
DESTINATION_DIR = temp/document

metadata:
    python metadata.py --source ${SOURCE_DIR}

extract:
    python extract.py --source ${SOURCE_DIR} --destination ${DESTINATION_DIR}

make document-metadata
make document-extract

import os
import string
import re
import hashlib
import pandas as pd
from thefuzz import process
import yaml


class Document:
    """Document is a class represented for any files in various type (txt, path, png)
    then bake it into a mechanic to enrich document metadatas.

    Return:
        [Document]: Document Class

    Example:
        samp = Document(path='data/source/sample.txt')
        samp.write_parquet('temp/t.parquet')
        samp.write_excel('temp/t.xlsx')
    """

    def __init__(self, path: str = None):

        # Mechanic
        self.path = path
        self.original_path, self.file = os.path.split(self.path)
        self.file_name, self.file_ext = os.path.splitext(self.file)
        self.file_metadata = os.path.join(self.original_path, self.file_name + ".yml")

        metadata = {}
        if os.path.isfile(self.file_metadata):
            try:
                with open(self.file_metadata) as f:
                    component: dict = yaml.load(f, Loader=yaml.FullLoader)
                    metadata = component.get('metadata')
            except Exception:
                self.file_metadata = None
                raise Exception
        else:
            import subprocess
            try:
                subprocess.run(
                    args=['python', 'metadata.py', '--file', self.file_metadata],
                    capture_output=True,
                    text=True,
                    check = True,
                    timeout=60
                )
            except Exception:
                self.file_metadata = None
                raise Exception

        # Attribute
        self.language = safe_dict_extract(d = metadata, key = 'language')
        self.title = safe_dict_extract(d = metadata, key = 'title')
        self.subject =  safe_dict_extract(d = metadata, key = 'subject')
        self.entity = safe_dict_extract(d = metadata, key = 'entity')
        self.source = safe_dict_extract(d = metadata, key = 'source')
        self.number = safe_dict_extract(d = metadata, key = 'number')
        self.type = safe_dict_extract(d = metadata, key = 'type')
        self.issued_date = safe_dict_extract(d = metadata, key = 'issued-date')
        self.effective_date = safe_dict_extract(d = metadata, key = 'effective-date')
        self.is_internal = safe_dict_extract(d = metadata, key = 'is-internal')

        # Component
        self.id = f"D{hashlib.md5(self.file.encode()).hexdigest()}"
        self.document = self._read_document()
        self.content = self._extract_content()
        self.paragraphs, self.sentences, self.words =  self._parse_dictionary()
        self.tokens = self._get_unique_words()
        self.document_dataframe = self._content_to_df()

    def _read_document(self):

        # Validate File Exits
        if not os.path.isfile(self.path):
            raise FileNotFoundError('There are folder, not exists any file. Check your path')

        # Processing
        if self.file_ext == ".txt":
            with open(self.path, encoding='UTF-8') as f:
                cont = f.read()
        elif self.file_ext in (".doc", ".docx"):
            import textract
            parse_content = textract.process(self.path, input_encoding='UTF-8', output_encoding='UTF-8', extension=self.file_ext).decode('UTF-8')
            cont = ''.join(parse_content)
        else:
            raise Exception(f"Currently not supported this extentsion {self.file_ext}")

        # Decoding
        content = cont.encode('UTF-8', "ignore").decode('UTF-8')

        # Replace Whitespace
        # \u200b is a "zero-width-space" in Unicode.
        # Ref: https://stackoverflow.com/questions/35657620/illegal-character-error-u200b
        content = content.replace('\u200b', ' ').replace('  ', ' ')

        # Special Chraracter
        # Ref: https://stackoverflow.com/questions/5843518/remove-all-special-characters-punctuation-and-spaces-from-string
        content = re.sub(r'\W+', ' ', content).strip()

        return content

    def _extract_content(self):

        # Special Character
        # b'\xe2\x80\x93' -> TODO: Need to revert to bytes then excluded Special characters

        # Define
        contents = {}
        _paragraph = list(filter(lambda x: x not in ('', string.punctuation), self.document.split('\n\n')))

        # Paragraphs
        for ind_par, par in zip(range(1, len(_paragraph) + 1), _paragraph):

            sentences = {}
            _sentences = list(filter(lambda x: x not in ('', string.punctuation), par.split(r'\.')))

            # Sentences
            for ind_sen, sen in zip(range(1, len(_sentences) + 1), _sentences):

                words = {}
                _words = list(filter(lambda x: x not in ('', string.punctuation), sen.split()))

                # Words
                for ind_wor, wor in zip(range(1, len(_words) + 1), _words):
                    words.update({ind_wor: {
                        'position': ind_wor,
                        'word': wor
                    }})

                sentences.update({ind_sen: {
                    'position': ind_sen,
                    'sentence': sen,
                    'words': words
                }})

            contents.update({ind_par: {
                'position': ind_par,
                'paragraph': par,
                'sentences': sentences
            }})

        return contents

    def _parse_dictionary(self):

        all_paragraphs = []
        all_sentences = []
        all_words = []

        for i in range(len(self.content)):
            all_paragraphs.append(self.content.get(i+1).get('paragraph'))
            sents = self.content.get(i+1).get('sentences')
            for j in range(len(sents)):
                all_sentences.append(sents.get(j+1).get('sentence'))
                words = sents.get(j+1).get('words')
                for k in range(len(words)):
                    w = words.get(k+1).get('word')
                    all_words.append(w)

        paragraphs_dict = {}
        for ind, p in zip(list(range(1, len(all_paragraphs) + 1)), all_paragraphs):
            paragraphs_dict.update({ind: {
                'position': ind,
                'paragraph': p
                }
            })

        sentences_dict = {}
        for ind, s in zip(list(range(1, len(all_sentences) + 1)), all_sentences):
            sentences_dict.update({ind: {
                'position': ind,
                'sentence': s
                }
            })

        word_dict = {}
        for ind, w in zip(list(range(1, len(all_words) + 1)), all_words):
            word_dict.update({ind: {
                'position': ind,
                'word': w
                }
            })

        return paragraphs_dict, sentences_dict, word_dict

    def _get_unique_words(self):
        all_words = []
        for i in range(len(self.words)):
            all_words.append(self.words.get(i+1).get('word'))
        return list(set(all_words))

    def _content_to_df(self):
        _tabu_tuples = []
        for cont, ty in zip(
            [self.paragraphs, self.sentences, self.words],
            ['paragraph', 'sentence', 'word']
        ):
            for _posi in range(1, len(cont)):
                _info = cont[_posi]
                _value = _info.get(ty)
                _tabu_tuples.append((self.id, ty, _posi, _value, hashlib.md5(_value.encode()).hexdigest()))
        df = pd.DataFrame(_tabu_tuples, columns=['document_id', 'type', 'position', 'value', 'hashed_content_md5'])
        return df

    def write_excel(self, target):

        # Valid File
        if self.check_valid_ext(target, "xlsx") is False:
            raise ValueError(f"Target file not good {target}")

        with pd.ExcelWriter(target, mode='w') as writer:
            self.document_dataframe.to_excel(writer, sheet_name='DATA', engine='xlsxwriter')

    def write_parquet(self, target):

        # Valid File
        if self.check_valid_ext(target, "parquet") is False:
            raise ValueError(f"Target file not good {target}")

        self.document_dataframe.to_parquet(path=target)

    def get_id(self):
        return self.id

    def get_document(self):
        return self.document

    def get_content(self):
        return self.content

    def get_paragraphs(self):
        return self.paragraphs

    def get_sentences(self):
        return self.sentences

    def get_words(self):
        return self.words

    def get_tokens(self):
        return self.tokens

    def get_metadata(self):
        return {
            'id': self.id,
            'language': self.language,
            'title': self.title,
            'subject': self.subject,
            'entity': self.entity,
            'source': self.source,
            'type': self.type,
            'issued_date': self.issued_date,
            'effective_date': self.effective_date,
            'is_internal': self.is_internal
        }

    def to_dataframe(self):
        return self.document_dataframe

    def statistics(self):
        return {
            'paragraph': len(self.content),
            'sentence': len(self.sentences),
            'word': len(self.words)
        }

    def update_metadata(self, path: str = None):
        fi = os.path.basename(path)
        fin, fie = os.path.split(fi)
        if os.path.isfile(path) and fie == 'yml':
            if path != self.file_metadata:
                # Metadata
                try:
                    with open(self.file_metadata) as f:
                        metadata = yaml.load(f, Loader=yaml.FullLoader).get('metadata')
                except Exception:
                    self.file_metadata = None
                    raise Exception

        # Update
        self.language = safe_dict_extract(d = metadata, key = 'language')
        self.title = safe_dict_extract(d = metadata, key = 'title')
        self.subject =  safe_dict_extract(d = metadata, key = 'subject')
        self.entity = safe_dict_extract(d = metadata, key = 'entity')
        self.source = safe_dict_extract(d = metadata, key = 'source')
        self.number = safe_dict_extract(d = metadata, key = 'number')
        self.type = safe_dict_extract(d = metadata, key = 'type')
        self.issued_date = safe_dict_extract(d = metadata, key = 'issued-date')
        self.effective_date = safe_dict_extract(d = metadata, key = 'effective-date')
        self.is_internal = safe_dict_extract(d = metadata, key = 'is-internal')

    def search(self, keyword: str, limit: int = 10, threshold: int = 50):
        _satified = []
        for wo, ty in zip(
            [self.paragraphs, self.sentences, self.words],
            ['paragraph', 'sentence', 'word']
        ):
            seach_pool = wo.values()
            _pool = process.extract(keyword, seach_pool, limit=limit)
            for w, thr in _pool:
                if thr >= threshold:
                    _satified.append((ty, w, thr))
        return _satified

    @staticmethod
    def check_valid_ext(path, ext):
        ffile = os.path.basename(path)
        _, fext = os.path.splitext(ffile)
        result = False
        if fext == "." + ext:
            result = True
        return result

    def _check_file(path):
        result = False
        if os.path.isfile(path):
            result = True
        return result


def safe_dict_extract(d: dict = {}, key: str = None):
    return d.get(key) if d.get(key) is not None else None