mirror of
https://github.com/migatu/conjurer.git
synced 2026-07-18 07:42:09 +00:00
Tag: 1.27
Intermediate commits (oldest → newest): - Merge branch 'main' of https://github.com/migatu/conjurer - Update AI and librarian commands to use 'GENERAL' context" - Added the following changes to the codebase: - bgfx - fx and upgd - Latex incoming changes - installer fix - var fx - fx to fx - don't trust ai - tectonic - debug - debug - disable exc masking - upgrade - rollback - Figure out - aaa
This commit is contained in:
@@ -0,0 +1,233 @@
|
||||
# latex_functions.py
|
||||
# -*- coding: utf-8 -*-
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
import re
|
||||
import tempfile
|
||||
import zipfile
|
||||
from pathlib import Path
|
||||
from typing import Dict, List, Optional, Tuple
|
||||
|
||||
from ai_functions import handle_response
|
||||
|
||||
SAFE_TEX_NAME = re.compile(r"^[\w\-. ]+\.tex$", 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_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
|
||||
)
|
||||
|
||||
|
||||
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,
|
||||
]
|
||||
|
||||
|
||||
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")
|
||||
|
||||
if rc1 != 0:
|
||||
log_text = "".join(logs)
|
||||
pdf = workdir / (Path(tex_filename).stem + ".pdf")
|
||||
return False, log_text, pdf
|
||||
|
||||
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
|
||||
|
||||
|
||||
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:
|
||||
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