personal-progress-monitor/app/repository.py

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)