236 lines
11 KiB
Python
236 lines
11 KiB
Python
import hashlib
|
|
import re
|
|
from urllib.parse import urljoin
|
|
|
|
from app.config import settings
|
|
from app.services.paperless_client import PaperlessClient
|
|
from app.services.repository import get_repository
|
|
from app.services.store import JsonStore
|
|
|
|
|
|
class PaperlessService:
|
|
def __init__(self) -> None:
|
|
self.store = JsonStore()
|
|
self.repo = get_repository()
|
|
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"),
|
|
"assignment_name": submission.get("assignment_name"),
|
|
"submission_kind": submission.get("submission_kind") or "assignment_turn_in",
|
|
"semester": submission.get("semester"),
|
|
"course_code": submission.get("course_code"),
|
|
"paper_kind": submission.get("paper_kind"),
|
|
"notes": submission.get("notes", ""),
|
|
"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": 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:
|
|
return {
|
|
"target_url": urljoin(settings.n8n_base_url, settings.n8n_recording_webhook),
|
|
"class_name": recording.get("class_name"),
|
|
"lecture_title": f"{recording.get('class_name')} {recording.get('session_date')}",
|
|
"notes": recording.get("notes", ""),
|
|
"local_file": recording.get("mp3_path") or recording.get("stored_path"),
|
|
"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 = [
|
|
"D2L grade note imported from Doris Schoolhouse.",
|
|
f"Assignment: {assignment.get('title', 'unknown')}",
|
|
]
|
|
if assignment.get("grade"):
|
|
note_lines.append(f"Grade: {assignment['grade']}")
|
|
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)
|
|
|
|
def describe_grade_note_plan(self, assignment_id: str) -> dict:
|
|
submissions = self.repo.list_submissions()
|
|
linked = [s for s in submissions if str(s.get("assignment_id")) == str(assignment_id)]
|
|
return {
|
|
"status": "stub",
|
|
"backend": self.repo.backend,
|
|
"assignment_id": assignment_id,
|
|
"paperless_base_url": settings.paperless_base_url,
|
|
"linked_submission_count": len(linked),
|
|
"note_preview": self.build_grade_note(assignment_id),
|
|
"message": "Resolve linked Paperless submission document, hash latest grade/comment payload, and append a dated Schoolhouse note if changed.",
|
|
}
|
|
|
|
def append_grade_note(self, assignment_id: str) -> dict:
|
|
submissions = self.repo.list_submissions()
|
|
linked = [s for s in submissions if str(s.get("assignment_id")) == str(assignment_id)]
|
|
if not linked:
|
|
raise ValueError("No linked submissions for assignment")
|
|
target = next((s for s in linked if s.get("paperless_document_id")), linked[0])
|
|
document_id = target.get("paperless_document_id")
|
|
if not document_id:
|
|
raise ValueError("Linked submission has no Paperless document id yet")
|
|
note = self.build_grade_note(assignment_id)
|
|
note_hash = hashlib.sha256(note.encode('utf-8')).hexdigest()
|
|
if target.get("paperless_note_last_hash") == note_hash:
|
|
return {
|
|
"status": "skipped",
|
|
"reason": "note-unchanged",
|
|
"assignment_id": assignment_id,
|
|
"document_id": document_id,
|
|
"note_hash": note_hash,
|
|
}
|
|
response = self.client.add_note(document_id, note)
|
|
self.repo.update_submission(
|
|
str(target.get("id")),
|
|
{
|
|
"paperless_note_last_hash": note_hash,
|
|
"paperless_title": target.get("paperless_title"),
|
|
"status": target.get("status", "paperless-linked"),
|
|
},
|
|
)
|
|
return {
|
|
"status": "ok",
|
|
"assignment_id": assignment_id,
|
|
"document_id": document_id,
|
|
"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)
|