Partial + reformatting

ai_functions.py

@@ -54,7 +54,9 @@ async def _openai_call(messages, model, temperature=0.2):
             input=messages,
         )
         # SDK zapewnia output_text dla zwykłych odpowiedzi
-        return (getattr(resp.output_text, "output_text", None) or "").strip() or str(resp.output_text)
+        return (getattr(resp.output_text, "output_text", None) or "").strip() or str(
+            resp.output_text
+        )
 
 
 def create_vector_store():
@@ -65,6 +67,7 @@ def create_vector_store():
     # "days": 7}
     # )
 
+
 def upload_files_to_vector_store(assistant):
 
     # Ready the files for upload to OpenAI
@@ -80,7 +83,6 @@ def upload_files_to_vector_store(assistant):
     # file_ids=['file_1', 'file_2', 'file_3', 'file_4', 'file_5']
     # )
 
-
     # Use the upload and poll SDK helper to upload the files, add them to the vector store,
     # and poll the status of the file batch for completion.
     file_batch = OPENAICLIENT.beta.vector_stores.file_batches.upload_and_poll(
@@ -95,6 +97,7 @@ def upload_files_to_vector_store(assistant):
         tool_resources={"file_search": {"vector_store_ids": [VECTOR_STORE_ID]}},
     )
 
+
 def delete_files_from_vector_store(assistant, file_id):
     result = OPENAICLIENT.beta.vector_stores.file_batches.delete(
         vector_store_id=VECTOR_STORE_ID, files=file_id
@@ -133,7 +136,14 @@ def num_tokens_from_string(message, model):
 
 
 async def handle_response(
-    prompt, vykidailo, bartender, history, username, request_type, algorithm="gpt-4o", none_request=""
+    prompt,
+    vykidailo,
+    bartender,
+    history,
+    username,
+    request_type,
+    algorithm="gpt-4o",
+    none_request="",
 ):
     """
     Handle responses by appending them to a history, use OpenAI to
@@ -222,7 +232,6 @@ async def handle_response(
             table = []
             token_amount = 10000
 
-
         prompt_gpt_request_size = num_tokens_from_string(
             {"role": "user", "content": final_prompt}, "gpt-4"
         )
@@ -254,7 +263,7 @@ async def handle_response(
         deadline = time.time() + timeout_sec
         response = await asyncio.wait_for(
             _openai_call(messages=history, model=algorithm),
-            timeout=max(0.1, deadline - time.time())
+            timeout=max(0.1, deadline - time.time()),
         )
 
     except openai.APITimeoutError as e:
@@ -336,6 +345,7 @@ async def handle_response(
     else:
         return response, []
 
+
 async def get_random_cyclic_message(client):
     """
     The function `get_random_cyclic_message` returns a random cyclic message from a list of cyclic

latex_functions.py

@@ -11,12 +11,16 @@ from pathlib import Path
 from typing import Dict, List, Optional, Tuple
 
 from ai_functions import handle_response
+import shutil
+from typing import Iterable, Tuple
 
-SAFE_TEX_NAME = re.compile(r"^[\w\-. ]+\.tex$", re.IGNORECASE)
+SAFE_ASSET_NAME = re.compile(
+    r"^[\w\-. /]+?\.(tex|png|jpg|jpeg|pdf|svg|eps|bmp|gif|sty|cls|bib|bst|bbx|cbx|def|cfg)$",
+    re.IGNORECASE,
+)
 
-
-def is_safe_tex_name(name: str) -> bool:
-    return bool(SAFE_TEX_NAME.match(name or "")) and name.lower().endswith(".tex")
+def is_safe_asset_name(name: str) -> bool:
+    return bool(SAFE_ASSET_NAME.match(name or ""))
 
 
 def is_tex_attachment(att, max_mb: int) -> bool:
@@ -122,90 +126,6 @@ async def run_latex_two_passes(
     ok = ok and pdf.exists()
     return ok, log_text, pdf
 
-
-async def compile_single_tex_bytes(
-    tex_bytes: bytes,
-    filename: str,
-    tex_engine: str,
-    timeout_sec: int,
-    logger: str = None,
-) -> Dict[str, Optional[object]]:
-    if not is_safe_tex_name(filename):
-        return {
-            "ok": False,
-            "pdf_bytes": None,
-            "log_text": "Invalid .tex filename",
-            "pdf_name": None,
-        }
-    with tempfile.TemporaryDirectory(prefix="latex_one_") as td:
-        wd = Path(td)
-        tex_path = wd / filename
-        tex_path.write_bytes(tex_bytes)
-        ok, log_text, pdf_path = await run_latex_two_passes(
-            filename, wd, tex_engine, timeout_sec, logger=logger
-        )
-        if ok and pdf_path.exists():
-            return {
-                "ok": True,
-                "pdf_bytes": pdf_path.read_bytes(),
-                "log_text": log_text,
-                "pdf_name": pdf_path.name,
-            }
-        return {
-            "ok": False,
-            "pdf_bytes": None,
-            "log_text": log_text,
-            "pdf_name": pdf_path.name,
-        }
-
-
-async def compile_zip_to_zip(
-    zip_bytes: bytes, tex_engine: str, timeout_sec: int, logger: str = None
-) -> Tuple[bytes, List[str]]:
-    failed: List[str] = []
-    out_pdf_paths: List[Path] = []
-    with tempfile.TemporaryDirectory(prefix="latex_zip_") as td:
-        td_path = Path(td)
-        in_zip = td_path / "in.zip"
-        in_zip.write_bytes(zip_bytes)
-        extract_dir = td_path / "in"
-        extract_dir.mkdir(parents=True, exist_ok=True)
-        with zipfile.ZipFile(in_zip, "r") as zf:
-            zf.extractall(extract_dir)
-
-        tex_files = [p for p in extract_dir.rglob("*.tex") if is_safe_tex_name(p.name)]
-        if not tex_files:
-            return b"", ["No .tex files found in archive."]
-
-        for tex in tex_files:
-            try:
-                work = td_path / f"build_{tex.stem}"
-                work.mkdir(parents=True, exist_ok=True)
-                target_tex = work / tex.name
-                target_tex.write_bytes(tex.read_bytes())
-                ok, log_text, pdf_path = await run_latex_two_passes(
-                    target_tex.name, work, tex_engine, timeout_sec, logger=logger
-                )
-                if ok and pdf_path.exists():
-                    out_pdf_paths.append(pdf_path)
-                else:
-                    failed.append(tex.name)
-                    if logger:
-                        logger.debug("BATCH FAIL %s\n%s", tex.name, log_text[-1500:])
-            except Exception as e:
-                failed.append(f"{tex.name} (exception: {e})")
-                if logger:
-                    logger.debug("BATCH EXC %s: %s", tex.name, e)
-                raise
-
-        out_zip = td_path / "compiled_pdfs.zip"
-        with zipfile.ZipFile(out_zip, "w", compression=zipfile.ZIP_DEFLATED) as zf:
-            for pdf in out_pdf_paths:
-                zf.write(pdf, arcname=pdf.name)
-
-        return out_zip.read_bytes(), failed
-
-
 async def ask_openai_diagnosis(
     log_text: str, tex_excerpt: str, model: str, logger: str = None
 ) -> str: