| from typing import Dict |
| import random |
| import datetime |
| import string |
|
|
| from src.domain.doc import Doc |
| from src.domain.wikidoc import WikiPage |
| from src.view.log_msg import create_msg_from |
| import src.tools.semantic_db as semantic_db |
| from src.tools.wiki import Wiki |
| 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 |
|
|
|
|
| class Controller: |
|
|
| def __init__(self, config: Dict): |
| self.templates_path = config['templates_path'] |
| self.generated_docs_path = config['generated_docs_path'] |
| self.styled_docs_path = config['styled_docs_path'] |
| self.new_docs = [] |
| self.gen_docs = [] |
|
|
| template_path = config['templates_path'] + '/' + config['templates'][config['default_template_index']] |
| self.default_template = Doc(template_path) |
| self.template = self.default_template |
| self.log = [] |
| self.differences = [] |
|
|
| def copy_docs(self, temp_docs: []): |
| get_name = lambda doc: doc.name.split('/')[-1].split('.')[0] |
| doc_names = [get_name(doc) for doc in temp_docs] |
| docs = [Doc(path=doc.name) for doc in temp_docs] |
| style_paths = [f"{self.generated_docs_path}/{dn}_.docx" for dn in doc_names] |
| gen_paths = [f"{self.generated_docs_path}/{dn}_e.docx" for dn in doc_names] |
| for doc, style_path, gen_path in zip(docs, style_paths, gen_paths): |
| new_doc = doc.copy(style_path) |
| self.new_docs.append(new_doc) |
|
|
| def clear_docs(self): |
| for new_doc in self.new_docs: |
| new_doc.clear() |
| for gen_doc in self.gen_docs: |
| gen_doc.clear() |
| self.new_docs = [] |
| self.gen_docs = [] |
| self.log = [] |
|
|
| def set_template(self, template_name: str = ""): |
| if not template_name: |
| self.template = self.default_template |
| else: |
| template_path = f"{self.templates_path}/{template_name}" |
| self.template = Doc(template_path) |
|
|
| def get_difference_with_template(self): |
| self.differences = [] |
| for new_doc in self.new_docs: |
| diff_styles = new_doc.get_different_styles_with_template(template=self.template) |
| diff_dicts = [{'doc': new_doc, 'style': s} for s in diff_styles] |
| self.differences += diff_dicts |
| template_styles = [name for name in self.template.styles.names if name.startswith('.')] |
| return self.differences, template_styles |
|
|
| def map_style(self, this_style_index: int, template_style_name: str): |
| """ |
| maps a style from 'this' document into a style from the template |
| """ |
| diff_dict = self.differences[this_style_index] |
| doc = diff_dict['doc'] |
| this_style_name = diff_dict['style'] |
| log = doc.copy_one_style(this_style_name, template_style_name, self.template) |
| self.log.append({doc.name: log}) |
|
|
| def apply_template(self, add_front_pages: bool): |
| for new_doc in self.new_docs: |
| log = new_doc.apply_template(template=self.template, add_front_pages=add_front_pages) |
| if log: |
| self.log.append({new_doc.name: log}) |
|
|
| def reset(self): |
| for new_doc in self.new_docs: |
| new_doc.delete() |
| for gen_doc in self.gen_docs: |
| gen_doc.delete() |
| self.new_docs = [] |
| self.gen_docs = [] |
|
|
|
|
| def get_log(self): |
| msg_log = create_msg_from(self.log, self.new_docs) |
| return msg_log |
|
|
| """ |
| Source Control |
| """ |
|
|
| def get_or_create_collection(self, id_: str) -> str: |
| """ |
| generates a new id if needed |
| """ |
| if id_ != '-1': |
| return id_ |
| else: |
| now = datetime.datetime.now().strftime("%m%d%H%M") |
| letters = string.ascii_lowercase + string.digits |
| id_ = now + '-' + ''.join(random.choice(letters) for _ in range(10)) |
| semantic_db.get_or_create_collection(id_) |
| return id_ |
|
|
| def wiki_fetch(self) -> [str]: |
| """ |
| returns the title of the wikipages corresponding to the tasks described in the input text |
| """ |
| all_tasks = [] |
| for new_doc in self.new_docs: |
| all_tasks += new_doc.tasks |
| wiki_lists = [get_wikilist(t) for t in all_tasks] |
| flatten_wiki_list = list(set().union(*[set(w) for w in wiki_lists])) |
| return flatten_wiki_list |
|
|
| async def wiki_upload_and_store(self, 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) |
|
|
| """ |
| Generate Control |
| """ |
|
|
| def generate_doc_from_db(self, collection_name: str, from_files: [str]) -> [str]: |
|
|
| def query_from_task(task): |
| return get_public_paragraph(task) |
|
|
| gen_paths = [] |
|
|
| for new_doc in self.new_docs: |
| queries = [query_from_task(t) for t in new_doc.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(new_doc.tasks, texts_list)] |
|
|
| gen_path = f"{self.generated_docs_path}/{new_doc.name}e.docx" |
| gen_doc = new_doc.copy(gen_path) |
|
|
| gen_doc.replace_tasks(task_resolutions) |
| gen_doc.save_as_docx() |
| gen_paths.append(gen_doc.path) |
| self.gen_docs.append(gen_doc) |
| return gen_paths |
|
|
