173 lines
7.5 KiB
Python
173 lines
7.5 KiB
Python
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(),
|
|
}
|