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)