src/control/control.py

import asyncio  # on va en avoir besoin :)
import string
import random
from datetime import datetime


from src.tools.semantic_db import get_or_create_collection, reset_collection
from src.tools.wiki import Wiki
from src.model.document import InputDoc, WikiPage
from src.tools.llm_tools import get_wikilist, get_public_paragraph, get_private_paragraph
from src.tools.semantic_db import add_texts_to_collection, query_collection

"""
Tools
"""


def get_long_id(id_):
    if id_ != -1:
        return id_
    else:
        now = datetime.now().strftime("%m%d%H%M")
        letters = string.ascii_lowercase + string.digits
        long_id = now+'-'+''.join(random.choice(letters) for _ in range(10))
    return long_id


"""
Input control
"""



"""
Source Control
"""

def wiki_fetch(input_text: str) -> [str]:
    """
    returns the title of the wikipages corresponding to the tasks described in the input text
    """
    tasks = InputDoc(input_text).tasks
    wiki_lists = [get_wikilist(t) for t in tasks]
    flatten_wiki_list = list(set().union(*[set(w) for w in wiki_lists]))
    return flatten_wiki_list


async def wiki_upload_and_store(wiki_title: str, collection_name: str):
    """
    uploads one wikipage and stores them into the right collection
    """
    wikipage = Wiki().fetch(wiki_title)
    wiki_title = wiki_title
    if type(wikipage) != str:
        texts = WikiPage(wikipage.page_content).get_paragraphs()
        add_texts_to_collection(coll_name=collection_name, texts=texts, file=wiki_title, source='wiki')
    else:
        print(wikipage)


async def my_files_upload_and_store(title: str, collection_name: str):
    doc = title
    title = title
    texts = InputDoc(doc).get_paragraphs()
    add_texts_to_collection(coll_name=collection_name, texts=texts, file=title, source='my_files')


"""
Generate Control
"""


def generate_doc_from_gpt(input_txt: str) -> str:
    input_doc = InputDoc(input_txt)
    tasks = input_doc.tasks
    task_resolutions = [get_public_paragraph(t) for t in tasks]
    # task_resolutions = ["ça c'est de la réso"]
    generated_doc = input_doc.replace_tasks(task_resolutions)
    return generated_doc


def generate_doc_from_db(input_txt: str, collection_name: str, from_files: [str]) -> str:

    def query_from_task(task):
        return get_public_paragraph(task)
    input_doc = InputDoc(input_txt)
    tasks = input_doc.tasks
    queries = [query_from_task(t) for t in tasks]
    texts_list = [query_collection(coll_name=collection_name, query=q, from_files=from_files) for q in queries]
    task_resolutions = [get_private_paragraph(task=task, texts=texts) for task, texts in zip(tasks, texts_list)]
    generated_doc = input_doc.replace_tasks(task_resolutions)
    return generated_doc