Finish Schoolhouse programming submission flow
This commit is contained in:
Fizzlepoof
172
home/doris-schoolhouse/app/services/bundles.py
Normal file
172
home/doris-schoolhouse/app/services/bundles.py
Normal file
@@ -0,0 +1,172 @@
|
||||
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(),
|
||||
}
|
||||
Reference in New Issue
Block a user