from __future__ import annotations import json import os import tempfile import threading from calendar import monthrange from copy import deepcopy from datetime import date, datetime, timedelta from pathlib import Path from typing import Any from uuid import uuid4 from zoneinfo import ZoneInfo from .models import ChartInput, DailyRecordInput, ScheduleInput, TaskInput SCHEMA_VERSION = 1 SHANGHAI = ZoneInfo("Asia/Shanghai") class NotFoundError(Exception): pass class ConflictError(Exception): pass class ValidationError(Exception): pass def now_iso() -> str: return datetime.now(SHANGHAI).isoformat(timespec="seconds") def period_window(period: str, ref: date, task_start: date | None = None) -> tuple[date, date] | None: """Inclusive [start, end] of the统计周期 containing ``ref``. ``lifetime`` starts at the task's own start date so a task created today never claims credit for facts recorded before it existed. When ``ref`` predates the task start the interval is returned ``None`` so no pre-start fact can ever match it. Month/week bounds are computed without stepping outside Python's date domain (``date.min``/``date.max``), so extreme-but-valid ``ref`` values cannot raise ``OverflowError``. """ if period == "day": return ref, ref if period == "week": start = ref - timedelta(days=ref.weekday()) if start > date.max - timedelta(days=6): return start, date.max return start, start + timedelta(days=6) if period == "month": start = ref.replace(day=1) return start, ref.replace(day=monthrange(ref.year, ref.month)[1]) if period == "lifetime": start = task_start or date.min if start > ref: return None return start, ref raise ValidationError(f"未知周期 {period}") def _half_up(value: float) -> int: """Round a non-negative value half-up to match browser ``Math.round``.""" return int(value + 0.5) def _percent(value: int, target: int | None) -> int | None: if not target: return None return min(100, _half_up(value / target * 100)) class JsonRepository: """Single-process JSON repository with atomic same-directory replacement.""" def __init__(self, root: Path): self.root = Path(root) self.tasks_dir = self.root / "tasks" self.revisions_dir = self.root / "revisions" self.daily_dir = self.root / "daily" self.archive_dir = self.root / "archive" self.schedules_dir = self.root / "schedules" self.schedule_revisions_dir = self.root / "schedule-revisions" self.charts_dir = self.root / "charts" self.meta_path = self.root / "meta.json" self._lock = threading.RLock() for folder in (self.root, self.tasks_dir, self.revisions_dir, self.daily_dir, self.archive_dir, self.schedules_dir, self.schedule_revisions_dir, self.charts_dir): folder.mkdir(parents=True, exist_ok=True) if not self.meta_path.exists(): yesterday = date.today() - timedelta(days=1) self._atomic_write(self.meta_path, { "schemaVersion": SCHEMA_VERSION, "version": 1, "taskOrder": [], "lastClosedDate": yesterday.isoformat(), }) def _read(self, path: Path, default: Any = None) -> Any: if not path.exists(): return deepcopy(default) with path.open("r", encoding="utf-8") as handle: return json.load(handle) def _atomic_write(self, path: Path, value: Any) -> None: path.parent.mkdir(parents=True, exist_ok=True) fd, temp_name = tempfile.mkstemp(prefix=f".{path.name}.", suffix=".tmp", dir=path.parent) try: with os.fdopen(fd, "w", encoding="utf-8", newline="\n") as handle: json.dump(value, handle, ensure_ascii=False, indent=2) handle.write("\n") handle.flush() os.fsync(handle.fileno()) os.replace(temp_name, path) if os.name != "nt": dir_fd = os.open(path.parent, os.O_DIRECTORY) try: os.fsync(dir_fd) finally: os.close(dir_fd) finally: if os.path.exists(temp_name): os.unlink(temp_name) def _meta(self) -> dict[str, Any]: return self._read(self.meta_path) def _task_path(self, task_id: str) -> Path: return self.tasks_dir / f"{task_id}.json" def _schedule_path(self, schedule_id: str) -> Path: return self.schedules_dir / f"{schedule_id}.json" def _chart_path(self, chart_id: str) -> Path: return self.charts_dir / f"{chart_id}.json" def _load_chart(self, chart_id: str) -> dict[str, Any]: chart = self._read(self._chart_path(chart_id)) if chart is None: raise NotFoundError("图表不存在") return chart def create_chart(self, payload: ChartInput) -> dict[str, Any]: with self._lock: chart_id = str(uuid4()) timestamp = now_iso() chart = { "schemaVersion": SCHEMA_VERSION, "chartId": chart_id, "version": 1, **payload.model_dump(mode="json"), "createdAt": timestamp, "updatedAt": timestamp, } self._atomic_write(self._chart_path(chart_id), chart) return chart def list_charts(self) -> dict[str, Any]: with self._lock: items = [self._read(path) for path in self.charts_dir.glob("*.json")] items = [item for item in items if item] items.sort(key=lambda item: item.get("createdAt") or "") return {"items": items} def get_chart(self, chart_id: str) -> dict[str, Any]: with self._lock: return self._load_chart(chart_id) def update_chart(self, chart_id: str, payload: ChartInput, expected_version: int) -> dict[str, Any]: with self._lock: old = self._load_chart(chart_id) if old["version"] != expected_version: raise ConflictError("图表已被更新，请刷新后重试") chart = {**old, **payload.model_dump(mode="json"), "version": old["version"] + 1, "updatedAt": now_iso()} self._atomic_write(self._chart_path(chart_id), chart) return chart def delete_chart(self, chart_id: str) -> dict[str, Any]: with self._lock: self._load_chart(chart_id) self._chart_path(chart_id).unlink() return {"deleted": True, "chartId": chart_id} def chart_data(self, chart_id: str, ref_date: date | None = None) -> dict[str, Any]: with self._lock: chart = self._load_chart(chart_id) ref = ref_date or datetime.now(SHANGHAI).date() days = int(chart["days"]) start = ref - timedelta(days=days - 1) points = [] cursor = start while cursor <= ref: daily = self._read(self._daily_path(cursor), {}) or {} value = 0 records = daily.get("records", {}) if chart.get("metricId"): for task_id, record in records.items(): if chart.get("taskId") and task_id != chart["taskId"]: continue value += int(record.get("values", {}).get(chart["metricId"], 0) or 0) else: seen = set() for task_id, record in records.items(): if chart.get("taskId") and task_id != chart["taskId"]: continue if any(int(item or 0) > 0 for item in record.get("values", {}).values()): seen.add(task_id) value = len(seen) points.append({"date": cursor.isoformat(), "value": value}) cursor += timedelta(days=1) return {"chartId": chart["chartId"], "title": chart["title"], "days": days, "points": points} def review_notes(self, start: date, end: date) -> dict[str, Any]: with self._lock: if end < start or (end - start).days > 730: raise ValidationError("心得查询范围无效或超过 730 天") task_names = {} for path in self.tasks_dir.glob("*.json"): task = self._read(path) if task and not task.get("deletedAt"): task_names[task.get("taskId")] = task.get("name", "") days = [] cursor = end while cursor >= start: daily = self._read(self._daily_path(cursor), {}) or {} records = [] for task_id, record in daily.get("records", {}).items(): if not (record.get("note") or record.get("values") or record.get("completedAt")): continue records.append({ "taskId": task_id, "taskName": task_names.get(task_id, "已删除任务"), "note": record.get("note", ""), "completedAt": record.get("completedAt"), "values": record.get("values", {}), }) if records: days.append({"date": cursor.isoformat(), "records": records}) cursor -= timedelta(days=1) return {"start": start.isoformat(), "end": end.isoformat(), "days": days} def _load_schedule(self, schedule_id: str) -> dict[str, Any]: schedule = self._read(self._schedule_path(schedule_id)) if schedule is None: raise NotFoundError("日程不存在") return schedule def _write_schedule_revision(self, schedule: dict[str, Any]) -> None: path = self.schedule_revisions_dir / schedule["scheduleId"] / f"{schedule['revision']}.json" self._atomic_write(path, schedule) def create_schedule(self, payload: ScheduleInput) -> dict[str, Any]: with self._lock: schedule_id = str(uuid4()) timestamp = now_iso() schedule = { "schemaVersion": SCHEMA_VERSION, "scheduleId": schedule_id, "version": 1, "revision": 1, **payload.model_dump(mode="json"), "completions": {}, "createdAt": timestamp, "updatedAt": timestamp, "deletedAt": None, } self._atomic_write(self._schedule_path(schedule_id), schedule) self._write_schedule_revision(schedule) return schedule def list_schedules(self, include_deleted: bool = False) -> dict[str, Any]: with self._lock: schedules = [self._read(path) for path in self.schedules_dir.glob("*.json")] schedules = [item for item in schedules if item and (include_deleted or not item.get("deletedAt"))] schedules.sort(key=lambda item: (item.get("date") or "9999-12-31", item.get("time") or "", item.get("createdAt") or "")) return {"items": schedules} def get_schedule(self, schedule_id: str, include_deleted: bool = False) -> dict[str, Any]: with self._lock: schedule = self._load_schedule(schedule_id) if schedule.get("deletedAt") and not include_deleted: raise NotFoundError("日程不存在") return schedule def update_schedule(self, schedule_id: str, payload: ScheduleInput, expected_version: int) -> dict[str, Any]: with self._lock: old = self._load_schedule(schedule_id) if old.get("deletedAt"): raise NotFoundError("日程已删除") if old["version"] != expected_version: raise ConflictError("日程已被更新，请刷新后重试") schedule = { **old, **payload.model_dump(mode="json"), "version": old["version"] + 1, "revision": old["revision"] + 1, "updatedAt": now_iso(), } self._atomic_write(self._schedule_path(schedule_id), schedule) self._write_schedule_revision(schedule) return schedule def _change_schedule_deleted(self, schedule_id: str, expected_version: int, restore: bool) -> dict[str, Any]: schedule = self._load_schedule(schedule_id) if schedule["version"] != expected_version: raise ConflictError("日程已被更新，请刷新后重试") if restore and not schedule.get("deletedAt"): return schedule if not restore and schedule.get("deletedAt"): return schedule schedule["deletedAt"] = None if restore else now_iso() schedule["version"] += 1 schedule["revision"] += 1 schedule["updatedAt"] = now_iso() self._atomic_write(self._schedule_path(schedule_id), schedule) self._write_schedule_revision(schedule) return schedule def soft_delete_schedule(self, schedule_id: str, expected_version: int) -> dict[str, Any]: with self._lock: return self._change_schedule_deleted(schedule_id, expected_version, False) def restore_schedule(self, schedule_id: str, expected_version: int) -> dict[str, Any]: with self._lock: return self._change_schedule_deleted(schedule_id, expected_version, True) def purge_schedule(self, schedule_id: str, confirmation: str) -> None: with self._lock: schedule = self._load_schedule(schedule_id) if not schedule.get("deletedAt"): raise ValidationError("永久删除前必须先放入回收站") if confirmation != schedule_id: raise ValidationError("确认内容必须是 scheduleId") self._schedule_path(schedule_id).unlink() revision_folder = self.schedule_revisions_dir / schedule_id for path in revision_folder.glob("*.json"): path.unlink() if revision_folder.exists(): revision_folder.rmdir() def complete_schedule(self, schedule_id: str, logical_date: date, completed: bool, expected_version: int) -> dict[str, Any]: with self._lock: schedule = self._load_schedule(schedule_id) if schedule.get("deletedAt"): raise NotFoundError("日程已删除") if schedule["version"] != expected_version: raise ConflictError("日程已被更新，请刷新后重试") key = logical_date.isoformat() if completed: schedule.setdefault("completions", {})[key] = now_iso() else: schedule.setdefault("completions", {}).pop(key, None) schedule["version"] += 1 schedule["updatedAt"] = now_iso() self._atomic_write(self._schedule_path(schedule_id), schedule) return schedule def schedules_for_date(self, day: date) -> list[dict[str, Any]]: with self._lock: result = [] for schedule in self.list_schedules(include_deleted=False)["items"]: if schedule.get("state") != "active": continue start = date.fromisoformat(schedule["date"]) recurrence = schedule.get("recurrence", "none") occurs = ( recurrence == "none" and start == day or recurrence == "daily" and start <= day or recurrence == "weekly" and start <= day and start.weekday() == day.weekday() or recurrence == "monthly" and start <= day and start.day == day.day ) if not occurs: continue completion_time = schedule.get("completions", {}).get(day.isoformat()) result.append({**schedule, "completed": bool(completion_time), "completionTime": completion_time}) result.sort(key=lambda item: (item.get("time") or "", item.get("createdAt") or "")) return result def calendar_month(self, year: int, month: int) -> dict[str, Any]: with self._lock: days = [] for day_number in range(1, monthrange(year, month)[1] + 1): current = date(year, month, day_number) schedules = [item for item in self.schedules_for_date(current) if item.get("state") == "active"] days.append({"date": current.isoformat(), "schedules": schedules}) return {"year": year, "month": month, "days": days} def agent_overview(self, day: date) -> dict[str, Any]: schedules = [ {key: item.get(key) for key in ("scheduleId", "title", "description", "date", "time", "recurrence", "priority", "state", "completed", "completionTime")} for item in self.schedules_for_date(day) if item.get("state") == "active" ] progress = self.summarize(day) for task_id, info in progress.get("tasks", {}).items(): task = self._read(self._task_path(task_id)) if task: info["name"] = task.get("name", "") info["priority"] = task.get("priority") info["state"] = task.get("state", "") else: info["name"] = "" return {"date": day.isoformat(), "schedules": schedules, "taskProgress": progress} def _daily_path(self, logical_date: date) -> Path: return self.daily_dir / str(logical_date.year) / f"{logical_date.isoformat()}.json" def _load_task(self, task_id: str) -> dict[str, Any]: task = self._read(self._task_path(task_id)) if task is None: raise NotFoundError("任务不存在") return task def list_tasks(self, include_deleted: bool = False) -> dict[str, Any]: with self._lock: meta = self._meta() tasks = [] for path in self.tasks_dir.glob("*.json"): task = self._read(path) if include_deleted or not task.get("deletedAt"): tasks.append(task) positions = {task_id: index for index, task_id in enumerate(meta["taskOrder"])} tasks.sort(key=lambda item: (positions.get(item["taskId"], 10**9), item["createdAt"])) return {"version": meta["version"], "items": tasks} def get_task(self, task_id: str, include_deleted: bool = False) -> dict[str, Any]: with self._lock: task = self._load_task(task_id) if task.get("deletedAt") and not include_deleted: raise NotFoundError("任务不存在") return task def _metric_dicts(self, payload: TaskInput, old: dict[str, Any] | None = None) -> list[dict[str, Any]]: old_ids = {item["metricId"] for item in (old or {}).get("metrics", [])} result = [] seen = set() for metric in payload.metrics: metric_data = metric.model_dump(mode="json") metric_id = metric_data.get("metricId") if not metric_id: metric_id = str(uuid4()) elif old is not None and metric_id not in old_ids: raise ValidationError("不能指定不属于该任务的 metricId") if metric_id in seen: raise ValidationError("metricId 不能重复") seen.add(metric_id) metric_data["metricId"] = metric_id result.append(metric_data) return result def _write_revision(self, task: dict[str, Any]) -> None: path = self.revisions_dir / task["taskId"] / f"{task['revision']}.json" self._atomic_write(path, task) def create_task(self, payload: TaskInput) -> dict[str, Any]: with self._lock: task_id = str(uuid4()) timestamp = now_iso() task = { "schemaVersion": SCHEMA_VERSION, "taskId": task_id, "version": 1, "revision": 1, **payload.model_dump(mode="json", exclude={"metrics"}), "metrics": self._metric_dicts(payload), "createdAt": timestamp, "updatedAt": timestamp, "deletedAt": None, } self._atomic_write(self._task_path(task_id), task) self._write_revision(task) meta = self._meta() meta["taskOrder"].append(task_id) meta["version"] += 1 self._atomic_write(self.meta_path, meta) return task def update_task(self, task_id: str, payload: TaskInput, expected_version: int) -> dict[str, Any]: with self._lock: old = self._load_task(task_id) if old.get("deletedAt"): raise NotFoundError("任务已删除") if old["version"] != expected_version: raise ConflictError("任务已被更新，请刷新后重试") timestamp = now_iso() task = { **old, **payload.model_dump(mode="json", exclude={"metrics"}), "metrics": self._metric_dicts(payload, old), "version": old["version"] + 1, "revision": old["revision"] + 1, "updatedAt": timestamp, } self._atomic_write(self._task_path(task_id), task) self._write_revision(task) return task def _change_deleted(self, task_id: str, expected_version: int, restore: bool) -> dict[str, Any]: task = self._load_task(task_id) if task["version"] != expected_version: raise ConflictError("任务已被更新，请刷新后重试") if restore and not task.get("deletedAt"): return task if not restore and task.get("deletedAt"): return task task["deletedAt"] = None if restore else now_iso() task["version"] += 1 task["revision"] += 1 task["updatedAt"] = now_iso() self._atomic_write(self._task_path(task_id), task) self._write_revision(task) return task def soft_delete_task(self, task_id: str, expected_version: int) -> dict[str, Any]: with self._lock: return self._change_deleted(task_id, expected_version, False) def restore_task(self, task_id: str, expected_version: int) -> dict[str, Any]: with self._lock: return self._change_deleted(task_id, expected_version, True) def purge_task(self, task_id: str, confirmation: str) -> None: with self._lock: task = self._load_task(task_id) if not task.get("deletedAt"): raise ValidationError("永久删除前必须先放入回收站") if confirmation != task_id: raise ValidationError("确认内容必须是 taskId") self._task_path(task_id).unlink() revision_folder = self.revisions_dir / task_id for path in revision_folder.glob("*.json"): path.unlink() if revision_folder.exists(): revision_folder.rmdir() touched_months: set[tuple[int, int]] = set() for path in self.daily_dir.glob("*/*.json"): daily = self._read(path) if task_id in daily.get("records", {}): del daily["records"][task_id] daily["version"] += 1 self._atomic_write(path, daily) day = date.fromisoformat(daily["date"]) touched_months.add((day.year, day.month)) meta = self._meta() meta["taskOrder"] = [item for item in meta["taskOrder"] if item != task_id] meta["version"] += 1 self._atomic_write(self.meta_path, meta) for year, month in touched_months: self._rebuild_month(year, month) def reorder(self, task_ids: list[str], expected_version: int) -> dict[str, Any]: with self._lock: meta = self._meta() if meta["version"] != expected_version: raise ConflictError("排序已被更新，请刷新后重试") active_ids = {item["taskId"] for item in self.list_tasks()["items"]} if set(task_ids) != active_ids: raise ValidationError("排序列表必须包含全部未删除任务") deleted_ids = [item for item in meta["taskOrder"] if item not in active_ids] meta["taskOrder"] = task_ids + deleted_ids meta["version"] += 1 self._atomic_write(self.meta_path, meta) return {"version": meta["version"], "taskIds": task_ids} def get_daily(self, task_id: str, logical_date: date) -> dict[str, Any]: with self._lock: task = self._load_task(task_id) daily = self._read(self._daily_path(logical_date), { "schemaVersion": SCHEMA_VERSION, "date": logical_date.isoformat(), "version": 1, "closedAt": None, "records": {}, }) return daily["records"].get(task_id, { "taskId": task_id, "taskRevision": task["revision"], "version": 0, "note": "", "completedAt": None, "values": {}, }) def upsert_daily(self, task_id: str, logical_date: date, payload: DailyRecordInput) -> dict[str, Any]: with self._lock: task = self._load_task(task_id) if task.get("deletedAt"): raise NotFoundError("任务已删除") path = self._daily_path(logical_date) daily = self._read(path, { "schemaVersion": SCHEMA_VERSION, "date": logical_date.isoformat(), "version": 1, "closedAt": None, "records": {}, }) old = daily["records"].get(task_id) current_version = old["version"] if old else 0 if payload.expectedVersion is not None and payload.expectedVersion != current_version: raise ConflictError("当日记录已被更新，请刷新后重试") metric_map = {metric["metricId"]: metric for metric in task["metrics"]} values: dict[str, int] = {} for item in payload.values: metric = metric_map.get(item.metricId) if metric is None: raise ValidationError("记录包含未知指标") if metric["kind"] == "checkin" and item.value not in (0, 1): raise ValidationError("打卡指标只能是 0 或 1") values[item.metricId] = item.value record = { "taskId": task_id, "taskRevision": task["revision"], "version": current_version + 1, "note": payload.note, "completedAt": payload.completedAt.isoformat() if payload.completedAt else None, "values": values, "updatedAt": now_iso(), } daily["records"][task_id] = record daily["version"] += 1 self._atomic_write(path, daily) if logical_date < datetime.now(SHANGHAI).date(): if not daily.get("closedAt"): daily["closedAt"] = now_iso() self._atomic_write(path, daily) self._rebuild_month(logical_date.year, logical_date.month) return record def history(self, task_id: str, start: date, end: date) -> dict[str, Any]: with self._lock: self._load_task(task_id) if end < start or (end - start).days > 730: raise ValidationError("历史查询范围无效或超过 730 天") records = [] cursor = start while cursor <= end: daily = self._read(self._daily_path(cursor)) if daily and task_id in daily.get("records", {}): records.append({"date": cursor.isoformat(), **daily["records"][task_id]}) cursor += timedelta(days=1) revisions = [self._read(path) for path in sorted((self.revisions_dir / task_id).glob("*.json"), key=lambda p: int(p.stem))] return {"taskId": task_id, "records": records, "revisions": revisions} def _iter_daily(self, start: date, end: date): """Yield ``(day, daily)`` for every existing fact file inside [start, end].""" for year in range(start.year, end.year + 1): folder = self.daily_dir / str(year) if not folder.is_dir(): continue for path in sorted(folder.glob("*.json")): try: day = date.fromisoformat(path.stem) except ValueError: continue if start <= day <= end: daily = self._read(path) if daily: yield day, daily def summarize(self, ref_date: date | None = None, task_id: str | None = None) -> dict[str, Any]: """Per-metric progress inside each metric's own period window. Facts are the source of truth here; month archives stay a rebuildable derivative and are never read back for statistics. """ with self._lock: ref = ref_date or datetime.now(SHANGHAI).date() if task_id is None: tasks = [task for task in (self._read(path) for path in self.tasks_dir.glob("*.json")) if task and not task.get("deletedAt")] else: tasks = [self.get_task(task_id)] result: dict[str, Any] = {"date": ref.isoformat(), "tasks": {}} if not tasks: return result windows = { task["taskId"]: { metric["metricId"]: period_window( metric["period"], ref, date.fromisoformat(task["startDate"]) if task.get("startDate") else None, ) for metric in task["metrics"] } for task in tasks } spans = [span for per_task in windows.values() for span in per_task.values() if span is not None] totals: dict[str, dict[str, int]] = {task["taskId"]: {} for task in tasks} today_record: dict[str, dict[str, Any]] = {} if spans: for day, daily in self._iter_daily(min(span[0] for span in spans), ref): for record_task_id, record in daily.get("records", {}).items(): per_metric = windows.get(record_task_id) if per_metric is None: continue if day == ref: today_record[record_task_id] = record for metric_id, value in record.get("values", {}).items(): span = per_metric.get(metric_id) # Facts after ``ref`` never count, even though the # display window (span[1]) may extend past it. if span and span[0] <= day <= min(span[1], ref): totals[record_task_id][metric_id] = totals[record_task_id].get(metric_id, 0) + int(value) for task in tasks: current_id = task["taskId"] metrics = [] for metric in task["metrics"]: span = windows[current_id][metric["metricId"]] value = totals[current_id].get(metric["metricId"], 0) if span is None: # Lifetime metric whose task starts after ``ref``: no # window, so no facts may count, but keep the field shape. window_start = window_end = ref.isoformat() else: window_start, window_end = span[0].isoformat(), span[1].isoformat() metrics.append({ **{key: metric.get(key) for key in ("metricId", "name", "kind", "period", "unit", "target", "durationUnit")}, "value": value, "percent": _percent(value, metric["target"]), "windowStart": window_start, "windowEnd": window_end, }) scored = [item["percent"] for item in metrics if item["percent"] is not None] today = today_record.get(current_id) result["tasks"][current_id] = { "taskId": current_id, "progress": _half_up(sum(scored) / len(scored)) if scored else None, "today": {"recorded": today is not None, "completedAt": (today or {}).get("completedAt")}, "metrics": metrics, } return result def _rebuild_month(self, year: int, month: int) -> dict[str, Any]: summaries: dict[str, Any] = {} folder = self.daily_dir / str(year) for path in sorted(folder.glob(f"{year:04d}-{month:02d}-*.json")): daily = self._read(path) if not daily.get("closedAt"): continue for task_id, record in daily.get("records", {}).items(): task_summary = summaries.setdefault(task_id, {"days": 0, "completedDays": 0, "metricTotals": {}}) task_summary["days"] += 1 if record.get("completedAt"): task_summary["completedDays"] += 1 for metric_id, value in record.get("values", {}).items(): task_summary["metricTotals"][metric_id] = task_summary["metricTotals"].get(metric_id, 0) + value archive = { "schemaVersion": SCHEMA_VERSION, "month": f"{year:04d}-{month:02d}", "rebuiltAt": now_iso(), "tasks": summaries, } self._atomic_write(self.archive_dir / str(year) / f"{year:04d}-{month:02d}.json", archive) return archive def close_day(self, logical_date: date) -> None: with self._lock: path = self._daily_path(logical_date) daily = self._read(path, { "schemaVersion": SCHEMA_VERSION, "date": logical_date.isoformat(), "version": 1, "closedAt": None, "records": {}, }) if not daily.get("closedAt"): daily["closedAt"] = now_iso() daily["version"] += 1 self._atomic_write(path, daily) self._rebuild_month(logical_date.year, logical_date.month) def close_missed_days(self, today: date | None = None) -> int: with self._lock: today = today or datetime.now(SHANGHAI).date() meta = self._meta() cursor = date.fromisoformat(meta["lastClosedDate"]) + timedelta(days=1) last = today - timedelta(days=1) count = 0 while cursor <= last: self.close_day(cursor) meta["lastClosedDate"] = cursor.isoformat() cursor += timedelta(days=1) count += 1 if count: meta["version"] += 1 self._atomic_write(self.meta_path, meta) return count def export_all(self) -> dict[str, Any]: with self._lock: tasks = [self._read(path) for path in self.tasks_dir.glob("*.json")] revisions = [self._read(path) for path in self.revisions_dir.glob("*/*.json")] daily = [self._read(path) for path in self.daily_dir.glob("*/*.json")] archives = [self._read(path) for path in self.archive_dir.glob("*/*.json")] schedules = [self._read(path) for path in self.schedules_dir.glob("*.json")] schedule_revisions = [self._read(path) for path in self.schedule_revisions_dir.glob("*/*.json")] charts = [self._read(path) for path in self.charts_dir.glob("*.json")] return { "schemaVersion": SCHEMA_VERSION, "exportedAt": now_iso(), "meta": self._meta(), "tasks": tasks, "revisions": revisions, "daily": daily, "archives": archives, "schedules": schedules, "scheduleRevisions": schedule_revisions, "charts": charts, } def stats(self) -> dict[str, Any]: with self._lock: tasks = [self._read(path) for path in self.tasks_dir.glob("*.json")] daily_files = [self._read(path) for path in self.daily_dir.glob("*/*.json")] revisions = [path for path in self.revisions_dir.glob("*/*.json")] archives = [path for path in self.archive_dir.glob("*/*.json")] schedules = [path for path in self.schedules_dir.glob("*.json")] schedule_revisions = [path for path in self.schedule_revisions_dir.glob("*/*.json")] charts = [path for path in self.charts_dir.glob("*.json")] def bytes_of(paths): total = 0 for path in paths: try: total += path.stat().st_size except OSError: pass return total return { "schemaVersion": SCHEMA_VERSION, "taskCount": len(tasks), "activeTaskCount": len([t for t in tasks if t and not t.get("deletedAt")]), "deletedTaskCount": len([t for t in tasks if t and t.get("deletedAt")]), "dailyFactCount": len(daily_files), "revisionCount": len(revisions), "archiveCount": len(archives), "scheduleCount": len(schedules), "scheduleRevisionCount": len(schedule_revisions), "chartCount": len(charts), "storageBytes": bytes_of( list(self.tasks_dir.glob("*.json")) + list(self.daily_dir.glob("*/*.json")) + list(self.revisions_dir.glob("*/*.json")) + list(self.archive_dir.glob("*/*.json")) + list(self.schedules_dir.glob("*.json")) + list(self.schedule_revisions_dir.glob("*/*.json")) + charts ), } def _validate_identifier(self, value: Any, field: str) -> str: """Task/revision identifiers must be plain, path-safe tokens.""" if not isinstance(value, str) or not value or "/" in value or "\\" in value or value in {".", ".."} or ":" in value: raise ValidationError(f"备份 {field} 非法") return value def import_all(self, payload: dict[str, Any]) -> int: """Restore an export produced by ``export_all``. Replaces the entire data set. The payload is fully validated and staged into a temp directory before the live directories are swapped, so a malformed backup never leaves a partially imported repository. """ with self._lock: if payload.get("schemaVersion") != SCHEMA_VERSION: raise ValidationError("备份 schemaVersion 不受支持") meta = payload.get("meta") tasks = payload.get("tasks") revisions = payload.get("revisions") or [] daily = payload.get("daily") or [] schedules = payload.get("schedules") or [] schedule_revisions = payload.get("scheduleRevisions") or [] charts = payload.get("charts") or [] if not isinstance(meta, dict) or not isinstance(tasks, list): raise ValidationError("备份数据缺少 meta/tasks") # 第一遍：全部校验并规整为「目标路径 -> 内容」，不落盘。 staged: list[tuple[Path, Any]] = [] seen_tasks: set[str] = set() for task in tasks: if not isinstance(task, dict): raise ValidationError("备份任务数据无效") task_id = self._validate_identifier(task.get("taskId"), "taskId") if task_id in seen_tasks: raise ValidationError("备份包含重复任务") seen_tasks.add(task_id) staged.append((self._task_path(task_id), task)) revision_folders: set[Path] = set() for revision in revisions: if not isinstance(revision, dict): continue task_id = self._validate_identifier(revision.get("taskId"), "taskId") rev = self._validate_identifier(str(revision.get("revision")), "revision") folder = self.revisions_dir / task_id revision_folders.add(folder) staged.append((folder / f"{rev}.json", revision)) daily_dates: set[date] = set() for fact in daily: if not isinstance(fact, dict) or not fact.get("date"): raise ValidationError("备份每日事实无效") day = date.fromisoformat(str(fact["date"])) if day in daily_dates: raise ValidationError("备份包含重复每日事实") daily_dates.add(day) staged.append((self._daily_path(day), fact)) seen_schedules: set[str] = set() for schedule in schedules: if not isinstance(schedule, dict): raise ValidationError("备份日程数据无效") schedule_id = self._validate_identifier(schedule.get("scheduleId"), "scheduleId") if schedule_id in seen_schedules: raise ValidationError("备份包含重复日程") seen_schedules.add(schedule_id) staged.append((self._schedule_path(schedule_id), schedule)) for schedule_revision in schedule_revisions: if not isinstance(schedule_revision, dict): continue schedule_id = self._validate_identifier(schedule_revision.get("scheduleId"), "scheduleId") rev = self._validate_identifier(str(schedule_revision.get("revision")), "revision") folder = self.schedule_revisions_dir / schedule_id staged.append((folder / f"{rev}.json", schedule_revision)) seen_charts: set[str] = set() for chart in charts: if not isinstance(chart, dict): raise ValidationError("备份图表数据无效") chart_id = self._validate_identifier(chart.get("chartId"), "chartId") if chart_id in seen_charts: raise ValidationError("备份包含重复图表") seen_charts.add(chart_id) staged.append((self._chart_path(chart_id), chart)) # 原子替换整个数据集：先清空旧数据，再整体写入。 for folder in (self.tasks_dir, self.revisions_dir, self.daily_dir, self.archive_dir, self.schedules_dir, self.schedule_revisions_dir, self.charts_dir): for path in folder.glob("*"): if path.is_dir(): for child in path.iterdir(): child.unlink() path.rmdir() else: path.unlink() for path, value in staged: self._atomic_write(path, value) self._atomic_write(self.meta_path, meta) # 归档是可重建派生数据，导入后重建所有涉及月份的归档。 for day in sorted(daily_dates): self._rebuild_month(day.year, day.month) return len(seen_tasks)