personal-progress-monitor/app/main.py

from __future__ import annotations

import asyncio
import os
import secrets
from contextlib import asynccontextmanager
from datetime import date, datetime, time, timedelta
from pathlib import Path
from typing import Annotated

from fastapi import FastAPI, Query, Request
from fastapi.responses import FileResponse, JSONResponse
from fastapi.staticfiles import StaticFiles

from .auth import SessionManager
from .layouts import load_layout
from .models import DailyRecordInput, LoginInput, PurgeInput, ReorderInput, TaskInput, TaskUpdate
from .repository import ConflictError, JsonRepository, NotFoundError, SHANGHAI, ValidationError


PROJECT_ROOT = Path(__file__).resolve().parents[1]
SAFE_METHODS = {"GET", "HEAD", "OPTIONS"}


def _cookie_secure() -> bool:
    return os.getenv("PPM_COOKIE_SECURE", "false").lower() in {"1", "true", "yes"}


def _set_session_cookie(response: JSONResponse, session_id: str) -> None:
    response.set_cookie(
        "ppm_session",
        session_id,
        httponly=True,
        secure=_cookie_secure(),
        samesite="strict",
        max_age=60 * 60 * 24 * 30,
    )


def _set_csrf_cookie(response: JSONResponse, token: str) -> None:
    response.set_cookie(
        "ppm_csrf",
        token,
        httponly=False,
        secure=_cookie_secure(),
        samesite="strict",
        max_age=60 * 60 * 24 * 30,
    )


async def _archive_worker(repository: JsonRepository) -> None:
    while True:
        now = datetime.now(SHANGHAI)
        tomorrow = now.date() + timedelta(days=1)
        next_run = datetime.combine(tomorrow, time(hour=0, minute=0, second=2), SHANGHAI)
        await asyncio.sleep(max((next_run - now).total_seconds(), 1))
        repository.close_missed_days()


def create_app(
    data_dir: Path | None = None,
    layout_dir: Path | None = None,
    sessions: SessionManager | None = None,
) -> FastAPI:
    repository = JsonRepository(data_dir or Path(os.getenv("PPM_DATA_DIR", PROJECT_ROOT / "data")))
    layouts = layout_dir or Path(os.getenv("PPM_LAYOUT_DIR", PROJECT_ROOT / "config" / "layouts"))
    session_manager = sessions or SessionManager.from_environment()

    @asynccontextmanager
    async def lifespan(_: FastAPI):
        repository.close_missed_days()
        worker = asyncio.create_task(_archive_worker(repository))
        yield
        worker.cancel()
        try:
            await worker
        except asyncio.CancelledError:
            pass

    app = FastAPI(title="个人进度监视系统", version="0.1.0", lifespan=lifespan)
    app.state.repository = repository
    app.state.sessions = session_manager
    app.mount("/static", StaticFiles(directory=PROJECT_ROOT / "app" / "static"), name="static")

    @app.exception_handler(NotFoundError)
    async def handle_not_found(_: Request, exc: NotFoundError) -> JSONResponse:
        return JSONResponse({"detail": str(exc)}, status_code=404)

    @app.exception_handler(ConflictError)
    async def handle_conflict(_: Request, exc: ConflictError) -> JSONResponse:
        return JSONResponse({"detail": str(exc)}, status_code=409)

    @app.exception_handler(ValidationError)
    async def handle_validation(_: Request, exc: ValidationError) -> JSONResponse:
        return JSONResponse({"detail": str(exc)}, status_code=422)

    @app.middleware("http")
    async def security_middleware(request: Request, call_next):
        path = request.url.path
        public = path in {"/api/health", "/api/auth/status", "/api/auth/login"}
        if path.startswith("/api/") and not public:
            if session_manager.enabled and not session_manager.get(request.cookies.get("ppm_session")):
                return JSONResponse({"detail": "需要登录"}, status_code=401)
        if path.startswith("/api/") and request.method not in SAFE_METHODS and path != "/api/auth/login":
            submitted = request.headers.get("X-CSRF-Token", "")
            if session_manager.enabled:
                session = session_manager.get(request.cookies.get("ppm_session"))
                expected = session.csrf if session else ""
            else:
                expected = ""
            if session_manager.enabled and (not expected or not secrets.compare_digest(submitted, expected)):
                return JSONResponse({"detail": "CSRF 校验失败"}, status_code=403)
        return await call_next(request)

    @app.get("/", include_in_schema=False)
    async def index() -> FileResponse:
        return FileResponse(PROJECT_ROOT / "app" / "static" / "index.html")

    @app.get("/api/health")
    async def health() -> dict:
        return {"status": "ok", "singleWorker": True}

    @app.get("/api/auth/status")
    async def auth_status(request: Request) -> JSONResponse:
        if session_manager.enabled:
            session = session_manager.get(request.cookies.get("ppm_session"))
            body = {"enabled": True, "authenticated": bool(session), "csrfToken": session.csrf if session else None}
            return JSONResponse(body)
        token = request.cookies.get("ppm_csrf") or secrets.token_urlsafe(32)
        response = JSONResponse({"enabled": False, "authenticated": True, "csrfToken": token})
        _set_csrf_cookie(response, token)
        return response

    @app.post("/api/auth/login")
    async def login(payload: LoginInput, request: Request) -> JSONResponse:
        client_key = request.client.host if request.client else ""
        if session_manager.is_login_locked(client_key):
            return JSONResponse({"detail": "尝试次数过多，请稍后再试"}, status_code=429)
        result = session_manager.login(payload.password, client_key)
        if result is None:
            return JSONResponse({"detail": "密码错误"}, status_code=401)
        session_id, session = result
        response = JSONResponse({"authenticated": True, "csrfToken": session.csrf})
        _set_session_cookie(response, session_id)
        _set_csrf_cookie(response, session.csrf)
        return response

    @app.post("/api/auth/logout")
    async def logout(request: Request) -> JSONResponse:
        session_manager.logout(request.cookies.get("ppm_session"))
        response = JSONResponse({"authenticated": False})
        response.delete_cookie("ppm_session")
        response.delete_cookie("ppm_csrf")
        return response

    @app.get("/api/layouts/{page}")
    async def get_layout(page: str) -> dict:
        if page not in {"home", "task-detail"}:
            raise NotFoundError("布局不存在")
        return load_layout(layouts, page)

    @app.get("/api/tasks")
    async def list_tasks(includeDeleted: bool = False) -> dict:
        return repository.list_tasks(includeDeleted)

    @app.post("/api/tasks", status_code=201)
    async def create_task(payload: TaskInput) -> dict:
        return repository.create_task(payload)

    @app.get("/api/tasks/{task_id}")
    async def get_task(task_id: str, includeDeleted: bool = False) -> dict:
        return repository.get_task(task_id, includeDeleted)

    @app.put("/api/tasks/{task_id}")
    async def update_task(task_id: str, payload: TaskUpdate) -> dict:
        task_input = TaskInput.model_validate(payload.model_dump(exclude={"expectedVersion"}))
        return repository.update_task(task_id, task_input, payload.expectedVersion)

    @app.delete("/api/tasks/{task_id}")
    async def delete_task(task_id: str, expectedVersion: int = Query(ge=1)) -> dict:
        return repository.soft_delete_task(task_id, expectedVersion)

    @app.post("/api/tasks/{task_id}/restore")
    async def restore_task(task_id: str, expectedVersion: int = Query(ge=1)) -> dict:
        return repository.restore_task(task_id, expectedVersion)

    @app.post("/api/tasks/{task_id}/purge")
    async def purge_task(task_id: str, payload: PurgeInput) -> dict:
        repository.purge_task(task_id, payload.confirmation)
        return {"purged": True, "taskId": task_id}

    @app.put("/api/tasks/reorder/all")
    async def reorder_tasks(payload: ReorderInput) -> dict:
        return repository.reorder(payload.taskIds, payload.expectedVersion)

    @app.get("/api/tasks/{task_id}/daily/{logical_date}")
    async def get_daily(task_id: str, logical_date: date) -> dict:
        return repository.get_daily(task_id, logical_date)

    @app.put("/api/tasks/{task_id}/daily/{logical_date}")
    async def upsert_daily(task_id: str, logical_date: date, payload: DailyRecordInput) -> dict:
        return repository.upsert_daily(task_id, logical_date, payload)

    @app.get("/api/summary")
    async def summary(date_: Annotated[date | None, Query(alias="date")] = None) -> dict:
        return repository.summarize(date_)

    @app.get("/api/tasks/{task_id}/summary")
    async def task_summary(task_id: str, date_: Annotated[date | None, Query(alias="date")] = None) -> dict:
        return repository.summarize(date_, task_id)

    @app.get("/api/tasks/{task_id}/history")
    async def history(
        task_id: str,
        start: Annotated[date | None, Query()] = None,
        end: Annotated[date | None, Query()] = None,
    ) -> dict:
        logical_today = datetime.now(SHANGHAI).date()
        history_end = end or logical_today
        history_start = start or history_end - timedelta(days=364)
        return repository.history(task_id, history_start, history_end)

    @app.get("/api/export")
    async def export_data() -> JSONResponse:
        return JSONResponse(
            repository.export_all(),
            headers={"Content-Disposition": "attachment; filename=progress-export.json"},
        )

    @app.post("/api/admin/archive")
    async def run_archive() -> dict:
        return {"closedDays": repository.close_missed_days()}

    @app.get("/api/admin/stats")
    async def admin_stats() -> dict:
        return repository.stats()

    @app.post("/api/admin/import")
    async def admin_import(payload: dict) -> dict:
        count = repository.import_all(payload)
        return {"imported": count}

    return app


app = create_app()