|
|
| import asyncio |
| 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] |
| |
| 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 |
|
|
