Spaces:
Running
Running
| """Teacher-only content administration for classroom plans and Demo projects.""" | |
| from __future__ import annotations | |
| import json | |
| import re | |
| import shutil | |
| from pathlib import Path, PurePosixPath | |
| from fastapi import APIRouter, HTTPException | |
| from fastapi.responses import FileResponse | |
| from pydantic import BaseModel | |
| import classroom_service as classroom | |
| ROOT = Path(__file__).resolve().parent | |
| PROJECTS = ROOT / "projects" | |
| STATIC = ROOT / "classroom_static" | |
| TEXT_EXTENSIONS = { | |
| ".py", ".html", ".css", ".js", ".json", ".txt", ".csv", ".md", | |
| ".toml", ".yaml", ".yml", | |
| } | |
| router = APIRouter() | |
| class AdminAuth(BaseModel): | |
| password: str | |
| class WeekRequest(AdminAuth): | |
| week_id: str | |
| class WeekSave(WeekRequest): | |
| title: str | |
| project: str | |
| output: str | |
| files: dict[str, str] | |
| class ProjectRef(BaseModel): | |
| class_name: str | |
| student_name: str | |
| project_name: str | |
| class DemoRequest(AdminAuth): | |
| project: ProjectRef | |
| class DemoSave(AdminAuth): | |
| original: ProjectRef | None = None | |
| class_name: str | |
| student_name: str | |
| project_name: str | |
| files: dict[str, str] | |
| class ClassroomSubmissionRequest(AdminAuth): | |
| class_code: str | |
| student_id: str | |
| week_id: str | |
| class ClassroomBackupRequest(AdminAuth): | |
| class_code: str | |
| week_id: str | |
| def segment(value: str, label: str) -> str: | |
| text = re.sub(r"[\x00-\x1f\x7f]", "", str(value).strip())[:100] | |
| if not text or text in {".", ".."} or "/" in text or "\\" in text: | |
| raise HTTPException(400, f"{label}不合法") | |
| return text | |
| def text_path(value: str) -> str: | |
| raw = str(value).replace("\\", "/").strip("/") | |
| path = PurePosixPath(raw) | |
| if ( | |
| not raw or path.is_absolute() or ".." in path.parts or len(raw) > 180 | |
| or path.suffix.lower() not in TEXT_EXTENSIONS | |
| ): | |
| raise HTTPException(400, f"檔案路徑不合法：{value}") | |
| return path.as_posix() | |
| def project_root(ref: ProjectRef) -> Path: | |
| return PROJECTS / segment(ref.class_name, "班級名稱") / segment(ref.student_name, "學生名稱") / segment(ref.project_name, "專題名稱") | |
| def relative_repo_path(path: Path) -> str: | |
| return path.relative_to(ROOT).as_posix() | |
| def commit_changes(additions: dict[str, bytes | Path], deletions: list[str], message: str): | |
| if not classroom.TOKEN: | |
| raise HTTPException(503, "請先在 Space Secrets 設定 HF_TOKEN") | |
| from huggingface_hub import CommitOperationAdd, CommitOperationDelete, HfApi | |
| add_paths = set(additions) | |
| operations = [ | |
| CommitOperationDelete(path_in_repo=path) | |
| for path in sorted(set(deletions) - add_paths) | |
| ] | |
| operations.extend( | |
| CommitOperationAdd(path_in_repo=path, path_or_fileobj=value) | |
| for path, value in sorted(additions.items()) | |
| ) | |
| if not operations: | |
| return None | |
| return HfApi(token=classroom.TOKEN).create_commit( | |
| repo_id=classroom.REPO, | |
| repo_type="space", | |
| operations=operations, | |
| commit_message=message, | |
| ) | |
| def load_course() -> dict: | |
| return json.loads((classroom.DATA / "course.json").read_text(encoding="utf-8")) | |
| def scan_demo() -> list[dict]: | |
| projects = [] | |
| if not PROJECTS.is_dir(): | |
| return projects | |
| for class_dir in sorted(path for path in PROJECTS.iterdir() if path.is_dir() and not path.name.startswith(".")): | |
| for student_dir in sorted(path for path in class_dir.iterdir() if path.is_dir() and not path.name.startswith(".")): | |
| for item in sorted(path for path in student_dir.iterdir() if path.is_dir() and not path.name.startswith(".")): | |
| kind = "python" if (item / "main.py").is_file() else "html" if (item / "index.html").is_file() else "external" if (item / "link.txt").is_file() else "asset" | |
| projects.append({ | |
| "class_name": class_dir.name, | |
| "student_name": student_dir.name, | |
| "project_name": item.name, | |
| "type": kind, | |
| }) | |
| return projects | |
| def read_editable_text(path: Path) -> str | None: | |
| if path.suffix.lower() not in TEXT_EXTENSIONS or path.stat().st_size > 800_000: | |
| return None | |
| try: | |
| return path.read_text(encoding="utf-8") | |
| except UnicodeDecodeError: | |
| return None | |
| def editable_project(root: Path) -> dict: | |
| if not root.is_dir(): | |
| raise HTTPException(404, "找不到這個 Demo 專題") | |
| files: dict[str, str] = {} | |
| assets: list[str] = [] | |
| for path in sorted(item for item in root.rglob("*") if item.is_file()): | |
| rel = path.relative_to(root).as_posix() | |
| content = read_editable_text(path) | |
| if content is None: | |
| assets.append(rel) | |
| else: | |
| files[rel] = content | |
| return {"files": files, "assets": assets} | |
| def admin_page(): | |
| return FileResponse(STATIC / "admin.html") | |
| def admin_state(req: AdminAuth): | |
| classroom.check_password(req.password) | |
| return {"course": load_course(), "demo_projects": scan_demo(), "classrooms": {"courses": classroom.catalog()["courses"], "submissions": classroom.submission_catalog()}} | |
| def admin_submission(req: ClassroomSubmissionRequest): | |
| classroom.check_password(req.password) | |
| state = classroom.get_submission(req.class_code, req.student_id, req.week_id) | |
| return {"submission": classroom.submission_payload(state)} | |
| def backup_submissions(req: ClassroomBackupRequest): | |
| classroom.check_password(req.password) | |
| class_code = classroom.clean(req.class_code, "") | |
| week_id = classroom.clean(req.week_id, "") | |
| states = classroom.finals(class_code, week_id) | |
| if not states: | |
| raise HTTPException(400, "這個 Lesson 還沒有學生提交") | |
| additions = {} | |
| for state in states: | |
| path = classroom.submission_path(class_code, state["student_id"], week_id) | |
| content = (json.dumps(classroom.submission_payload(state), ensure_ascii=False, indent=2) + "\n").encode() | |
| additions[relative_repo_path(path)] = content | |
| result = commit_changes(additions, [], f"Backup {class_code} {week_id} classroom submissions") | |
| for repo_path, content in additions.items(): | |
| path = ROOT / repo_path | |
| path.parent.mkdir(parents=True, exist_ok=True) | |
| path.write_bytes(content) | |
| return {"ok": True, "count": len(states), "commit_url": str(result.commit_url) if result else ""} | |
| def delete_submission(req: ClassroomSubmissionRequest): | |
| classroom.check_password(req.password) | |
| state = classroom.get_submission(req.class_code, req.student_id, req.week_id) | |
| path = classroom.submission_path(state["class_code"], state["student_id"], state["week_id"]) | |
| result = commit_changes({}, [relative_repo_path(path)] if path.is_file() else [], f"Delete {state['class_code']} {state['week_id']} submission for {state['display_name']}") | |
| with classroom.lock, classroom.sqlite3.connect(classroom.DB) as db: | |
| db.execute("UPDATE classroom_projects SET final_files=NULL,submitted_at=NULL WHERE class_code=? AND student_id=? AND week_id=?",(state["class_code"],state["student_id"],state["week_id"])) | |
| path.unlink(missing_ok=True) | |
| return {"ok": True, "commit_url": str(result.commit_url) if result else ""} | |
| def admin_week(req: WeekRequest): | |
| classroom.check_password(req.password) | |
| week_id = classroom.clean(req.week_id, "") | |
| course = load_course() | |
| week = next((item for item in course.get("weeks", []) if item.get("id") == week_id), None) | |
| if not week: | |
| raise HTTPException(404, "找不到這個 Lesson") | |
| root = classroom.TEMPLATES / week_id | |
| files = {} | |
| if root.is_dir(): | |
| for path in sorted(item for item in root.rglob("*") if item.is_file()): | |
| rel = path.relative_to(root).as_posix() | |
| if path.suffix.lower() in TEXT_EXTENSIONS: | |
| files[rel] = path.read_text(encoding="utf-8") | |
| return {"week": week, "files": files} | |
| def save_week(req: WeekSave): | |
| classroom.check_password(req.password) | |
| week_id = classroom.clean(req.week_id, "") | |
| files = {text_path(path): str(content) for path, content in req.files.items()} | |
| course = load_course() | |
| week = next((item for item in course.get("weeks", []) if item.get("id") == week_id), None) | |
| if not week: | |
| raise HTTPException(404, "找不到這個 Lesson") | |
| week.update({ | |
| "title": classroom.clean_label(req.title, week.get("title", "")), | |
| "project": classroom.clean_label(req.project, week.get("project", "")), | |
| "output": classroom.clean_label(req.output, week.get("output", ""), 160), | |
| "template": bool(files), | |
| }) | |
| course_bytes = (json.dumps(course, ensure_ascii=False, indent=2) + "\n").encode() | |
| root = classroom.TEMPLATES / week_id | |
| old = [path for path in root.rglob("*") if path.is_file()] if root.is_dir() else [] | |
| additions = {"classroom_data/course.json": course_bytes} | |
| additions.update({f"classroom_data/templates/{week_id}/{path}": content.encode() for path, content in files.items()}) | |
| deletions = [relative_repo_path(path) for path in old if path.relative_to(root).as_posix() not in files] | |
| result = commit_changes(additions, deletions, f"Update classroom {week_id}") | |
| (classroom.DATA / "course.json").write_bytes(course_bytes) | |
| root.mkdir(parents=True, exist_ok=True) | |
| for path in old: | |
| if path.relative_to(root).as_posix() not in files: | |
| path.unlink(missing_ok=True) | |
| for rel, content in files.items(): | |
| target = root / rel | |
| target.parent.mkdir(parents=True, exist_ok=True) | |
| target.write_text(content, encoding="utf-8") | |
| return {"ok": True, "commit_url": str(result.commit_url) if result else ""} | |
| def admin_demo_project(req: DemoRequest): | |
| classroom.check_password(req.password) | |
| return editable_project(project_root(req.project)) | |
| def save_demo(req: DemoSave): | |
| classroom.check_password(req.password) | |
| target_ref = ProjectRef( | |
| class_name=segment(req.class_name, "班級名稱"), | |
| student_name=segment(req.student_name, "學生名稱"), | |
| project_name=segment(req.project_name, "專題名稱"), | |
| ) | |
| target = project_root(target_ref) | |
| original = project_root(req.original) if req.original else None | |
| if original and not original.is_dir(): | |
| raise HTTPException(404, "原始 Demo 專題不存在") | |
| if target != original and target.exists(): | |
| raise HTTPException(409, "目標位置已有同名專題") | |
| files = {text_path(path): str(content) for path, content in req.files.items()} | |
| if not any(name in files for name in ("main.py", "index.html", "link.txt")): | |
| raise HTTPException(400, "專題至少需要 main.py、index.html 或 link.txt 其中一個入口") | |
| existing = [path for path in original.rglob("*") if path.is_file()] if original else [] | |
| additions: dict[str, bytes | Path] = {} | |
| target_prefix = relative_repo_path(target) | |
| if target != original: | |
| for path in existing: | |
| rel = path.relative_to(original).as_posix() | |
| if read_editable_text(path) is None: | |
| additions[f"{target_prefix}/{rel}"] = path | |
| additions.update({f"{target_prefix}/{rel}": content.encode() for rel, content in files.items()}) | |
| if target == original: | |
| deletions = [relative_repo_path(path) for path in existing if read_editable_text(path) is not None and path.relative_to(original).as_posix() not in files] | |
| else: | |
| deletions = [relative_repo_path(path) for path in existing] | |
| result = commit_changes(additions, deletions, f"Update Demo project {target_ref.project_name}") | |
| if original and target != original: | |
| target.parent.mkdir(parents=True, exist_ok=True) | |
| shutil.copytree(original, target) | |
| shutil.rmtree(original) | |
| else: | |
| target.mkdir(parents=True, exist_ok=True) | |
| for path in list(target.rglob("*")): | |
| if path.is_file() and read_editable_text(path) is not None and path.relative_to(target).as_posix() not in files: | |
| path.unlink() | |
| for rel, content in files.items(): | |
| path = target / rel | |
| path.parent.mkdir(parents=True, exist_ok=True) | |
| path.write_text(content, encoding="utf-8") | |
| project_data = target_ref.model_dump() if hasattr(target_ref, "model_dump") else target_ref.dict() | |
| return {"ok": True, "project": project_data, "commit_url": str(result.commit_url) if result else ""} | |
| def delete_demo(req: DemoRequest): | |
| classroom.check_password(req.password) | |
| root = project_root(req.project) | |
| if not root.is_dir(): | |
| raise HTTPException(404, "找不到這個 Demo 專題") | |
| paths = [path for path in root.rglob("*") if path.is_file()] | |
| result = commit_changes({}, [relative_repo_path(path) for path in paths], f"Delete Demo project {root.name}") | |
| shutil.rmtree(root) | |
| return {"ok": True, "commit_url": str(result.commit_url) if result else ""} | |