topicModelingTickets/test.py

# -*- coding: utf-8 -*-
import time
start = time.time()


import csv
import functools
import os.path
import re
import subprocess
import time
import xml.etree.ElementTree as ET
import sys
import spacy
import textacy
from scipy import *
from textacy import Vectorizer
import warnings
csv.field_size_limit(sys.maxsize)



# Load the configuration file
import configparser as ConfigParser
config = ConfigParser.ConfigParser()
with open("config.ini") as f:
    config.read_file(f)



path2xml = config.get("default","path2xml")
thesauruspath = config.get("default","thesauruspath")


DE_PARSER = spacy.load("de")

de_stop_words=list(__import__("spacy." + DE_PARSER.lang, globals(), locals(), ['object']).STOP_WORDS)


corpus = textacy.Corpus(DE_PARSER)

THESAURUS = list(textacy.fileio.read_csv(thesauruspath, delimiter=";"))




############# misc
def compose(*functions):
    def compose2(f, g):
        return lambda x: f(g(x))
    return functools.reduce(compose2, functions, lambda x: x)

def get_calling_function():
    """finds the calling function in many decent cases.
    https://stackoverflow.com/questions/39078467/python-how-to-get-the-calling-function-not-just-its-name
    """
    fr = sys._getframe(1)   # inspect.stack()[1][0]
    co = fr.f_code
    for get in (
        lambda:fr.f_globals[co.co_name],
        lambda:getattr(fr.f_locals['self'], co.co_name),
        lambda:getattr(fr.f_locals['cls'], co.co_name),
        lambda:fr.f_back.f_locals[co.co_name], # nested
        lambda:fr.f_back.f_locals['func'],  # decorators
        lambda:fr.f_back.f_locals['meth'],
        lambda:fr.f_back.f_locals['f'],
        ):
        try:
            func = get()
        except (KeyError, AttributeError):
            pass
        else:
            if func.__code__ == co:
                return func
    raise AttributeError("func not found")

def printRandomDoc(textacyCorpus):
    import random
    print()

    print("len(textacyCorpus) = %i" % len(textacyCorpus))
    randIndex = int((len(textacyCorpus) - 1) * random.random())
    print("Index: {0} ; Text: {1} ; Metadata: {2}".format(randIndex, textacyCorpus[randIndex].text, textacyCorpus[randIndex].metadata))

    print()





#############  on xml
def generateMainTextfromTicketXML(path2xml, main_textfield='Beschreibung'):
    """
    generates strings from XML
    :param path2xml:
    :param main_textfield:
    :param cleaning_function:
    :yields strings
    """
    tree = ET.parse(path2xml, ET.XMLParser(encoding="utf-8"))
    root = tree.getroot()

    for ticket in root:
        for field in ticket:
            if field.tag == main_textfield:
                    yield field.text
def generateMetadatafromTicketXML(path2xml, leave_out=['Beschreibung']):
    tree = ET.parse(path2xml, ET.XMLParser(encoding="utf-8"))
    root = tree.getroot()

    for ticket in root:
        metadata = {}
        for field in ticket:
            if field.tag not in leave_out:

                    metadata[field.tag] = field.text

        yield metadata



#############  on csv

def csv_to_contentStream(path2csv: str, content_collumn_name: str):
    """
    :param path2csv: string
    :param content_collumn_name: string
    :return: string-generator
    """
    stream = textacy.fileio.read_csv(path2csv, delimiter=";")  # ,encoding='utf8')
    content_collumn = 0  # standardvalue

    for i,lst in enumerate(stream):
        if i == 0:
            # look for desired column
            for j,col in enumerate(lst):
                if col == content_collumn_name:
                    content_collumn = j
        else:
            yield lst[content_collumn]
def csv_to_metaStream(path2csv: str, metalist: [str]):
    """
    :param path2csv: string
    :param metalist: list of strings
    :return: dict-generator
    """
    stream = textacy.fileio.read_csv(path2csv, delimiter=";")  # ,encoding='utf8')

    content_collumn = 0  # standardvalue
    metaindices = []
    metadata_temp = {}
    for i,lst in enumerate(stream):
        if i == 0:
            for j,col in enumerate(lst):        # geht bestimmt effizienter... egal, weil passiert nur einmal
                for key in metalist:
                    if key == col:
                        metaindices.append(j)
            metadata_temp = dict(zip(metalist,metaindices)) # zB {'Subject' : 1, 'categoryName' : 3, 'Solution' : 10}

        else:
            metadata = metadata_temp.copy()
            for key,value in metadata.items():
                metadata[key] = lst[value]
            yield metadata


#############  on str-gen




def processTokens(tokens, funclist, parser):
    # in:tokenlist, funclist
    # out: tokenlist
    for f in funclist:
        if 'bool' in str(f.__annotations__):
            tokens = list(filter(f, tokens))

        elif 'str' in str(f.__annotations__):
            tokens = list(map(f, tokens))  # purer text
            doc = parser(" ".join(tokens))  # geparsed
            tokens = [tok for tok in doc]  # nur tokens

        elif 'spacy.tokens.doc.Doc' in str(f.__annotations__):
            toks = f(tokens)
            tokens = [tok for tok in toks]

        else:
            warnings.warn("Unknown Annotation while preprocessing. Function: {0}".format(str(f)))

    return tokens




#############  return docs

def keepUniqueTokens() -> spacy.tokens.Doc:
    #todo in:tok out:doc
    ret = lambda doc: (set([tok.lower_ for tok in doc]))

    ret.__annotations__ = get_calling_function().__annotations__
    return ret











def processTextstream(textstream, funclist, parser=DE_PARSER):
    """
    :param textstream: string-gen
    :param funclist: [func]
    :param parser: spacy-parser
    :return: string-gen
    """
    # input:str-stream output:str-stream
    pipe = parser.pipe(textstream)

    for doc in pipe:
        tokens = [tok for tok in doc]
        tokens = processTokens(tokens,funclist,parser)
        yield " ".join([tok.lower_ for tok in tokens])

def processDictstream(dictstream, funcdict, parser=DE_PARSER):
    """

    :param dictstream: dict-gen
    :param funcdict:
                    clean_in_meta = {
                        "Solution":funclist,
                        ...
                    }

    :param parser: spacy-parser
    :return: dict-gen
    """
    for dic in dictstream:
        result = {}
        for key, value in dic.items():

            if key in funcdict:

                doc = parser(value)
                tokens = [tok for tok in doc]
                funclist = funcdict[key]

                tokens = processTokens(tokens,funclist,parser)


                result[key] = " ".join([tok.lower_ for tok in tokens])


            else:
                result[key] = value
        yield result



#############  return tokens

def keepPOS(pos_list) -> bool:
    ret = lambda tok : tok.pos_ in pos_list

    ret.__annotations__ = get_calling_function().__annotations__
    return ret

def removePOS(pos_list)-> bool:
    ret = lambda tok : tok.pos_ not in pos_list

    ret.__annotations__ = get_calling_function().__annotations__
    return ret

def removeWords(words, keep=None)-> bool:
    #todo in:str oder str-list
    if hasattr(keep, '__iter__'):
        for k in keep:
            try:
                words.remove(k)
            except ValueError:
                pass

    ret = lambda tok :  tok.lower_ not in words

    ret.__annotations__ = get_calling_function().__annotations__
    return ret

def keepENT(ent_list) -> bool:
    ret = lambda tok : tok.ent_type_ in ent_list

    ret.__annotations__ = get_calling_function().__annotations__
    return ret

def removeENT(ent_list) -> bool:
    ret = lambda tok: tok.ent_type_ not in ent_list

    ret.__annotations__ = get_calling_function().__annotations__
    return ret

def lemmatize() -> str:
    ret = lambda tok:  tok.lemma_

    ret.__annotations__ = get_calling_function().__annotations__
    return ret


def normalizeSynonyms(default_return_first_Syn=False) -> str:
    ret = lambda tok : getFirstSynonym(tok.lower_, default_return_first_Syn=default_return_first_Syn)

    ret.__annotations__ = get_calling_function().__annotations__
    return ret

def getFirstSynonym(word, thesaurus=THESAURUS, default_return_first_Syn=False):
    if not isinstance(word, str):
        return str(word)

    word = word.lower()

    # durch den thesaurrus iterieren
    for syn_block in thesaurus:  # syn_block ist eine liste mit Synonymen

        for syn in syn_block:
            syn = syn.lower()
            if re.match(r'\A[\w-]+\Z', syn):  # falls syn einzelwort ist
                if word == syn:
                    return str(getHauptform(syn_block, word, default_return_first_Syn=default_return_first_Syn))
            else:  # falls es ein satz ist
                if word in syn:
                    return str(getHauptform(syn_block, word, default_return_first_Syn=default_return_first_Syn))
    return str(word)  # zur Not, das ursrpüngliche Wort zurückgeben

def getHauptform(syn_block, word, default_return_first_Syn=False):
    for syn in syn_block:
        syn = syn.lower()

        if "hauptform" in syn and len(syn.split(" ")) <= 2:
            # nicht ausgeben, falls es in Klammern steht#todo gibts macnmal?? klammern aus
            for w in syn.split(" "):
                if not re.match(r'\([^)]+\)', w):
                    return w

    if default_return_first_Syn:
        # falls keine hauptform enthalten ist, das erste Synonym zurückgeben, was kein satz ist und nicht in klammern steht
        for w in syn_block:
            if not re.match(r'\([^)]+\)', w):
                return w
    return word  # zur Not, das ursrpüngliche Wort zurückgeben



#############  return strings

mentionFinder = re.compile(r"@[a-z0-9_]{1,15}", re.IGNORECASE)
emailFinder = re.compile(r"\b[A-Z0-9._%+-]+@[A-Z0-9.-]+\.[A-Z]{2,}\b", re.IGNORECASE)
urlFinder = re.compile(r"^(?:https?:\/\/)?(?:www\.)?[a-zA-Z0-9./]+$", re.IGNORECASE)

def replaceEmails(replace_with="EMAIL") -> str:
    ret = lambda tok : emailFinder.sub(replace_with, tok.lower_)

    ret.__annotations__ = get_calling_function().__annotations__
    return ret

def replaceURLs(replace_with="URL") -> str:
    ret = lambda tok: textacy.preprocess.replace_urls(tok.lower_,replace_with=replace_with)
    #ret = lambda tok: urlFinder.sub(replace_with,tok.lower_)

    ret.__annotations__ = get_calling_function().__annotations__
    return ret

def replaceTwitterMentions(replace_with="TWITTER_MENTION") -> str:
    ret = lambda tok : mentionFinder.sub(replace_with,tok.lower_)

    ret.__annotations__ = get_calling_function().__annotations__
    return ret

def replaceNumbers(replace_with="NUMBER") -> str:
    ret = lambda tok: textacy.preprocess.replace_numbers(tok.lower_, replace_with=replace_with)

    ret.__annotations__ = get_calling_function().__annotations__
    return ret

def replacePhonenumbers(replace_with="PHONENUMBER") -> str:
    ret = lambda tok: textacy.preprocess.replace_phone_numbers(tok.lower_, replace_with=replace_with)

    ret.__annotations__ = get_calling_function().__annotations__
    return ret


def resolveAbbreviations():
    pass #todo


metaliste = [
    "Subject",
    "categoryName",
    "Solution"
]
path2csv = "M42-Export/Tickets_small.csv"



clean_in_meta = {
    "Solution":[removePOS(["SPACE"])],
    "Subject":[removePOS(["SPACE","PUNCT"])]
}


clean_in_content=[
    removePOS(["SPACE","PUNCT","NUM"]),
    keepPOS(["NOUN"]),
    replaceURLs(),
    replaceEmails(),
    removeWords(de_stop_words),
    lemmatize()

]



## add files to textacy-corpus,
print("add texts to textacy-corpus...")
corpus.add_texts(
    processTextstream(csv_to_contentStream(path2csv,"Description"), clean_in_content),
    processDictstream(csv_to_metaStream(path2csv,metaliste),clean_in_meta)
)

printRandomDoc(corpus)











end = time.time()
print("\n\n\nTime Elapsed:{0}".format(end - start))