406 lines
19 KiB
Python
406 lines
19 KiB
Python
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from app.config import settings
|
|
from app.services.store import JsonStore
|
|
|
|
|
|
class JsonRepository:
|
|
backend = "json"
|
|
|
|
def __init__(self) -> None:
|
|
self.store = JsonStore()
|
|
|
|
def list_courses(self) -> list[dict[str, Any]]:
|
|
return self.store.load("courses", [])
|
|
|
|
def replace_courses(self, items: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
return self.store.replace_collection("courses", items)
|
|
|
|
def list_assignments(self) -> list[dict[str, Any]]:
|
|
return self.store.load("assignments", [])
|
|
|
|
def replace_assignments(self, items: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
return self.store.replace_collection("assignments", items)
|
|
|
|
def list_sync_runs(self) -> list[dict[str, Any]]:
|
|
return self.store.load("sync_runs", [])
|
|
|
|
def save_sync_runs(self, items: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
return self.store.save("sync_runs", items)
|
|
|
|
def list_submissions(self) -> list[dict[str, Any]]:
|
|
return self.store.load("submissions", [])
|
|
|
|
def create_submission(self, item: dict[str, Any]) -> dict[str, Any]:
|
|
return self.store.append_collection("submissions", item)
|
|
|
|
def update_submission(self, submission_id: str, updates: dict[str, Any]) -> dict[str, Any] | None:
|
|
return self.store.update_collection_item(
|
|
"submissions",
|
|
lambda item: str(item.get("id")) == str(submission_id),
|
|
lambda item: {**item, **updates},
|
|
)
|
|
|
|
def list_recordings(self) -> list[dict[str, Any]]:
|
|
return self.store.load("recordings", [])
|
|
|
|
def create_recording(self, item: dict[str, Any]) -> dict[str, Any]:
|
|
return self.store.append_collection("recordings", item)
|
|
|
|
def update_recording(self, recording_id: str, updates: dict[str, Any]) -> dict[str, Any] | None:
|
|
return self.store.update_collection_item(
|
|
"recordings",
|
|
lambda item: str(item.get("id")) == str(recording_id),
|
|
lambda item: {**item, **updates},
|
|
)
|
|
|
|
|
|
class PostgresRepository:
|
|
backend = "postgres"
|
|
|
|
def __init__(self) -> None:
|
|
import psycopg
|
|
from psycopg.rows import dict_row
|
|
|
|
self._psycopg = psycopg
|
|
self._dict_row = dict_row
|
|
self.dsn = settings.database_url.replace("postgresql+psycopg://", "postgresql://")
|
|
|
|
def _connect(self):
|
|
return self._psycopg.connect(self.dsn, row_factory=self._dict_row)
|
|
|
|
def _ensure_course(self, cur, item: dict[str, Any]) -> dict[str, Any] | None:
|
|
course_id = item.get("class_id") or item.get("course_id")
|
|
if not course_id:
|
|
return None
|
|
cur.execute("SELECT id, d2l_course_id, name FROM schoolhouse_course WHERE d2l_course_id = %s", (course_id,))
|
|
row = cur.fetchone()
|
|
if row:
|
|
return row
|
|
course_name = item.get("class_name") or item.get("course_name") or str(course_id)
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO schoolhouse_course (d2l_course_id, name, last_synced_at)
|
|
VALUES (%s, %s, NOW())
|
|
RETURNING id, d2l_course_id, name
|
|
""",
|
|
(course_id, course_name),
|
|
)
|
|
return cur.fetchone()
|
|
|
|
def _ensure_assignment(self, cur, item: dict[str, Any]) -> dict[str, Any] | None:
|
|
course = self._ensure_course(cur, item)
|
|
if not course:
|
|
return None
|
|
assignment_id = item.get("assignment_id")
|
|
assignment_name = item.get("assignment_name") or item.get("title")
|
|
if assignment_id:
|
|
cur.execute(
|
|
"SELECT id, title FROM schoolhouse_assignment WHERE course_id = %s AND d2l_assignment_key = %s LIMIT 1",
|
|
(course["id"], assignment_id),
|
|
)
|
|
row = cur.fetchone()
|
|
if row:
|
|
return row
|
|
if assignment_name:
|
|
cur.execute(
|
|
"SELECT id, title FROM schoolhouse_assignment WHERE course_id = %s AND title = %s LIMIT 1",
|
|
(course["id"], assignment_name),
|
|
)
|
|
row = cur.fetchone()
|
|
if row:
|
|
return row
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO schoolhouse_assignment (course_id, d2l_assignment_key, title, source, status, created_at, updated_at)
|
|
VALUES (%s, %s, %s, %s, %s, NOW(), NOW())
|
|
RETURNING id, title
|
|
""",
|
|
(course["id"], assignment_id, assignment_name, item.get("source", "schoolhouse-intake"), item.get("status", "assigned")),
|
|
)
|
|
return cur.fetchone()
|
|
return None
|
|
|
|
def list_courses(self) -> list[dict[str, Any]]:
|
|
with self._connect() as conn, conn.cursor() as cur:
|
|
cur.execute("SELECT id::text AS id, d2l_course_id, name, code, semester, last_synced_at FROM schoolhouse_course ORDER BY name")
|
|
return list(cur.fetchall())
|
|
|
|
def replace_courses(self, items: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
with self._connect() as conn, conn.cursor() as cur:
|
|
for item in items:
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO schoolhouse_course (d2l_course_id, name, code, semester, last_synced_at)
|
|
VALUES (%s, %s, %s, %s, NOW())
|
|
ON CONFLICT (d2l_course_id)
|
|
DO UPDATE SET name = EXCLUDED.name, code = EXCLUDED.code, semester = EXCLUDED.semester, last_synced_at = NOW()
|
|
""",
|
|
(item.get("d2l_course_id"), item.get("name"), item.get("code"), item.get("semester")),
|
|
)
|
|
conn.commit()
|
|
return items
|
|
|
|
def list_assignments(self) -> list[dict[str, Any]]:
|
|
with self._connect() as conn, conn.cursor() as cur:
|
|
cur.execute(
|
|
"""
|
|
SELECT a.id::text AS id,
|
|
c.d2l_course_id AS course_id,
|
|
c.name AS course_name,
|
|
a.title,
|
|
a.grade_text AS grade,
|
|
a.professor_comments AS comments,
|
|
a.status,
|
|
a.updated_at
|
|
FROM schoolhouse_assignment a
|
|
JOIN schoolhouse_course c ON c.id = a.course_id
|
|
ORDER BY c.name, a.title
|
|
"""
|
|
)
|
|
return list(cur.fetchall())
|
|
|
|
def replace_assignments(self, items: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
with self._connect() as conn, conn.cursor() as cur:
|
|
for item in items:
|
|
cur.execute("SELECT id FROM schoolhouse_course WHERE d2l_course_id = %s", (item.get("course_id"),))
|
|
row = cur.fetchone()
|
|
if not row:
|
|
continue
|
|
course_pk = row["id"]
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO schoolhouse_assignment (course_id, d2l_assignment_key, title, grade_text, professor_comments, source, status, created_at, updated_at)
|
|
VALUES (%s, %s, %s, %s, %s, %s, %s, NOW(), NOW())
|
|
ON CONFLICT DO NOTHING
|
|
""",
|
|
(course_pk, item.get("id"), item.get("title"), item.get("grade"), item.get("comments"), item.get("source", "d2l"), item.get("status", "assigned")),
|
|
)
|
|
cur.execute(
|
|
"""
|
|
UPDATE schoolhouse_assignment
|
|
SET grade_text = %s,
|
|
professor_comments = %s,
|
|
status = COALESCE(status, %s),
|
|
updated_at = NOW()
|
|
WHERE course_id = %s AND title = %s
|
|
""",
|
|
(item.get("grade"), item.get("comments"), item.get("status", "assigned"), course_pk, item.get("title")),
|
|
)
|
|
conn.commit()
|
|
return items
|
|
|
|
def list_sync_runs(self) -> list[dict[str, Any]]:
|
|
with self._connect() as conn, conn.cursor() as cur:
|
|
cur.execute("SELECT id::text AS id, started_at, completed_at, status, raw_snapshot_path AS snapshot_path, changes_summary_json FROM schoolhouse_d2l_sync_run ORDER BY started_at DESC")
|
|
return list(cur.fetchall())
|
|
|
|
def save_sync_runs(self, items: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
if not items:
|
|
return items
|
|
last = items[-1]
|
|
with self._connect() as conn, conn.cursor() as cur:
|
|
cur.execute(
|
|
"INSERT INTO schoolhouse_d2l_sync_run (completed_at, status, raw_snapshot_path, changes_summary_json) VALUES (NOW(), %s, %s, %s::jsonb)",
|
|
(last.get("status", "completed"), last.get("snapshot_path"), json.dumps(last)),
|
|
)
|
|
conn.commit()
|
|
return items
|
|
|
|
def list_submissions(self) -> list[dict[str, Any]]:
|
|
with self._connect() as conn, conn.cursor() as cur:
|
|
cur.execute(
|
|
"""
|
|
SELECT s.id::text AS id,
|
|
a.id::text AS assignment_id,
|
|
c.d2l_course_id AS class_id,
|
|
c.name AS class_name,
|
|
a.title AS assignment_name,
|
|
s.version_label,
|
|
s.original_filename,
|
|
s.mime_type,
|
|
s.status,
|
|
s.paperless_document_id,
|
|
s.paperless_title,
|
|
s.paperless_note_last_hash,
|
|
s.submitted_at,
|
|
s.confirmed_metadata_json,
|
|
s.confirmed_metadata_json->>'batch_id' AS batch_id,
|
|
s.confirmed_metadata_json->>'file_count' AS file_count,
|
|
s.confirmed_metadata_json->>'file_index' AS file_index,
|
|
s.confirmed_metadata_json->>'submission_kind' AS submission_kind,
|
|
s.confirmed_metadata_json->>'artifact_kind' AS artifact_kind,
|
|
s.confirmed_metadata_json->>'course_code' AS course_code,
|
|
s.confirmed_metadata_json->>'paper_kind' AS paper_kind,
|
|
s.confirmed_metadata_json->>'source' AS source,
|
|
s.confirmed_metadata_json->>'notes' AS notes,
|
|
s.confirmed_metadata_json->>'received_at' AS received_at,
|
|
s.confirmed_metadata_json->>'bundle_path' AS bundle_path,
|
|
s.confirmed_metadata_json->>'bundle_filename' AS bundle_filename,
|
|
s.confirmed_metadata_json->>'source_listing_path' AS source_listing_path,
|
|
s.confirmed_metadata_json->>'receipt_pdf_path' AS receipt_pdf_path,
|
|
s.confirmed_metadata_json->>'paperless_task_id' AS paperless_task_id,
|
|
s.confirmed_metadata_json->'file_manifest_json' AS file_manifest_json,
|
|
s.confirmed_metadata_json->'stored_paths' AS stored_paths,
|
|
s.confirmed_metadata_json->>'stored_path' AS stored_path,
|
|
s.confirmed_metadata_json->>'local_file' AS local_file
|
|
FROM schoolhouse_submission s
|
|
JOIN schoolhouse_assignment a ON a.id = s.assignment_id
|
|
JOIN schoolhouse_course c ON c.id = a.course_id
|
|
ORDER BY s.created_at DESC
|
|
"""
|
|
)
|
|
return list(cur.fetchall())
|
|
|
|
def create_submission(self, item: dict[str, Any]) -> dict[str, Any]:
|
|
with self._connect() as conn, conn.cursor() as cur:
|
|
row = self._ensure_assignment(cur, item)
|
|
if not row:
|
|
raise ValueError("Assignment not found and could not be created in Postgres repository")
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO schoolhouse_submission (
|
|
assignment_id, version_label, original_filename, mime_type, checksum_sha256,
|
|
submitted_at, suggested_metadata_json, confirmed_metadata_json, status
|
|
) VALUES (%s, %s, %s, %s, %s, NOW(), %s::jsonb, %s::jsonb, %s)
|
|
RETURNING id::text AS id
|
|
""",
|
|
(
|
|
row["id"], item.get("version_label") or "v1", item.get("original_filename"), item.get("mime_type", "application/octet-stream"), item.get("checksum_sha256"),
|
|
json.dumps(item.get("paperless_handoff", {})), json.dumps(item), item.get("status", "uploaded-local"),
|
|
),
|
|
)
|
|
created = cur.fetchone()
|
|
conn.commit()
|
|
item["id"] = created["id"]
|
|
return item
|
|
|
|
def update_submission(self, submission_id: str, updates: dict[str, Any]) -> dict[str, Any] | None:
|
|
with self._connect() as conn, conn.cursor() as cur:
|
|
cur.execute("SELECT confirmed_metadata_json FROM schoolhouse_submission WHERE id = %s::bigint", (submission_id,))
|
|
current = cur.fetchone()
|
|
if not current:
|
|
return None
|
|
merged_metadata = {**(current.get("confirmed_metadata_json") or {}), **updates}
|
|
cur.execute(
|
|
"""
|
|
UPDATE schoolhouse_submission
|
|
SET status = COALESCE(%s, status),
|
|
paperless_document_id = COALESCE(%s, paperless_document_id),
|
|
paperless_title = COALESCE(%s, paperless_title),
|
|
paperless_note_last_hash = COALESCE(%s, paperless_note_last_hash),
|
|
paperless_note_appended_at = CASE WHEN %s::text IS NOT NULL THEN NOW() ELSE paperless_note_appended_at END,
|
|
confirmed_metadata_json = %s::jsonb,
|
|
updated_at = NOW()
|
|
WHERE id = %s::bigint
|
|
RETURNING id::text AS id, status, paperless_document_id, paperless_title, paperless_note_last_hash
|
|
""",
|
|
(
|
|
updates.get("status"), updates.get("paperless_document_id"), updates.get("paperless_title"), updates.get("paperless_note_last_hash"), updates.get("paperless_note_last_hash"),
|
|
json.dumps(merged_metadata), submission_id,
|
|
),
|
|
)
|
|
row = cur.fetchone()
|
|
conn.commit()
|
|
return {**merged_metadata, **dict(row)} if row else None
|
|
|
|
def list_recordings(self) -> list[dict[str, Any]]:
|
|
with self._connect() as conn, conn.cursor() as cur:
|
|
cur.execute(
|
|
"""
|
|
SELECT r.id::text AS id,
|
|
c.d2l_course_id AS class_id,
|
|
c.name AS class_name,
|
|
r.assignment_id::text AS assignment_id,
|
|
r.course_id::text AS course_id,
|
|
r.session_date::text AS session_date,
|
|
r.original_filename,
|
|
r.mp3_filename,
|
|
r.transcription_status,
|
|
r.summary_status,
|
|
r.recording_paperless_document_id,
|
|
r.recording_metadata_json,
|
|
r.recording_metadata_json->>'stored_path' AS stored_path,
|
|
r.recording_metadata_json->>'mp3_path' AS mp3_path,
|
|
r.recording_metadata_json->>'status' AS status,
|
|
r.recording_metadata_json->>'conversion_status' AS conversion_status,
|
|
r.recording_metadata_json->>'notes' AS notes,
|
|
r.recording_metadata_json->>'delete_wav_after_success' AS delete_wav_after_success
|
|
FROM schoolhouse_recording r
|
|
JOIN schoolhouse_course c ON c.id = r.course_id
|
|
ORDER BY r.created_at DESC
|
|
"""
|
|
)
|
|
return list(cur.fetchall())
|
|
|
|
def create_recording(self, item: dict[str, Any]) -> dict[str, Any]:
|
|
with self._connect() as conn, conn.cursor() as cur:
|
|
row = self._ensure_course(cur, item)
|
|
if not row:
|
|
raise ValueError("Course not found and could not be created in Postgres repository")
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO schoolhouse_recording (
|
|
course_id, assignment_id, session_date, original_filename, mp3_filename, notes,
|
|
transcription_status, summary_status, recording_metadata_json
|
|
) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s::jsonb)
|
|
RETURNING id::text AS id
|
|
""",
|
|
(
|
|
row["id"], item.get("assignment_id") or None, item.get("session_date"), item.get("original_filename"),
|
|
Path(item.get("mp3_path")).name if item.get("mp3_path") else None, item.get("notes"), item.get("transcription_status"),
|
|
item.get("summary_status"), json.dumps(item),
|
|
),
|
|
)
|
|
created = cur.fetchone()
|
|
conn.commit()
|
|
item["id"] = created["id"]
|
|
return item
|
|
|
|
def update_recording(self, recording_id: str, updates: dict[str, Any]) -> dict[str, Any] | None:
|
|
with self._connect() as conn, conn.cursor() as cur:
|
|
cur.execute("SELECT recording_metadata_json FROM schoolhouse_recording WHERE id = %s::bigint", (recording_id,))
|
|
current = cur.fetchone()
|
|
if not current:
|
|
return None
|
|
merged_metadata = {**(current.get("recording_metadata_json") or {}), **updates}
|
|
cur.execute(
|
|
"""
|
|
UPDATE schoolhouse_recording
|
|
SET transcription_status = COALESCE(%s, transcription_status),
|
|
summary_status = COALESCE(%s, summary_status),
|
|
transcript_paperless_document_id = COALESCE(%s, transcript_paperless_document_id),
|
|
summary_paperless_document_id = COALESCE(%s, summary_paperless_document_id),
|
|
recording_paperless_document_id = COALESCE(%s, recording_paperless_document_id),
|
|
mp3_filename = COALESCE(%s, mp3_filename),
|
|
notes = COALESCE(%s, notes),
|
|
recording_metadata_json = %s::jsonb,
|
|
updated_at = NOW()
|
|
WHERE id = %s::bigint
|
|
RETURNING id::text AS id, transcription_status, summary_status, recording_paperless_document_id
|
|
""",
|
|
(
|
|
updates.get("transcription_status"), updates.get("summary_status"), updates.get("transcript_paperless_document_id"),
|
|
updates.get("summary_paperless_document_id"), updates.get("recording_paperless_document_id"),
|
|
Path(updates.get("mp3_path")).name if updates.get("mp3_path") else None,
|
|
updates.get("notes"), json.dumps(merged_metadata), recording_id,
|
|
),
|
|
)
|
|
row = cur.fetchone()
|
|
conn.commit()
|
|
return {**merged_metadata, **dict(row)} if row else None
|
|
|
|
|
|
def get_repository():
|
|
backend = getattr(settings, "storage_backend", "auto")
|
|
if backend == "json":
|
|
return JsonRepository()
|
|
if settings.database_url.startswith("postgresql"):
|
|
try:
|
|
return PostgresRepository()
|
|
except Exception:
|
|
return JsonRepository()
|
|
return JsonRepository()
|