151 lines
6.1 KiB
Python
151 lines
6.1 KiB
Python
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from app.config import settings
|
|
from app.services.repository import get_repository
|
|
from app.services.store import JsonStore
|
|
|
|
|
|
class D2LService:
|
|
def __init__(self) -> None:
|
|
self.store = JsonStore()
|
|
self.repo = get_repository()
|
|
|
|
def _snapshot_path(self) -> Path:
|
|
candidate_dirs = [
|
|
Path(settings.d2l_scraper_workdir),
|
|
Path('/home/fizzlepoof/.openclaw/workspace/school'),
|
|
]
|
|
for workdir in candidate_dirs:
|
|
for name in ('d2l_current_snapshot.json', 'd2l_grades.json', 'd2l_scraper_output.json'):
|
|
candidate = workdir / name
|
|
if candidate.exists():
|
|
return candidate
|
|
return candidate_dirs[0] / 'd2l_current_snapshot.json'
|
|
|
|
def load_snapshot(self) -> dict[str, Any]:
|
|
path = self._snapshot_path()
|
|
if not path.exists():
|
|
return {"courses": [], "source_path": str(path), "missing": True}
|
|
data = json.loads(path.read_text())
|
|
if isinstance(data, dict):
|
|
data["source_path"] = str(path)
|
|
return data
|
|
return {"courses": [], "source_path": str(path), "missing": True}
|
|
|
|
def _normalize_courses(self, snapshot: dict[str, Any]) -> list[dict[str, Any]]:
|
|
items = []
|
|
for course in snapshot.get("courses", []):
|
|
items.append({
|
|
"id": str(course.get("id")),
|
|
"d2l_course_id": str(course.get("id")),
|
|
"name": course.get("name"),
|
|
"code": course.get("code"),
|
|
"finalSummary": course.get("finalSummary"),
|
|
"gradeItemCount": course.get("gradeItemCount"),
|
|
"synced_at": self.store.now(),
|
|
})
|
|
return items
|
|
|
|
def _normalize_assignments(self, snapshot: dict[str, Any]) -> list[dict[str, Any]]:
|
|
items = []
|
|
for course in snapshot.get("courses", []):
|
|
course_id = str(course.get("id"))
|
|
for grade_item in course.get("gradeItems", []):
|
|
if grade_item.get("isCategory"):
|
|
continue
|
|
title = grade_item.get("name") or "Untitled"
|
|
items.append({
|
|
"id": f"{course_id}:{title.lower().strip()}",
|
|
"course_id": course_id,
|
|
"course_name": course.get("name"),
|
|
"title": title,
|
|
"points": grade_item.get("points"),
|
|
"grade": grade_item.get("grade"),
|
|
"comments": grade_item.get("comments"),
|
|
"source": "d2l-grade-item",
|
|
"synced_at": self.store.now(),
|
|
})
|
|
return items
|
|
|
|
def sync_to_store(self) -> dict[str, Any]:
|
|
snapshot = self.load_snapshot()
|
|
courses = self._normalize_courses(snapshot)
|
|
assignments = self._normalize_assignments(snapshot)
|
|
existing_assignments = {item.get('id'): item for item in self.repo.list_assignments()}
|
|
for item in assignments:
|
|
prior = existing_assignments.get(item['id']) or {}
|
|
item['status'] = prior.get('status', 'assigned')
|
|
item['updated_at'] = self.store.now()
|
|
item['created_at'] = prior.get('created_at', self.store.now())
|
|
self.repo.replace_courses(courses)
|
|
self.repo.replace_assignments(assignments)
|
|
sync_runs = self.repo.list_sync_runs()
|
|
sync_runs.append({
|
|
"id": self.store.new_id("sync"),
|
|
"snapshot_path": snapshot.get("source_path"),
|
|
"course_count": len(courses),
|
|
"assignment_count": len(assignments),
|
|
"completed_at": self.store.now(),
|
|
"status": "completed",
|
|
})
|
|
self.repo.save_sync_runs(sync_runs)
|
|
grade_note_results = []
|
|
try:
|
|
from app.services.paperless import PaperlessService
|
|
paperless = PaperlessService()
|
|
for item in assignments:
|
|
result = paperless.append_grade_note_if_available(str(item.get("id")))
|
|
if result.get("status") != "skipped":
|
|
grade_note_results.append(result)
|
|
except Exception as exc:
|
|
grade_note_results.append({"status": "error", "reason": str(exc)})
|
|
return {
|
|
"status": "ok",
|
|
"snapshot_path": snapshot.get("source_path"),
|
|
"course_count": len(courses),
|
|
"assignment_count": len(assignments),
|
|
"backend": self.repo.backend,
|
|
"grade_note_results": grade_note_results,
|
|
}
|
|
|
|
def get_courses(self) -> list[dict[str, Any]]:
|
|
items = self.repo.list_courses()
|
|
if items:
|
|
return items
|
|
self.sync_to_store()
|
|
return self.repo.list_courses()
|
|
|
|
def get_assignments(self, course_id: str) -> list[dict[str, Any]]:
|
|
items = self.repo.list_assignments()
|
|
if not items:
|
|
self.sync_to_store()
|
|
items = self.repo.list_assignments()
|
|
return [item for item in items if str(item.get("course_id")) == str(course_id)]
|
|
|
|
def get_grades(self, course_id: str) -> dict[str, Any]:
|
|
snapshot = self.load_snapshot()
|
|
for course in snapshot.get("courses", []):
|
|
if str(course.get("id")) == str(course_id):
|
|
return {
|
|
"course_id": str(course.get("id")),
|
|
"course_name": course.get("name"),
|
|
"final": course.get("final"),
|
|
"finalSummary": course.get("finalSummary"),
|
|
"gradeItems": course.get("gradeItems", []),
|
|
}
|
|
return {"course_id": course_id, "gradeItems": []}
|
|
|
|
def describe_sync_plan(self) -> dict:
|
|
snapshot = self.load_snapshot()
|
|
return {
|
|
"status": "stub",
|
|
"backend": self.repo.backend,
|
|
"scraper_workdir": settings.d2l_scraper_workdir,
|
|
"scraper_command": settings.d2l_scraper_command,
|
|
"snapshot_path": snapshot.get("source_path"),
|
|
"course_count": len(snapshot.get("courses", [])),
|
|
"message": "Run existing D2L scraper, normalize courses/assignments/grades, then store into Schoolhouse persistence.",
|
|
}
|