import hashlib import json from pathlib import Path from zipfile import ZIP_DEFLATED, ZipFile def slugify(value: str, fallback: str = "item") -> str: cleaned = "".join(ch.lower() if ch.isalnum() else "-" for ch in (value or fallback)) parts = [part for part in cleaned.split("-") if part] return "-".join(parts)[:80] or fallback class SubmissionBundleService: def _wrap_pdf_text(self, text: str, width: int = 96) -> list[str]: wrapped: list[str] = [] for raw_line in text.splitlines(): line = raw_line.rstrip() if not line: wrapped.append("") continue while len(line) > width: wrapped.append(line[:width]) line = line[width:] wrapped.append(line) return wrapped def _escape_pdf_text(self, value: str) -> str: return value.replace("\\", "\\\\").replace("(", "\\(").replace(")", "\\)") def _write_text_pdf(self, path: Path, title: str, body: str) -> None: lines = self._wrap_pdf_text(f"{title}\n\n{body}") page_height = 792 top = 756 line_height = 12 lines_per_page = 58 pages = [lines[i:i + lines_per_page] for i in range(0, len(lines), lines_per_page)] or [[""]] objects: list[bytes] = [] page_object_numbers: list[int] = [] object_number = 4 for page_lines in pages: content_lines = ["BT", "/F1 10 Tf", f"72 {top} Td"] first = True for line in page_lines: escaped = self._escape_pdf_text(line) if first: content_lines.append(f"({escaped}) Tj") first = False else: content_lines.append(f"0 -{line_height} Td ({escaped}) Tj") content_lines.append("ET") content_bytes = "\n".join(content_lines).encode("utf-8") page_object_numbers.append(object_number) objects.append( f"{object_number} 0 obj\n<< /Type /Page /Parent 2 0 R /MediaBox [0 0 612 {page_height}] /Contents {object_number + 1} 0 R /Resources << /Font << /F1 3 0 R >> >> >>\nendobj\n".encode("utf-8") ) objects.append( f"{object_number + 1} 0 obj\n<< /Length {len(content_bytes)} >>\nstream\n".encode("utf-8") + content_bytes + b"\nendstream\nendobj\n" ) object_number += 2 pages_kids = " ".join(f"{num} 0 R" for num in page_object_numbers) header = b"%PDF-1.4\n" root_objects = [ b"1 0 obj\n<< /Type /Catalog /Pages 2 0 R >>\nendobj\n", f"2 0 obj\n<< /Type /Pages /Kids [{pages_kids}] /Count {len(page_object_numbers)} >>\nendobj\n".encode("utf-8"), b"3 0 obj\n<< /Type /Font /Subtype /Type1 /BaseFont /Courier >>\nendobj\n", ] all_objects = root_objects + objects offsets = [] current = len(header) for obj in all_objects: offsets.append(current) current += len(obj) xref_start = current xref_lines = [b"xref\n", f"0 {len(all_objects) + 1}\n".encode("utf-8"), b"0000000000 65535 f \n"] for offset in offsets: xref_lines.append(f"{offset:010d} 00000 n \n".encode("utf-8")) trailer = f"trailer\n<< /Size {len(all_objects) + 1} /Root 1 0 R >>\nstartxref\n{xref_start}\n%%EOF\n".encode("utf-8") pdf_bytes = header + b"".join(all_objects) + b"".join(xref_lines) + trailer path.write_bytes(pdf_bytes) def build_bundle( self, *, upload_root: Path, submission_id: str, class_name: str, assignment_name: str, version_label: str, uploads: list[dict], notes: str = "", ) -> dict: bundle_root = upload_root / "submission-bundles" / submission_id source_root = bundle_root / "source" source_root.mkdir(parents=True, exist_ok=True) manifest_files = [] combined_hash = hashlib.sha256() for item in uploads: source_path = source_root / item["filename"] source_path.write_bytes(item["payload"]) combined_hash.update(item["payload"]) manifest_files.append( { "filename": item["filename"], "stored_path": str(source_path), "mime_type": item["mime_type"], "size_bytes": len(item["payload"]), "checksum_sha256": hashlib.sha256(item["payload"]).hexdigest(), } ) source_listing_path = bundle_root / f"{slugify(assignment_name, 'assignment')}-{slugify(version_label, 'v1')}-source-listing.txt" listing_parts = [] for item in manifest_files: listing_parts.append(f"===== {item['filename']} =====\n") listing_parts.append(Path(item["stored_path"]).read_text(encoding="utf-8", errors="replace")) listing_parts.append("\n\n") source_listing_path.write_text("".join(listing_parts), encoding="utf-8") manifest_path = bundle_root / "manifest.json" manifest = { "class_name": class_name, "assignment_name": assignment_name, "version_label": version_label, "notes": notes, "file_count": len(manifest_files), "files": manifest_files, "source_listing_path": str(source_listing_path), } manifest_path.write_text(json.dumps(manifest, indent=2, sort_keys=True), encoding="utf-8") bundle_name = f"{slugify(class_name, 'class')}-{slugify(assignment_name, 'assignment')}-{slugify(version_label, 'v1')}.zip" bundle_path = bundle_root / bundle_name with ZipFile(bundle_path, "w", compression=ZIP_DEFLATED) as archive: for item in manifest_files: archive.write(item["stored_path"], arcname=item["filename"]) archive.write(source_listing_path, arcname=source_listing_path.name) archive.write(manifest_path, arcname="manifest.json") receipt_pdf_path = bundle_root / f"{slugify(assignment_name, 'assignment')}-{slugify(version_label, 'v1')}-submission-receipt.pdf" receipt_body = "\n".join([ f"Class: {class_name}", f"Assignment: {assignment_name}", f"Version: {version_label}", f"Bundle file: {bundle_name}", f"Source file count: {len(manifest_files)}", "", "Included files:", *[f"- {item['filename']} ({item['size_bytes']} bytes)" for item in manifest_files], "", "Source listing:", source_listing_path.read_text(encoding="utf-8", errors="replace"), ]) self._write_text_pdf(receipt_pdf_path, f"{class_name} — {assignment_name} — {version_label}", receipt_body) return { "artifact_kind": "source_bundle", "bundle_path": str(bundle_path), "bundle_filename": bundle_name, "bundle_mime_type": "application/zip", "bundle_size_bytes": bundle_path.stat().st_size, "bundle_checksum_sha256": hashlib.sha256(bundle_path.read_bytes()).hexdigest(), "file_manifest_json": manifest_files, "file_count": len(manifest_files), "source_root": str(source_root), "source_listing_path": str(source_listing_path), "receipt_pdf_path": str(receipt_pdf_path), "manifest_path": str(manifest_path), "combined_source_checksum_sha256": combined_hash.hexdigest(), }