| import json |
| import threading |
| import uuid |
| from copy import deepcopy |
| from datetime import datetime, timedelta |
| from pathlib import Path |
| from typing import Any |
| from zoneinfo import ZoneInfo |
|
|
| TZ = ZoneInfo("Asia/Shanghai") |
| DEFAULT_SCHEDULE_SETTINGS = { |
| "semester_start": "2026-03-09", |
| "day_start": "08:15", |
| "day_end": "23:00", |
| "default_task_duration_minutes": 45, |
| } |
| COURSE_COLOR_PALETTE = [ |
| "#64d7f5", |
| "#7ee082", |
| "#ffbf69", |
| "#ff8a80", |
| "#b197fc", |
| "#83c5be", |
| ] |
| DEFAULT_COURSE_SEED_VERSION = "2026-spring-v1" |
| DEFAULT_PERIOD_TIMES = { |
| 1: ("08:15", "09:00"), |
| 2: ("09:10", "09:55"), |
| 3: ("10:15", "11:00"), |
| 4: ("11:10", "11:55"), |
| 5: ("13:50", "14:35"), |
| 6: ("14:45", "15:30"), |
| 7: ("15:40", "16:25"), |
| 8: ("16:45", "17:30"), |
| 9: ("17:40", "18:25"), |
| 10: ("19:20", "20:05"), |
| 11: ("20:15", "21:00"), |
| 12: ("21:10", "21:55"), |
| } |
|
|
|
|
| def build_default_time_slots() -> list[dict[str, str]]: |
| slots: list[dict[str, str]] = [] |
| for index, (start, end) in DEFAULT_PERIOD_TIMES.items(): |
| slots.append( |
| { |
| "label": f"第{index:02d}节课", |
| "start": start, |
| "end": end, |
| } |
| ) |
| return slots |
|
|
|
|
| DEFAULT_SCHEDULE_SETTINGS["time_slots"] = build_default_time_slots() |
| DEFAULT_IMPORTED_COURSES = [ |
| { |
| "title": "形势与政策-2_20", |
| "day_of_week": 7, |
| "start_period": 10, |
| "end_period": 11, |
| "start_week": 9, |
| "end_week": 15, |
| "week_pattern": "odd", |
| "location": "江安综合楼C座C407", |
| "color": "#E0B100", |
| }, |
| { |
| "title": "数字逻辑：应用与设计_09", |
| "day_of_week": 1, |
| "start_period": 5, |
| "end_period": 7, |
| "start_week": 1, |
| "end_week": 16, |
| "week_pattern": "all", |
| "location": "江安一教A座A412", |
| "color": "#FF5AA5", |
| }, |
| { |
| "title": "中国近现代史纲要_59", |
| "day_of_week": 1, |
| "start_period": 10, |
| "end_period": 12, |
| "start_week": 1, |
| "end_week": 16, |
| "week_pattern": "all", |
| "location": "江安综合楼C座C403", |
| "color": "#66BB6A", |
| }, |
| { |
| "title": "人工智能导论_666", |
| "day_of_week": 2, |
| "start_period": 1, |
| "end_period": 2, |
| "start_week": 1, |
| "end_week": 16, |
| "week_pattern": "all", |
| "location": "江安一教B座B201", |
| "color": "#C77400", |
| }, |
| { |
| "title": "微积分（I）-2_33", |
| "day_of_week": 2, |
| "start_period": 3, |
| "end_period": 4, |
| "start_week": 1, |
| "end_week": 16, |
| "week_pattern": "all", |
| "location": "江安一教B座B101", |
| "color": "#DD8E88", |
| }, |
| { |
| "title": "体育-2游泳_12", |
| "day_of_week": 2, |
| "start_period": 5, |
| "end_period": 6, |
| "start_week": 1, |
| "end_week": 12, |
| "week_pattern": "all", |
| "location": "江安未来游泳馆", |
| "color": "#717171", |
| }, |
| { |
| "title": "城市经济学_03", |
| "day_of_week": 2, |
| "start_period": 8, |
| "end_period": 9, |
| "start_week": 1, |
| "end_week": 16, |
| "week_pattern": "all", |
| "location": "江安一教A座A308", |
| "color": "#F6AD9A", |
| }, |
| { |
| "title": "新中国史_02", |
| "day_of_week": 2, |
| "start_period": 10, |
| "end_period": 12, |
| "start_week": 1, |
| "end_week": 11, |
| "week_pattern": "all", |
| "location": "江安综合楼C座C407", |
| "color": "#FF9A6A", |
| }, |
| { |
| "title": "通用英语 I-2_49", |
| "day_of_week": 3, |
| "start_period": 1, |
| "end_period": 2, |
| "start_week": 1, |
| "end_week": 16, |
| "week_pattern": "all", |
| "location": "江安二基楼B座B409", |
| "color": "#55C08D", |
| }, |
| { |
| "title": "线性代数（理工）_35", |
| "day_of_week": 3, |
| "start_period": 3, |
| "end_period": 4, |
| "start_week": 1, |
| "end_week": 16, |
| "week_pattern": "all", |
| "location": "江安综合楼C座C303", |
| "color": "#9A6EAB", |
| }, |
| { |
| "title": "微积分（I）-2_33", |
| "day_of_week": 4, |
| "start_period": 1, |
| "end_period": 3, |
| "start_week": 1, |
| "end_week": 16, |
| "week_pattern": "all", |
| "location": "江安一教B座B101", |
| "color": "#DD8E88", |
| }, |
| { |
| "title": "面向对象程序设计（Java篇）_03", |
| "day_of_week": 4, |
| "start_period": 5, |
| "end_period": 8, |
| "start_week": 1, |
| "end_week": 13, |
| "week_pattern": "all", |
| "location": "江安综合楼B座B205", |
| "color": "#C99B89", |
| }, |
| { |
| "title": "深度学习_01", |
| "day_of_week": 4, |
| "start_period": 10, |
| "end_period": 12, |
| "start_week": 6, |
| "end_week": 16, |
| "week_pattern": "all", |
| "location": "江安一教A座A207", |
| "color": "#FA8D92", |
| }, |
| { |
| "title": "大学物理（理工）III-1_09", |
| "day_of_week": 5, |
| "start_period": 1, |
| "end_period": 2, |
| "start_week": 1, |
| "end_week": 16, |
| "week_pattern": "all", |
| "location": "江安一教B座B401", |
| "color": "#8A860C", |
| }, |
| { |
| "title": "线性代数（理工）_35", |
| "day_of_week": 5, |
| "start_period": 3, |
| "end_period": 4, |
| "start_week": 1, |
| "end_week": 16, |
| "week_pattern": "all", |
| "location": "江安一教B座B301", |
| "color": "#9A6EAB", |
| }, |
| { |
| "title": "线性代数习题课_35", |
| "day_of_week": 6, |
| "start_period": 7, |
| "end_period": 8, |
| "start_week": 2, |
| "end_week": 16, |
| "week_pattern": "all", |
| "location": "江安一教B座B104", |
| "color": "#F0A794", |
| }, |
| ] |
|
|
|
|
| def beijing_now() -> datetime: |
| return datetime.now(TZ) |
|
|
|
|
| def iso_now() -> str: |
| return beijing_now().isoformat() |
|
|
|
|
| def make_id(prefix: str) -> str: |
| return f"{prefix}_{uuid.uuid4().hex[:10]}" |
|
|
|
|
| def build_default_courses() -> list[dict[str, Any]]: |
| created_at = iso_now() |
| courses: list[dict[str, Any]] = [] |
| for index, item in enumerate(DEFAULT_IMPORTED_COURSES): |
| start_time = DEFAULT_PERIOD_TIMES[item["start_period"]][0] |
| end_time = DEFAULT_PERIOD_TIMES[item["end_period"]][1] |
| courses.append( |
| { |
| "id": make_id("course"), |
| "title": item["title"], |
| "day_of_week": item["day_of_week"], |
| "start_time": start_time, |
| "end_time": end_time, |
| "start_week": item["start_week"], |
| "end_week": item["end_week"], |
| "week_pattern": item["week_pattern"], |
| "location": item["location"], |
| "color": item.get("color") or COURSE_COLOR_PALETTE[index % len(COURSE_COLOR_PALETTE)], |
| "created_at": created_at, |
| } |
| ) |
| return courses |
|
|
|
|
| class ReminderStore: |
| def __init__(self, path: Path): |
| self.path = path |
| self.lock = threading.Lock() |
| self.path.parent.mkdir(parents=True, exist_ok=True) |
| if not self.path.exists(): |
| self._write(self._seed_data()) |
| else: |
| self._ensure_schema() |
|
|
| def _seed_data(self) -> dict[str, Any]: |
| now = beijing_now().replace(minute=0, second=0, microsecond=0) |
|
|
| def task(title: str, hours_from_now: int) -> dict[str, Any]: |
| created = now - timedelta(hours=max(hours_from_now - 4, 1)) |
| due = now + timedelta(hours=hours_from_now) |
| return { |
| "id": make_id("task"), |
| "title": title, |
| "created_at": created.isoformat(), |
| "due_at": due.isoformat(), |
| "completed": False, |
| "completed_at": None, |
| "schedule": None, |
| } |
|
|
| return { |
| "categories": [ |
| { |
| "id": make_id("cat"), |
| "name": "今日节奏", |
| "created_at": iso_now(), |
| "tasks": [ |
| task("整理今天的重点任务", 10), |
| task("晚间复盘 15 分钟", 14), |
| ], |
| }, |
| { |
| "id": make_id("cat"), |
| "name": "学习推进", |
| "created_at": iso_now(), |
| "tasks": [ |
| task("完成一节课程并记笔记", 26), |
| ], |
| }, |
| { |
| "id": make_id("cat"), |
| "name": "生活安排", |
| "created_at": iso_now(), |
| "tasks": [ |
| task("补充下周需要采购的清单", 40), |
| ], |
| }, |
| ], |
| "courses": build_default_courses(), |
| "course_seed_version": DEFAULT_COURSE_SEED_VERSION, |
| "schedule_settings": deepcopy(DEFAULT_SCHEDULE_SETTINGS), |
| } |
|
|
| def _normalize_data(self, data: dict[str, Any]) -> tuple[dict[str, Any], bool]: |
| changed = False |
|
|
| categories = data.setdefault("categories", []) |
| courses = data.setdefault("courses", []) |
| settings = data.setdefault("schedule_settings", {}) |
| course_seed_version = data.get("course_seed_version") |
|
|
| for key, value in DEFAULT_SCHEDULE_SETTINGS.items(): |
| if key not in settings: |
| settings[key] = deepcopy(value) |
| changed = True |
|
|
| time_slots = settings.get("time_slots") |
| if not isinstance(time_slots, list) or not time_slots: |
| settings["time_slots"] = build_default_time_slots() |
| changed = True |
| else: |
| normalized_time_slots = [] |
| for index, slot in enumerate(time_slots, start=1): |
| normalized_time_slots.append( |
| { |
| "label": str(slot.get("label", "")).strip() or f"第{index:02d}节课", |
| "start": str(slot.get("start", DEFAULT_PERIOD_TIMES.get(index, ("08:15", "09:00"))[0])), |
| "end": str(slot.get("end", DEFAULT_PERIOD_TIMES.get(index, ("08:15", "09:00"))[1])), |
| } |
| ) |
| if normalized_time_slots != time_slots: |
| settings["time_slots"] = normalized_time_slots |
| changed = True |
|
|
| if course_seed_version != DEFAULT_COURSE_SEED_VERSION: |
| if not courses: |
| data["courses"] = build_default_courses() |
| courses = data["courses"] |
| data["course_seed_version"] = DEFAULT_COURSE_SEED_VERSION |
| changed = True |
|
|
| for category in categories: |
| if "created_at" not in category: |
| category["created_at"] = iso_now() |
| changed = True |
| tasks = category.setdefault("tasks", []) |
| for task in tasks: |
| if "completed" not in task: |
| task["completed"] = False |
| changed = True |
| if "completed_at" not in task: |
| task["completed_at"] = None |
| changed = True |
| if "schedule" not in task: |
| task["schedule"] = None |
| changed = True |
|
|
| for index, course in enumerate(courses): |
| if "created_at" not in course: |
| course["created_at"] = iso_now() |
| changed = True |
| if "week_pattern" not in course: |
| course["week_pattern"] = "all" |
| changed = True |
| if "location" not in course: |
| course["location"] = "" |
| changed = True |
| if "color" not in course: |
| course["color"] = COURSE_COLOR_PALETTE[index % len(COURSE_COLOR_PALETTE)] |
| changed = True |
|
|
| return data, changed |
|
|
| def _ensure_schema(self) -> None: |
| with self.lock: |
| data = self._read() |
| normalized, changed = self._normalize_data(data) |
| if changed: |
| self._write(normalized) |
|
|
| def _read(self) -> dict[str, Any]: |
| with self.path.open("r", encoding="utf-8") as file: |
| return json.load(file) |
|
|
| def _write(self, data: dict[str, Any]) -> None: |
| temp_path = self.path.with_suffix(".tmp") |
| with temp_path.open("w", encoding="utf-8") as file: |
| json.dump(data, file, ensure_ascii=False, indent=2) |
| temp_path.replace(self.path) |
|
|
| def snapshot(self) -> dict[str, Any]: |
| with self.lock: |
| data = self._read() |
| normalized, changed = self._normalize_data(data) |
| if changed: |
| self._write(normalized) |
| return deepcopy(normalized) |
|
|
| def list_categories(self) -> list[dict[str, Any]]: |
| data = self.snapshot() |
| categories = data.get("categories", []) |
| for category in categories: |
| category["tasks"] = sorted( |
| category.get("tasks", []), |
| key=lambda item: ( |
| item.get("completed", False), |
| item.get("due_at", ""), |
| item.get("created_at", ""), |
| ), |
| ) |
| return categories |
|
|
| def list_tasks(self) -> list[dict[str, Any]]: |
| data = self.snapshot() |
| flattened: list[dict[str, Any]] = [] |
| for category in data.get("categories", []): |
| for task in category.get("tasks", []): |
| task_copy = deepcopy(task) |
| task_copy["category_id"] = category["id"] |
| task_copy["category_name"] = category["name"] |
| flattened.append(task_copy) |
| return sorted( |
| flattened, |
| key=lambda task: ( |
| task.get("completed", False), |
| task.get("due_at", ""), |
| task.get("created_at", ""), |
| ), |
| ) |
|
|
| def list_courses(self) -> list[dict[str, Any]]: |
| data = self.snapshot() |
| return sorted( |
| data.get("courses", []), |
| key=lambda course: ( |
| course.get("day_of_week", 1), |
| course.get("start_time", ""), |
| course.get("start_week", 1), |
| ), |
| ) |
|
|
| def get_schedule_settings(self) -> dict[str, Any]: |
| data = self.snapshot() |
| return deepcopy(data.get("schedule_settings", DEFAULT_SCHEDULE_SETTINGS)) |
|
|
| def update_schedule_settings(self, payload: dict[str, Any]) -> dict[str, Any]: |
| with self.lock: |
| data = self._read() |
| settings = data.setdefault("schedule_settings", deepcopy(DEFAULT_SCHEDULE_SETTINGS)) |
| for key, value in payload.items(): |
| settings[key] = value |
| self._write(data) |
| return deepcopy(settings) |
|
|
| def create_category(self, name: str) -> dict[str, Any]: |
| with self.lock: |
| data = self._read() |
| category = { |
| "id": make_id("cat"), |
| "name": name, |
| "created_at": iso_now(), |
| "tasks": [], |
| } |
| data.setdefault("categories", []).append(category) |
| self._write(data) |
| return deepcopy(category) |
|
|
| def rename_category(self, category_id: str, name: str) -> dict[str, Any]: |
| with self.lock: |
| data = self._read() |
| for category in data.get("categories", []): |
| if category["id"] == category_id: |
| category["name"] = name |
| self._write(data) |
| return deepcopy(category) |
| raise KeyError("Category not found") |
|
|
| def delete_category(self, category_id: str) -> None: |
| with self.lock: |
| data = self._read() |
| original_count = len(data.get("categories", [])) |
| data["categories"] = [ |
| category |
| for category in data.get("categories", []) |
| if category["id"] != category_id |
| ] |
| if len(data["categories"]) == original_count: |
| raise KeyError("Category not found") |
| self._write(data) |
|
|
| def add_task(self, category_id: str, title: str, due_at: str) -> dict[str, Any]: |
| with self.lock: |
| data = self._read() |
| for category in data.get("categories", []): |
| if category["id"] == category_id: |
| item = { |
| "id": make_id("task"), |
| "title": title, |
| "created_at": iso_now(), |
| "due_at": due_at, |
| "completed": False, |
| "completed_at": None, |
| "schedule": None, |
| } |
| category.setdefault("tasks", []).append(item) |
| self._write(data) |
| return deepcopy(item) |
| raise KeyError("Category not found") |
|
|
| def toggle_task(self, task_id: str, completed: bool) -> dict[str, Any]: |
| with self.lock: |
| data = self._read() |
| for category in data.get("categories", []): |
| for task in category.get("tasks", []): |
| if task["id"] == task_id: |
| task["completed"] = completed |
| task["completed_at"] = iso_now() if completed else None |
| self._write(data) |
| return deepcopy(task) |
| raise KeyError("Task not found") |
|
|
| def delete_task(self, task_id: str) -> None: |
| with self.lock: |
| data = self._read() |
| for category in data.get("categories", []): |
| original_count = len(category.get("tasks", [])) |
| category["tasks"] = [ |
| task for task in category.get("tasks", []) if task["id"] != task_id |
| ] |
| if len(category["tasks"]) != original_count: |
| self._write(data) |
| return |
| raise KeyError("Task not found") |
|
|
| def schedule_task(self, task_id: str, schedule: dict[str, Any] | None) -> dict[str, Any]: |
| with self.lock: |
| data = self._read() |
| for category in data.get("categories", []): |
| for task in category.get("tasks", []): |
| if task["id"] == task_id: |
| task["schedule"] = deepcopy(schedule) if schedule else None |
| self._write(data) |
| task_copy = deepcopy(task) |
| task_copy["category_id"] = category["id"] |
| task_copy["category_name"] = category["name"] |
| return task_copy |
| raise KeyError("Task not found") |
|
|
| def create_course(self, payload: dict[str, Any]) -> dict[str, Any]: |
| with self.lock: |
| data = self._read() |
| courses = data.setdefault("courses", []) |
| course = { |
| "id": make_id("course"), |
| "title": payload["title"], |
| "day_of_week": payload["day_of_week"], |
| "start_time": payload["start_time"], |
| "end_time": payload["end_time"], |
| "start_week": payload["start_week"], |
| "end_week": payload["end_week"], |
| "week_pattern": payload.get("week_pattern", "all"), |
| "location": payload.get("location", ""), |
| "color": payload.get( |
| "color", |
| COURSE_COLOR_PALETTE[len(courses) % len(COURSE_COLOR_PALETTE)], |
| ), |
| "created_at": iso_now(), |
| } |
| courses.append(course) |
| self._write(data) |
| return deepcopy(course) |
|
|
| def update_course(self, course_id: str, payload: dict[str, Any]) -> dict[str, Any]: |
| with self.lock: |
| data = self._read() |
| for course in data.get("courses", []): |
| if course["id"] == course_id: |
| course.update(payload) |
| self._write(data) |
| return deepcopy(course) |
| raise KeyError("Course not found") |
|
|
| def delete_course(self, course_id: str) -> None: |
| with self.lock: |
| data = self._read() |
| original_count = len(data.get("courses", [])) |
| data["courses"] = [ |
| course |
| for course in data.get("courses", []) |
| if course["id"] != course_id |
| ] |
| if len(data["courses"]) == original_count: |
| raise KeyError("Course not found") |
| self._write(data) |
|
|
