fizzlepoof/truenas-stacks
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Finish Schoolhouse programming submission flow

Browse Source
This commit is contained in:
Fizzlepoof
2026-05-16 19:13:10 +00:00
parent 1aa4b16eef
commit 754e98c6a5
15 changed files with 753 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files

138
home/doris-schoolhouse/app/services/paperless.py
View File

@@ -1,4 +1,5 @@
import hashlib
import re
from urllib.parse import urljoin
from app.config import settings
@@ -14,6 +15,10 @@ class PaperlessService:
self.client = PaperlessClient()
def assignment_intake_payload(self, submission: dict) -> dict:
local_files = submission.get("stored_paths") or []
if not local_files and submission.get("stored_path"):
local_files = [submission.get("stored_path")]
local_file = submission.get("receipt_pdf_path") or submission.get("bundle_path") or submission.get("stored_path")
return {
"target_url": urljoin(settings.n8n_base_url, settings.n8n_assignment_webhook),
"class_name": submission.get("class_name"),
@@ -26,8 +31,14 @@ class PaperlessService:
"source": submission.get("source") or "doris-schoolhouse",
"telegram_chat_id": submission.get("telegram_chat_id"),
"telegram_message_id": submission.get("telegram_message_id"),
"local_file": submission.get("stored_path"),
"local_file": local_file,
"local_files": local_files,
"version": submission.get("version_label"),
"batch_id": submission.get("batch_id"),
"file_count": submission.get("file_count"),
"artifact_kind": submission.get("artifact_kind") or "document",
"source_listing_path": submission.get("source_listing_path"),
"receipt_pdf_path": submission.get("receipt_pdf_path"),
}
def recording_intake_payload(self, recording: dict) -> dict:
@@ -40,21 +51,122 @@ class PaperlessService:
"source": "doris-schoolhouse",
}
def upload_source_bundle_receipt(self, submission: dict) -> dict:
file_path = submission.get("receipt_pdf_path") or submission.get("bundle_path") or submission.get("stored_path")
title = self.build_submission_title(submission)
return self.client.upload_document(file_path, title)
def _course_short_label(self, submission: dict) -> str:
course_code = str(submission.get("course_code") or "").strip()
match = re.search(r"([A-Z]{3,4})_([A-Z]{3,4})_(\d{4})", course_code)
if match:
return f"{match.group(2)}-{match.group(3)}"
class_name = str(submission.get("class_name") or "").strip()
match = re.search(r"([A-Za-z]+(?:\s+[A-Za-z]+)*)\s+(\d{4})", class_name)
if not match:
return class_name
words = match.group(1).split()
prefix = ''.join(word[0] for word in words[:4]).upper()
return f"{prefix}-{match.group(2)}"
def _course_tag_name(self, submission: dict) -> str | None:
label = self._course_short_label(submission)
if not label or "-" not in label:
return None
return label.replace("-", " ").lower()
def build_submission_title(self, submission: dict) -> str:
return " — ".join(
part for part in [
self._course_short_label(submission) or submission.get("class_name"),
submission.get("assignment_name"),
submission.get("version_label"),
"source bundle" if submission.get("artifact_kind") == "source_bundle" else submission.get("submission_kind"),
] if part
)
def build_direct_metadata_payload(self, submission: dict, current_document: dict | None = None) -> dict:
current_document = current_document or {}
known_tags = {
str(tag.get("name", "")).strip().casefold(): tag
for tag in self.client.list_tags()
}
merged_tags = {
int(tag_id)
for tag_id in (current_document.get("tags") or [])
if str(tag_id).isdigit()
}
for tag_name in ("school", "assignments", "spring-2026"):
tag = known_tags.get(tag_name)
if tag and tag.get("id"):
merged_tags.add(int(tag["id"]))
course_tag_name = self._course_tag_name(submission)
if course_tag_name:
tag = known_tags.get(course_tag_name)
if tag and tag.get("id"):
merged_tags.add(int(tag["id"]))
if submission.get("artifact_kind") == "source_bundle":
for tag_name in ("programming", "java"):
tag = known_tags.get(tag_name)
if tag and tag.get("id"):
merged_tags.add(int(tag["id"]))
payload = {
"title": self.build_submission_title(submission),
"tags": sorted(merged_tags),
}
for doc_type in self.client.list_document_types():
if str(doc_type.get("name", "")).strip().casefold() == "school" and doc_type.get("id"):
payload["document_type"] = int(doc_type["id"])
break
for correspondent in self.client.list_correspondents():
if str(correspondent.get("name", "")).strip().casefold() == "austin peay state university" and correspondent.get("id"):
payload["correspondent"] = int(correspondent["id"])
break
return payload
def enrich_direct_submission(self, submission: dict, upload_response: dict | None = None) -> dict:
upload_response = upload_response or {}
task_id = upload_response.get("task_id") or upload_response.get("id") or submission.get("paperless_task_id")
if not task_id:
raise ValueError("Paperless upload did not return a task id")
task = self.client.wait_for_task(str(task_id))
status = str(task.get("status") or "").upper()
document_id = task.get("related_document")
if status != "SUCCESS":
duplicate_match = re.search(r"#(\d+)\)", str(task.get("result") or ""))
if duplicate_match:
document_id = duplicate_match.group(1)
else:
raise ValueError(f"Paperless task did not complete successfully: {status or 'unknown'}")
if not document_id:
raise ValueError("Paperless task completed without a related document id")
current = self.client.get_document(document_id)
payload = self.build_direct_metadata_payload(submission, current)
updated = self.client.update_document(document_id, payload)
return {
"task": task,
"document": updated,
"document_id": updated.get("id") or current.get("id") or int(document_id),
"title": updated.get("title") or payload.get("title"),
"metadata_payload": payload,
}
def build_grade_note(self, assignment_id: str) -> str:
submissions = self.repo.list_submissions()
assignments = {item.get("id"): item for item in self.repo.list_assignments()}
assignment = assignments.get(str(assignment_id), {})
linked = [s for s in submissions if str(s.get("assignment_id")) == str(assignment_id)]
note_lines = [
"Grade sync from Doris Schoolhouse",
"D2L grade note imported from Doris Schoolhouse.",
f"Assignment: {assignment.get('title', 'unknown')}",
]
if assignment.get("grade"):
note_lines.append(f"Grade: {assignment['grade']}")
if assignment.get("points"):
note_lines.append(f"Points: {assignment['points']}")
if assignment.get("comments"):
note_lines.append(f"Professor comments: {assignment['comments']}")
comments = str(assignment.get("comments") or "").strip()
if comments:
note_lines.append(f"Comments: {comments}")
else:
note_lines.append("Comments: No professor comment captured in the D2L scrape.")
if linked:
note_lines.append(f"Linked submissions: {len(linked)}")
return "\n".join(note_lines)
@@ -107,3 +219,17 @@ class PaperlessService:
"note_hash": note_hash,
"paperless_response": response,
}
def append_grade_note_if_available(self, assignment_id: str) -> dict:
assignments = {item.get("id"): item for item in self.repo.list_assignments()}
assignment = assignments.get(str(assignment_id), {})
if not assignment:
return {"status": "skipped", "reason": "assignment-not-found", "assignment_id": assignment_id}
if not (assignment.get("grade") or str(assignment.get("comments") or "").strip()):
return {"status": "skipped", "reason": "no-grade-or-comments", "assignment_id": assignment_id}
linked = [s for s in self.repo.list_submissions() if str(s.get("assignment_id")) == str(assignment_id)]
if not linked:
return {"status": "skipped", "reason": "no-linked-submissions", "assignment_id": assignment_id}
if not any(item.get("paperless_document_id") for item in linked):
return {"status": "skipped", "reason": "no-paperless-document-id", "assignment_id": assignment_id}
return self.append_grade_note(assignment_id)