mirror of
https://github.com/migatu/conjurer.git
synced 2026-07-16 23:02:10 +00:00
Tag: 1.28
Intermediate commits (oldest → newest): - Partial + reformatting - upgrade to latest OpenAI API usage and logging improvements - refix - Vibe coding :P - Merge branch 'main' of https://github.com/migatu/conjurer - fx - fx - ffx - testing - ttt - fix - aaa - FX - aaa - sa
This commit is contained in:
+121
-175
@@ -1,149 +1,121 @@
|
||||
# latex_functions.py
|
||||
# -*- coding: utf-8 -*-
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
import re
|
||||
import shutil
|
||||
import tempfile
|
||||
import zipfile
|
||||
from pathlib import Path
|
||||
from typing import Dict, List, Optional, Tuple
|
||||
|
||||
from ai_functions import handle_response
|
||||
from typing import Dict, Iterable, List, Optional, Tuple
|
||||
|
||||
# ===== Bezpieczeństwo nazw/rozszerzeń =====
|
||||
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")
|
||||
return bool(SAFE_TEX_NAME.match(name or ""))
|
||||
|
||||
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:
|
||||
"""Filtruje discord.Attachment: tylko .tex + limit MB + nazwa bezpieczna."""
|
||||
return bool(
|
||||
getattr(att, "filename", "")
|
||||
and is_safe_tex_name(att.filename)
|
||||
and getattr(att, "size", 0) <= max_mb * 1024 * 1024
|
||||
# ===== Uruchamianie kompilatora =====
|
||||
|
||||
async def _run(cmd: List[str], cwd: Path, timeout_sec: int, logger_name: Optional[str]) -> Tuple[int, str]:
|
||||
"""Uruchamia proces kompilatora w cwd, zwraca (rc, sklejone_stdout_stderr)."""
|
||||
logger = logging.getLogger(logger_name) if logger_name else logging.getLogger()
|
||||
logger.debug("RUN %s (cwd=%s, timeout=%s)", " ".join(cmd), cwd, timeout_sec)
|
||||
|
||||
proc = await asyncio.create_subprocess_exec(
|
||||
*cmd,
|
||||
stdout=asyncio.subprocess.PIPE,
|
||||
stderr=asyncio.subprocess.STDOUT,
|
||||
cwd=str(cwd),
|
||||
)
|
||||
try:
|
||||
out = await asyncio.wait_for(proc.communicate(), timeout=timeout_sec)
|
||||
except asyncio.TimeoutError:
|
||||
try:
|
||||
proc.kill()
|
||||
finally:
|
||||
pass
|
||||
return 124, "TIMEOUT: compiler took too long"
|
||||
rc = proc.returncode
|
||||
log = (out[0] or b"").decode("utf-8", errors="ignore")
|
||||
logger.debug("RC=%s, LOG TAIL:\n%s", rc, log[-1500:])
|
||||
return rc, log
|
||||
|
||||
|
||||
def _build_cmd(engine: str, tex_filename: str) -> list[str]:
|
||||
base = (engine or "").strip().lower()
|
||||
if base in ("pdflatex", "xelatex", "lualatex"):
|
||||
return [
|
||||
base,
|
||||
"-interaction=nonstopmode",
|
||||
"-halt-on-error",
|
||||
"-file-line-error",
|
||||
"-no-shell-escape",
|
||||
tex_filename,
|
||||
]
|
||||
if base == "latexmk":
|
||||
return [
|
||||
"latexmk",
|
||||
"-pdf",
|
||||
"-interaction=nonstopmode",
|
||||
"-halt-on-error",
|
||||
"-file-line-error",
|
||||
tex_filename,
|
||||
]
|
||||
if base == "tectonic":
|
||||
return [
|
||||
"tectonic",
|
||||
"-X",
|
||||
"compile",
|
||||
"--keep-logs",
|
||||
"--outdir",
|
||||
".",
|
||||
tex_filename,
|
||||
]
|
||||
return [
|
||||
"pdflatex",
|
||||
"-interaction=nonstopmode",
|
||||
"-halt-on-error",
|
||||
"-file-line-error",
|
||||
"-no-shell-escape",
|
||||
tex_filename,
|
||||
]
|
||||
|
||||
def _build_cmd(tex_engine: str, main_tex: str) -> List[str]:
|
||||
"""Buduje komendę kompilatora; domyślnie tectonic/pdflatex/xelatex."""
|
||||
if tex_engine.lower() == "tectonic":
|
||||
# Tectonic sam robi wielofazowość, ale i tak uruchamiamy 2x (bezpiecznie dla bib/refs)
|
||||
return ["tectonic", "--keep-intermediates", "--synctex", main_tex]
|
||||
elif tex_engine.lower() in ("pdflatex", "xelatex", "lualatex"):
|
||||
return [tex_engine, "-interaction=nonstopmode", "-halt-on-error", main_tex]
|
||||
else:
|
||||
# fallback – traktujemy jak pdflatex
|
||||
return ["pdflatex", "-interaction=nonstopmode", "-halt-on-error", main_tex]
|
||||
|
||||
async def run_latex_two_passes(
|
||||
tex_filename: str,
|
||||
workdir: Path,
|
||||
tex_engine: str,
|
||||
timeout_sec: int,
|
||||
logger: str = None,
|
||||
):
|
||||
logger = logging.getLogger(logger) if logger else None
|
||||
|
||||
async def run_once(cmd: list[str]) -> tuple[int, str]:
|
||||
proc = await asyncio.create_subprocess_exec(
|
||||
*cmd,
|
||||
cwd=str(workdir),
|
||||
stdout=asyncio.subprocess.PIPE,
|
||||
stderr=asyncio.subprocess.STDOUT,
|
||||
)
|
||||
try:
|
||||
out_b, _ = await asyncio.wait_for(proc.communicate(), timeout=timeout_sec)
|
||||
return (proc.returncode or 0), (out_b or b"").decode(
|
||||
"utf-8", errors="replace"
|
||||
)
|
||||
except asyncio.TimeoutError:
|
||||
try:
|
||||
proc.kill()
|
||||
except Exception:
|
||||
raise
|
||||
return 124, f"[Timeout] Compilation exceeded {timeout_sec}s.\n"
|
||||
|
||||
logs = []
|
||||
cmd = _build_cmd(tex_engine, tex_filename)
|
||||
if logger:
|
||||
logger.debug("LaTeX cmd: %s (cwd=%s)", " ".join(cmd), workdir)
|
||||
|
||||
rc1, out1 = await run_once(cmd)
|
||||
logs.append(out1)
|
||||
second_pass = not ((tex_engine or "").strip().lower() == "tectonic")
|
||||
tex_filename: str, workdir: Path, tex_engine: str, timeout_sec: int, logger: Optional[str] = None
|
||||
) -> Tuple[bool, str, Path]:
|
||||
"""
|
||||
Uruchamia kompilację 2 razy (bezpieczeństwo referencji/bib).
|
||||
Zwraca: (ok, log_text, pdf_path).
|
||||
"""
|
||||
logger_obj = logging.getLogger(logger) if logger else logging.getLogger()
|
||||
logger_obj.info("Compiling %s with %s engine two passes", tex_filename, tex_engine)
|
||||
logs: List[str] = []
|
||||
pdf_path = workdir / (Path(tex_filename).stem + ".pdf")
|
||||
|
||||
# PASS 1
|
||||
rc1, log1 = await _run(_build_cmd(tex_engine, tex_filename), workdir, timeout_sec, logger)
|
||||
logs.append(log1)
|
||||
if rc1 != 0:
|
||||
log_text = "".join(logs)
|
||||
pdf = workdir / (Path(tex_filename).stem + ".pdf")
|
||||
return False, log_text, pdf
|
||||
return False, "".join(logs), pdf_path
|
||||
|
||||
if second_pass:
|
||||
rc2, out2 = await run_once(cmd)
|
||||
logs.append(out2)
|
||||
ok = rc2 == 0
|
||||
else:
|
||||
ok = rc1 == 0
|
||||
|
||||
log_text = "".join(logs)
|
||||
pdf = workdir / (Path(tex_filename).stem + ".pdf")
|
||||
ok = ok and pdf.exists()
|
||||
return ok, log_text, pdf
|
||||
# PASS 2 (często już nic nie robi, ale nie szkodzi)
|
||||
rc2, log2 = await _run(_build_cmd(tex_engine, tex_filename), workdir, timeout_sec, logger)
|
||||
logs.append(log2)
|
||||
ok = (rc2 == 0) and pdf_path.exists()
|
||||
return ok, "".join(logs), pdf_path
|
||||
|
||||
# ===== Kompilacje: pojedynczy TEX + ZIP =====
|
||||
|
||||
async def compile_single_tex_bytes(
|
||||
tex_bytes: bytes,
|
||||
filename: str,
|
||||
tex_engine: str,
|
||||
timeout_sec: int,
|
||||
logger: str = None,
|
||||
logger: Optional[str] = None,
|
||||
support_files: Optional[Iterable[Tuple[str, bytes]]] = None,
|
||||
) -> Dict[str, Optional[object]]:
|
||||
"""
|
||||
Kompiluje pojedynczy .tex + dowolne pliki pomocnicze (obrazy, sty/cls/bib, inne .tex).
|
||||
Każdy plik zapisujemy w katalogu roboczym widocznym dla kompilatora.
|
||||
"""
|
||||
log = logging.getLogger(logger) if logger else logging.getLogger()
|
||||
log.info("Compiling %s with %s engine single tex", filename, tex_engine)
|
||||
if not is_safe_tex_name(filename):
|
||||
return {
|
||||
"ok": False,
|
||||
"pdf_bytes": None,
|
||||
"log_text": "Invalid .tex filename",
|
||||
"pdf_name": None,
|
||||
}
|
||||
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
|
||||
)
|
||||
(wd / Path(filename).name).write_bytes(tex_bytes)
|
||||
|
||||
# Zapisz wszystkie supporty obok głównego pliku
|
||||
if support_files:
|
||||
for fname, data in support_files:
|
||||
if not fname or not is_safe_asset_name(fname):
|
||||
continue
|
||||
# spłaszczamy do nazwy bazowej (bez podkatalogów z zewn. źródeł)
|
||||
(wd / Path(fname).name).write_bytes(data)
|
||||
|
||||
ok, log_text, pdf_path = await run_latex_two_passes(Path(filename).name, wd, tex_engine, timeout_sec, logger)
|
||||
log.info("LaTeX compile finished: %s", "OK" if ok else "FAIL")
|
||||
log.info("Log text:\n%s", log_text[-2000:]) # Log last 2000 chars
|
||||
log.info("PDF path: %s", pdf_path)
|
||||
if ok and pdf_path.exists():
|
||||
return {
|
||||
"ok": True,
|
||||
@@ -151,83 +123,57 @@ async def compile_single_tex_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,
|
||||
}
|
||||
|
||||
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
|
||||
zip_bytes: bytes, tex_engine: str, timeout_sec: int, logger: Optional[str] = None
|
||||
) -> Tuple[bytes, List[str]]:
|
||||
"""
|
||||
Rozpakowuje ZIP, znajduję wszystkie .tex i dla KAŻDEGO tworzy osobny katalog roboczy
|
||||
ze skopiowanym CAŁYM drzewem, by zachować ścieżki względne obrazków/plików.
|
||||
Zwraca (zip_pdf_bytes, lista_rel_sciezek_które_padły).
|
||||
"""
|
||||
log = logging.getLogger(logger) if logger else logging.getLogger()
|
||||
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)
|
||||
src_root = td_path / "src"
|
||||
src_root.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
tex_files = [p for p in extract_dir.rglob("*.tex") if is_safe_tex_name(p.name)]
|
||||
# 1) Rozpakuj
|
||||
zpath = td_path / "in.zip"
|
||||
zpath.write_bytes(zip_bytes)
|
||||
with zipfile.ZipFile(zpath, "r") as zf:
|
||||
zf.extractall(src_root)
|
||||
|
||||
# 2) Zbierz .tex
|
||||
tex_files = [p for p in src_root.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:
|
||||
# 3) Kompiluj każdy .tex w izolacji (kopiujemy całe drzewo do osobnego workdiru)
|
||||
for tex_path in tex_files:
|
||||
rel_tex = tex_path.relative_to(src_root).as_posix()
|
||||
work = td_path / f"build_{tex_path.stem}"
|
||||
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)
|
||||
shutil.copytree(src_root, work / "src")
|
||||
ok, log_text, pdf = await run_latex_two_passes(rel_tex, work / "src", tex_engine, timeout_sec, logger)
|
||||
if ok and pdf.exists():
|
||||
out_pdf_paths.append(pdf)
|
||||
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)
|
||||
failed.append(rel_tex)
|
||||
log.info("LaTeX FAIL %s\n%s", rel_tex, (log_text or "")[-2000:])
|
||||
except Exception:
|
||||
# let it crash – log + re-raise (to zobaczy wyższa warstwa/discord.py)
|
||||
log.exception("Exception while compiling %s", rel_tex)
|
||||
raise
|
||||
|
||||
# 4) Spakuj wyjściowe PDF-y
|
||||
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:
|
||||
logger = logging.getLogger(logger) if logger else None
|
||||
sys = (
|
||||
"You are a LaTeX build diagnostician. Analyze pdflatex log and return:\n"
|
||||
"1) Root causes (bullets)\n2) Minimal working fix (code block)\n3) Alternative fix"
|
||||
)
|
||||
logger.debug("LaTeX DIAG (%s)\n%s", model, log_text[-9000:])
|
||||
usr = f"---LOG---\n{log_text[-9000:]}\n---END LOG---"
|
||||
if tex_excerpt:
|
||||
usr += f"\n---TEX EXCERPT---\n{tex_excerpt[:4000]}\n---END EXCERPT---"
|
||||
prompt = [{"role": sys, "content": "none"}, {"role": "user", "content": usr}]
|
||||
|
||||
response, _ = await handle_response(
|
||||
prompt,
|
||||
None,
|
||||
None,
|
||||
prompt,
|
||||
"latex_diagnosis",
|
||||
"NONE",
|
||||
algorithm=model,
|
||||
none_request=prompt,
|
||||
)
|
||||
return response
|
||||
|
||||
Reference in New Issue
Block a user