feat(tooling): JDG-11 remainder + RO-6 closure — validator runner, output generation, package import, archive axis [Wave C]

backend/problems/pdf_bundle.py

@@ -0,0 +1,190 @@
+"""Offline study-bundle PDFs (competitive roadmap RO-6).
+
+A dependency-free PDF writer: the archive's "PDF bundle downloads for
+offline study" is statement + sample tests in one printable file, and the
+backend deps stay pinned (no reportlab — a minimal Type1-Helvetica PDF is
+~80 lines and fully testable).
+
+Only latin-1 encodable text is written (Helvetica WinAnsi); characters
+outside it are replaced with '?' so any statement degrades visibly rather
+than crashing the download.
+"""
+
+PAGE_WIDTH = 595.28  # A4, points
+PAGE_HEIGHT = 841.89
+MARGIN = 56.0
+BODY_SIZE = 10.0
+HEADING_SIZE = 14.0
+LEADING = 14.0
+MAX_CHARS_PER_LINE = 88  # Helvetica 10pt on a ~483pt text column
+
+
+def _escape(text: str) -> str:
+    return (
+        text.replace("\\", "\\\\")
+        .replace("(", "\\(")
+        .replace(")", "\\)")
+        .encode("latin-1", errors="replace")
+        .decode("latin-1")
+    )
+
+
+def _wrap(text: str, width: int = MAX_CHARS_PER_LINE) -> list[str]:
+    """Character-wrap preserving existing line breaks (statements carry
+    their own layout; the bundle is for offline reading, not typesetting)."""
+    lines: list[str] = []
+    for raw_line in text.splitlines() or [""]:
+        if len(raw_line) <= width:
+            lines.append(raw_line)
+            continue
+        words = raw_line.split(" ")
+        current = ""
+        for word in words:
+            candidate = f"{current} {word}".strip()
+            if len(candidate) > width and current:
+                lines.append(current)
+                current = word
+            else:
+                current = candidate
+        lines.append(current)
+    return lines or [""]
+
+
+def build_pdf(sections: list[tuple[str, str]]) -> bytes:
+    """Render (heading, body) sections into a valid single-font PDF.
+
+    Returns raw PDF bytes with one page per ~52 lines; the caller streams
+    them as ``application/pdf``.
+    """
+    lines_per_page = int((PAGE_HEIGHT - 2 * MARGIN) // LEADING) - 1
+    pages: list[list[str]] = []
+    current: list[str] = []
+
+    def flush() -> None:
+        if current:
+            pages.append(current)
+
+    for heading, body in sections:
+        if current and len(current) + 2 > lines_per_page:
+            flush()
+            current = []
+        current.append(("heading", heading))
+        for line in _wrap(body):
+            if len(current) >= lines_per_page:
+                flush()
+                current = []
+            current.append(("body", line))
+    flush()
+    if not pages:
+        pages = [[("body", "")]]
+
+    content_streams: list[bytes] = []
+    for page_lines in pages:
+        ops: list[str] = ["BT"]
+        y = PAGE_HEIGHT - MARGIN
+        for kind, line in page_lines:
+            size = HEADING_SIZE if kind == "heading" else BODY_SIZE
+            font = "/F2" if kind == "heading" else "/F1"
+            ops.append(f"{font} {size:.1f} Tf")
+            ops.append(f"1 0 0 1 {MARGIN:.2f} {y:.2f} Tm")
+            ops.append(f"({_escape(line)}) Tj")
+            y -= LEADING
+        ops.append("ET")
+        content_streams.append("\n".join(ops).encode("latin-1", errors="replace"))
+
+    objects: list[bytes] = []
+    n_content = len(content_streams)
+    # Object numbering: 1 catalog, 2 pages, 3 F1, 4 F2, then page pairs.
+    first_page_obj = 5
+    page_ids = [first_page_obj + 2 * i for i in range(n_content)]
+    kids = " ".join(f"{pid} 0 R" for pid in page_ids)
+    objects.append(b"<< /Type /Catalog /Pages 2 0 R >>")
+    objects.append(f"<< /Type /Pages /Kids [{kids}] /Count {n_content} >>".encode())
+    objects.append(
+        b"<< /Type /Font /Subtype /Type1 /BaseFont /Helvetica /Encoding /WinAnsiEncoding >>"
+    )
+    objects.append(
+        b"<< /Type /Font /Subtype /Type1 /BaseFont /Helvetica-Bold /Encoding /WinAnsiEncoding >>"
+    )
+    for index, stream in enumerate(content_streams):
+        page_id = page_ids[index]
+        content_id = page_id + 1
+        objects.append(
+            (
+                f"<< /Type /Page /Parent 2 0 R /MediaBox [0 0 "
+                f"{PAGE_WIDTH:.2f} {PAGE_HEIGHT:.2f}] /Resources << /Font "
+                f"<< /F1 3 0 R /F2 4 0 R >> >> /Contents {content_id} 0 R >>"
+            ).encode()
+        )
+        objects.append(
+            b"<< /Length "
+            + str(len(stream)).encode()
+            + b" >>\nstream\n"
+            + stream
+            + b"\nendstream"
+        )
+
+    out = bytearray(b"%PDF-1.4\n%\xe2\xe3\xcf\xd3\n")
+    offsets = [0]
+    for number, body in enumerate(objects, start=1):
+        offsets.append(len(out))
+        out += f"{number} 0 obj\n".encode() + body + b"\nendobj\n"
+    xref_at = len(out)
+    out += f"xref\n0 {len(objects) + 1}\n".encode()
+    out += b"0000000000 65535 f \n"
+    for offset in offsets[1:]:
+        out += f"{offset:010d} 00000 n \n".encode()
+    out += (
+        f"trailer\n<< /Size {len(objects) + 1} /Root 1 0 R >>\n"
+        f"startxref\n{xref_at}\n%%EOF\n"
+    ).encode()
+    return bytes(out)
+
+
+def problem_bundle_pdf(problem, samples) -> bytes:
+    """The offline study bundle for one problem: title, statements (EN then
+    RO when present), limits, provenance and every sample test."""
+    title = (
+        problem.title_i18n.get("en") or next(iter(problem.title_i18n.values()), "")
+        if isinstance(problem.title_i18n, dict)
+        else ""
+    )
+    sections: list[tuple[str, str]] = [(title or f"Problem {problem.id}", "")]
+    for lang in ("en", "ro"):
+        statement = (
+            problem.statement_i18n.get(lang, "")
+            if isinstance(problem.statement_i18n, dict)
+            else ""
+        )
+        if statement:
+            sections.append((f"Statement ({lang.upper()})", statement))
+    sections.append(
+        (
+            "Limits",
+            f"Time: {problem.default_time_limit_ms} ms\n"
+            f"Memory: {problem.default_memory_limit_kb} KB",
+        )
+    )
+    if problem.source_contest or problem.source_year or problem.grade_group:
+        sections.append(
+            (
+                "Provenance",
+                " ".join(
+                    part
+                    for part in (
+                        problem.source_contest,
+                        str(problem.source_year or ""),
+                        problem.grade_group,
+                    )
+                    if part
+                ),
+            )
+        )
+    for index, test in enumerate(samples, start=1):
+        sections.append(
+            (
+                f"Sample {index}",
+                f"Input:\n{test.input_data}\nExpected output:\n{test.expected_output_data}",
+            )
+        )
+    return build_pdf(sections)

backend/problems/serializers.py

@@ -6,6 +6,7 @@
     normalize_localized_dict,
 )
 from .models import (
+    ImportSource,
     CurriculumNode,
     ProblemEditorial,
     Tag,
@@ -85,6 +86,22 @@ class ProblemSerializer(serializers.ModelSerializer):
     # from the registry multipliers/floors + this problem's overrides. A
     # read-only projection — authors edit limit_overrides, not this.
     effective_limits = serializers.SerializerMethodField()
+    # Archive provenance (competitive roadmap RO-6/CNT-2): where the
+    # problem came from. ``import_source`` points at the permission-ledger
+    # row covering the import; approve-time gating enforces that a problem
+    # wired to a ledger row only publishes while that row is GRANTED.
+    import_source = serializers.PrimaryKeyRelatedField(
+        queryset=ImportSource.objects.all(),
+        required=False,
+        allow_null=True,
+    )
+    import_source_name = serializers.SerializerMethodField()
+
+    def get_import_source_name(self, obj) -> str | None:
+        # Dotted sources raise on null FKs; the method form renders the
+        # ledger row's name or None without traversal gymnastics.
+        return obj.import_source.name if obj.import_source_id else None
+
     # Statement-language availability (i18n wave, owner rule (c)/(d)):
     # which languages the problem carries — clients render "available in
     # your language" badges and the EN-fallback affordance from this list
@@ -168,6 +185,12 @@ class Meta:
             "last_modified_date",
             "submission_count",
             "solved_count",
+            # Archive provenance (RO-6 browse axis + CNT-2 ledger link).
+            "source_contest",
+            "source_year",
+            "grade_group",
+            "import_source",
+            "import_source_name",
             "tags",
             "tag_ids",
             # 'test_cases' # Add if you want to nest test cases in problem detail view

backend/problems/test_pdf_bundle.py

@@ -0,0 +1,135 @@
+"""Unit tests for the dependency-free PDF writer (problems/pdf_bundle.py)."""
+
+from django.test import TestCase
+
+from problems.models import Problem
+from problems.models import TestCase as ProblemTestCase
+from problems.pdf_bundle import (
+    MAX_CHARS_PER_LINE,
+    _escape,
+    _wrap,
+    build_pdf,
+    problem_bundle_pdf,
+)
+
+
+class EscapeTests(TestCase):
+    def test_escapes_pdf_parens_and_backslash(self):
+        self.assertEqual(_escape("a(b)c"), "a\\(b\\)c")
+        self.assertEqual(_escape("back\\slash"), "back\\\\slash")
+
+    def test_non_latin1_replaced(self):
+        self.assertEqual(_escape("șăÎ"), "??Î")
+        self.assertEqual(_escape("plain"), "plain")
+
+
+class WrapTests(TestCase):
+    def test_short_lines_preserved(self):
+        self.assertEqual(_wrap("one\ntwo"), ["one", "two"])
+
+    def test_long_line_wraps_on_words(self):
+        wrapped = _wrap("word " * 40)
+        self.assertTrue(all(len(line) <= MAX_CHARS_PER_LINE for line in wrapped))
+        self.assertGreater(len(wrapped), 1)
+
+    def test_single_long_word_kept(self):
+        wrapped = _wrap("x" * 200)
+        self.assertEqual(wrapped, ["x" * 200])
+
+    def test_empty_text(self):
+        self.assertEqual(_wrap(""), [""])
+
+
+class BuildPdfTests(TestCase):
+    def test_valid_pdf_skeleton(self):
+        pdf = build_pdf([("Title", "Body line")])
+        self.assertTrue(pdf.startswith(b"%PDF-1.4"))
+        self.assertIn(b"/Type /Catalog", pdf)
+        self.assertIn(b"/Type /Pages", pdf)
+        self.assertIn(b"/Count 1", pdf)
+        self.assertIn(b"/F2 14.0 Tf", pdf)  # heading font + size
+        self.assertIn(b"(Body line) Tj", pdf)
+        self.assertTrue(pdf.rstrip().endswith(b"%%EOF"))
+        xref_at = int(pdf.rsplit(b"startxref", 1)[1].split()[0])
+        self.assertEqual(pdf[xref_at : xref_at + 4], b"xref")
+
+    def test_multiple_pages(self):
+        pdf = build_pdf([("Section", "\n".join(f"line {i}" for i in range(200)))])
+        self.assertIn(b"/Count 4", pdf)
+
+    def test_empty_sections_renders_blank_page(self):
+        pdf = build_pdf([])
+        self.assertTrue(pdf.startswith(b"%PDF-1.4"))
+        self.assertIn(b"/Count 1", pdf)
+
+    def test_escaped_content(self):
+        pdf = build_pdf([("T", "a(b)")])
+        self.assertIn(b"\\(", pdf)
+
+
+def _problem(**kw):
+    defaults = dict(
+        title_i18n={"en": "Sum", "ro": "Suma"},
+        statement_i18n={"en": "Add numbers.", "ro": "Adunați numerele."},
+        status=Problem.ProblemStatus.APPROVED,
+    )
+    defaults.update(kw)
+    return Problem.objects.create(**defaults)
+
+
+class ProblemBundlePdfTests(TestCase):
+    def test_full_bundle(self):
+        problem = _problem(
+            source_contest="OJI 2024",
+            source_year=2024,
+            grade_group="IX-X",
+        )
+        ProblemTestCase.objects.create(
+            problem=problem,
+            input_data="1 2",
+            expected_output_data="3",
+            is_sample=True,
+            points=10,
+        )
+        pdf = problem_bundle_pdf(problem, problem.test_cases.all())
+        self.assertTrue(pdf.startswith(b"%PDF-1.4"))
+        # Headings are PDF-escaped (parens) inside the content stream.
+        escaped_en = "(Statement \\(EN\\)) Tj".encode()
+        escaped_ro = "(Statement \\(RO\\)) Tj".encode()
+        self.assertIn(escaped_en, pdf)
+        self.assertIn(escaped_ro, pdf)
+        self.assertIn(b"(Limits) Tj", pdf)
+        self.assertIn(b"(Provenance) Tj", pdf)
+        self.assertIn(b"(Sample 1) Tj", pdf)
+        self.assertIn(b"(OJI 2024 2024 IX-X) Tj", pdf)
+
+    def test_no_provenance_section_when_plain(self):
+        problem = _problem()
+        pdf = problem_bundle_pdf(problem, [])
+        self.assertNotIn(b"(Provenance) Tj", pdf)
+
+    def test_title_fallback_to_first_available(self):
+        problem = _problem(title_i18n={"ro": "Doar română"})
+        pdf = problem_bundle_pdf(problem, [])
+        self.assertIn(b"Doar", pdf)
+
+
+class PaginationBoundaryTests(TestCase):
+    """The section-heading page-break path (build_pdf lines 68-70)."""
+
+    def test_new_section_flushes_full_page(self):
+        # One full page of body lines, then a second section whose heading
+        # lands past the page capacity -> the heading opens page 2.
+        first_body = "\n".join(f"l{i}" for i in range(50))
+        pdf = build_pdf([("A", first_body), ("B", "tail")])
+        self.assertIn(b"/Count 2", pdf)
+        self.assertIn(b"(B) Tj", pdf)
+
+
+class MissingStatementLocaleTests(TestCase):
+    def test_empty_ro_statement_skips_section(self):
+        problem = _problem(statement_i18n={"en": "Only EN.", "ro": ""})
+        pdf = problem_bundle_pdf(problem, [])
+        # Headings are paren-escaped inside the PDF content stream.
+        self.assertIn("(Statement \(EN\)) Tj".encode(), pdf)
+        self.assertNotIn("(Statement \(RO\)) Tj".encode(), pdf)