Personal tex compilation fixes
This commit is contained in:
1 parent
ec6ac9580b
commit
705db1068f
8 files changed
+217
-21
No files matched your search
@@ -1,6 +1,7 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import hashlib
|
||||
import io
|
||||
import json
|
||||
import re
|
||||
@@ -1016,24 +1017,35 @@ def write_exam_outputs(workspace: EvaluationWorkspace, grouped_extraction: Group
|
||||
save_missing(path, '\n\n'.join(contents))
|
||||
|
||||
print(f"Wrote {len(grouped_extraction.groups)} groups; checking PDF compilation...")
|
||||
# ==========================================
|
||||
# PDF COMPILATION (4 Threads)
|
||||
# ==========================================
|
||||
all_tex_files = list(text2_dir.glob("*.tex")) + list(sol2_dir.glob("*.tex"))
|
||||
compilation_status = compile_existing_pdfs(workspace)
|
||||
return ExitCode.PARTIAL if processing_errors or compilation_status != ExitCode.SUCCESS else ExitCode.SUCCESS
|
||||
|
||||
|
||||
def compile_existing_pdfs(workspace: EvaluationWorkspace) -> ExitCode:
|
||||
"""Compile existing fragments without reading checkpoints or calling Gemini."""
|
||||
folder = workspace.root
|
||||
all_tex_files = list(workspace.rendered_text_dir.glob("*.tex")) + list(workspace.rendered_solution_dir.glob("*.tex"))
|
||||
|
||||
def compile_worker(tex_file: Path) -> str | None:
|
||||
"""Helper to read content and call the utility function."""
|
||||
try:
|
||||
pdf_path = tex_file.with_suffix('.pdf')
|
||||
status_path = folder / 'Cache' / 'statement_pdf_status' / tex_file.parent.name / (tex_file.stem + '.json')
|
||||
failed_before = status_path.exists() and not json.loads(status_path.read_text()).get('complete')
|
||||
source_path = (workspace.correction_source_file if tex_file.parent.name == 'Sol2'
|
||||
else workspace.statement_source_file)
|
||||
if not source_path.is_file():
|
||||
source_path = None
|
||||
fingerprint = hashlib.sha256((utils.fragment_latex_header(source_path)
|
||||
+ utils.LATEX_AFTER + str(source_path)).encode()).hexdigest()
|
||||
status = json.loads(status_path.read_text()) if status_path.exists() else {}
|
||||
failed_before = not status.get('complete') or status.get('preamble_hash') != fingerprint
|
||||
if (pdf_path.is_file() and pdf_path.stat().st_size > 0
|
||||
and pdf_path.stat().st_mtime_ns >= tex_file.stat().st_mtime_ns and not failed_before):
|
||||
return None
|
||||
atomic_write_json(status_path, {'complete': False})
|
||||
atomic_write_json(status_path, {'complete': False, 'preamble_hash': fingerprint})
|
||||
content = tex_file.read_text(encoding='utf-8')
|
||||
compile_to_pdf(content, pdf_path, strict=True)
|
||||
atomic_write_json(status_path, {'complete': True})
|
||||
compile_to_pdf(content, pdf_path, strict=True, source_path=source_path)
|
||||
atomic_write_json(status_path, {'complete': True, 'preamble_hash': fingerprint})
|
||||
except Exception as e: # noqa: BLE001 - compiler worker boundary
|
||||
return f"Error compiling {tex_file.name}: {e}"
|
||||
return None
|
||||
@@ -1045,8 +1057,7 @@ def write_exam_outputs(workspace: EvaluationWorkspace, grouped_extraction: Group
|
||||
]
|
||||
for error in compile_errors:
|
||||
report_error(folder, 'PDF compilation', error)
|
||||
processing_errors.extend(compile_errors)
|
||||
return ExitCode.PARTIAL if processing_errors else ExitCode.SUCCESS
|
||||
return ExitCode.PARTIAL if compile_errors else ExitCode.SUCCESS
|
||||
|
||||
|
||||
def build_parser() -> argparse.ArgumentParser:
|
||||
@@ -1056,6 +1067,8 @@ def build_parser() -> argparse.ArgumentParser:
|
||||
help="Regroup existing questions with Gemini; update only label_groups")
|
||||
actions.add_argument("--persp-only", action="store_true",
|
||||
help="Generate only missing Persp rubrics for existing groups")
|
||||
actions.add_argument("--compile-only", action="store_true",
|
||||
help="Retry existing fragment PDFs using source preambles; no Gemini requests")
|
||||
parser.add_argument('--delete-persp', action='store_true',
|
||||
help='Delete current Persp before generation (requires --persp-only)')
|
||||
parser.add_argument(
|
||||
@@ -1071,6 +1084,10 @@ def main(argv: Sequence[str] | None = None) -> int:
|
||||
def run(args):
|
||||
if args.delete_persp and not args.persp_only:
|
||||
raise CliError('--delete-persp requires --persp-only', ExitCode.INVALID_ARGUMENTS)
|
||||
if args.restart and (args.compile_only or args.groups_only or args.persp_only):
|
||||
raise CliError('--restart cannot be combined with a partial action', ExitCode.INVALID_ARGUMENTS)
|
||||
if args.compile_only:
|
||||
return compile_existing_pdfs(workspace_from_args(args))
|
||||
if args.groups_only or args.persp_only:
|
||||
kwargs = {'delete_persp': True} if args.delete_persp else {}
|
||||
return refine_existing(workspace_from_args(args),
|
||||
|
||||
+39
-3
@@ -149,9 +149,43 @@ def report_latex_warning(output_pdf_path) -> None:
|
||||
print(f"{LATEX_WARNING_MARKER}{Path(output_pdf_path).resolve()}")
|
||||
|
||||
|
||||
def compile_to_pdf(text, output_pdf_path, *, strict=False):
|
||||
def fragment_latex_header(source_path=None):
|
||||
"""Keep source packages/macros while using the configured fragment layout."""
|
||||
if source_path is None:
|
||||
return LATEX_BEFORE
|
||||
source = Path(source_path).read_text(encoding="utf-8")
|
||||
# Ignore commented document commands when locating the preamble.
|
||||
source = re.sub(r"(?m)(?<!\\)%[^\n]*", "", source)
|
||||
begin = re.search(r"\\begin\s*\{document\}", source)
|
||||
document_class = r"\\documentclass(?:\s*\[[^\]]*\])?\s*\{[^}]+\}"
|
||||
configured_class = re.search(document_class, LATEX_BEFORE)
|
||||
configured_begin = re.search(r"\\begin\s*\{document\}", LATEX_BEFORE)
|
||||
if begin is None or configured_class is None or configured_begin is None:
|
||||
return LATEX_BEFORE
|
||||
preamble = source[:begin.start()]
|
||||
if not re.search(document_class, preamble):
|
||||
return LATEX_BEFORE
|
||||
preamble = re.sub(document_class, lambda _: configured_class.group(), preamble, count=1)
|
||||
# geometry overrides standalone's crop coordinates and can produce blank or
|
||||
# clipped PDFs even when LaTeX succeeds. Keep other source packages intact.
|
||||
def without_geometry(match):
|
||||
packages = [name.strip() for name in match.group(2).split(',')
|
||||
if name.strip() != 'geometry']
|
||||
return (r'\usepackage' + (match.group(1) or '') + '{' + ','.join(packages) + '}'
|
||||
if packages else '')
|
||||
preamble = re.sub(r'\\usepackage\s*(\[[^\]]*\])?\s*\{([^}]+)\}',
|
||||
without_geometry, preamble)
|
||||
preamble = re.sub(r'\\(?:geometry|newgeometry)\s*\{[^}]*\}', '', preamble)
|
||||
return preamble + LATEX_BEFORE[configured_begin.start():]
|
||||
|
||||
|
||||
def compile_to_pdf(text, output_pdf_path, *, strict=False, source_path=None):
|
||||
"""Wraps text in standalone header/footer templates and compiles it to PDF."""
|
||||
latex_template = f"{LATEX_BEFORE}{text}{LATEX_AFTER}"
|
||||
latex_template = f"{fragment_latex_header(source_path)}{text}{LATEX_AFTER}"
|
||||
env = os.environ.copy()
|
||||
if source_path is not None:
|
||||
source_dir = str(Path(source_path).resolve().parent)
|
||||
env['TEXINPUTS'] = source_dir + os.pathsep + env.get('TEXINPUTS', '')
|
||||
|
||||
with tempfile.TemporaryDirectory() as temp_dir:
|
||||
tex_filename = 'text.tex'
|
||||
@@ -170,6 +204,7 @@ def compile_to_pdf(text, output_pdf_path, *, strict=False):
|
||||
result = subprocess.run(
|
||||
['pdflatex', '-interaction=nonstopmode', tex_filename],
|
||||
cwd=temp_dir,
|
||||
env=env,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.STDOUT,
|
||||
text=True,
|
||||
@@ -180,6 +215,7 @@ def compile_to_pdf(text, output_pdf_path, *, strict=False):
|
||||
result = subprocess.run(
|
||||
['pdflatex', '-interaction=nonstopmode', tex_filename],
|
||||
cwd=temp_dir,
|
||||
env=env,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.STDOUT,
|
||||
text=True,
|
||||
@@ -188,7 +224,7 @@ def compile_to_pdf(text, output_pdf_path, *, strict=False):
|
||||
|
||||
generated_pdf = os.path.join(temp_dir, pdf_filename)
|
||||
pdf_generated = os.path.exists(generated_pdf)
|
||||
if pdf_generated:
|
||||
if pdf_generated and (not strict or result.returncode == 0):
|
||||
shutil.move(generated_pdf, output_pdf_path)
|
||||
|
||||
if result.returncode != 0:
|
||||
|
||||
+12
-5
@@ -65,8 +65,15 @@ class EvaluationWorkspace:
|
||||
)
|
||||
if not path.is_dir():
|
||||
return False
|
||||
aliases = (f"{path.name}.pdf", f"{path.name}.tex", f"{path.name}c.tex")
|
||||
return any((path / marker).exists() for marker in (*markers, *aliases))
|
||||
workspace = EvaluationWorkspace(path)
|
||||
inputs = (
|
||||
candidate
|
||||
for name in ("enonce.pdf", "enonce.tex", "correction.tex")
|
||||
for candidate in workspace.input_candidates(name)
|
||||
)
|
||||
return any((path / marker).exists() for marker in markers) or any(
|
||||
candidate.exists() for candidate in inputs
|
||||
)
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
@@ -75,9 +82,9 @@ class EvaluationWorkspace:
|
||||
def input_candidates(self, canonical_name: str) -> tuple[Path, ...]:
|
||||
"""Return preferred paths for one of the three statement inputs."""
|
||||
aliases = {
|
||||
"enonce.pdf": ("enonce.pdf", "énoncé.pdf", f"{self.name}.pdf"),
|
||||
"enonce.tex": ("enonce.tex", f"{self.name}.tex"),
|
||||
"correction.tex": ("correction.tex", f"{self.name}c.tex"),
|
||||
"enonce.pdf": ("enonce.pdf", "énonce.pdf", "énoncé.pdf", f"{self.name}.pdf"),
|
||||
"enonce.tex": ("enonce.tex", "énonce.tex", "énoncé.tex", f"{self.name}.tex"),
|
||||
"correction.tex": ("correction.tex", "corrige.tex", "corrigé.tex", f"{self.name}c.tex"),
|
||||
}
|
||||
names = aliases.get(canonical_name, (canonical_name,))
|
||||
return tuple(self.root / name for name in names)
|
||||
|
||||
Reference in new issue
Block a user