import base64 import json import logging import os import re import shutil import time from collections import defaultdict, deque from contextlib import asynccontextmanager from datetime import UTC, datetime, timedelta from pathlib import Path, PureWindowsPath from uuid import UUID, uuid4 from fastapi import Depends, FastAPI, HTTPException, Query, Request, Response from fastapi.openapi.docs import get_swagger_ui_html from fastapi.responses import FileResponse, HTMLResponse, JSONResponse from fastapi.staticfiles import StaticFiles from sqlalchemy import and_, case, delete, exists, func, or_, select, tuple_, update from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession from .auth import ( COOKIE_NAME, CSRF_COOKIE_NAME, current_user, hash_password, hash_token, issue_session, session_token, verify_password, ) from .db import create_schema, get_db from .models import ( AppState, Attachment, Folder, PurgeOperation, RecurrenceException, RecurrenceTemplate, Session, Task, TaskList, User, utcnow, ) from .mvp import audit, occurrences from .mvp import router as mvp_router from .schemas import ( BatchResult, BatchTaskUpdate, ChangePasswordRequest, FolderCreate, FolderOut, InitializeRequest, ListCreate, ListMove, ListOut, ListReorder, LoginRequest, NameUpdate, SessionOut, TaskCreate, TaskDetailOut, TaskOut, TaskPage, TaskReorder, TaskUpdate, UserOut, ) @asynccontextmanager async def lifespan(app: FastAPI): from .config import get_settings if get_settings().auto_create_schema: await create_schema() yield app = FastAPI( title="dodo", version="0.2.0", lifespan=lifespan, docs_url=None, openapi_url=None, ) @app.middleware("http") async def security(request: Request, call_next): if request.method not in {"GET", "HEAD", "OPTIONS"} and request.cookies.get(COOKIE_NAME): csrf_cookie = request.cookies.get(CSRF_COOKIE_NAME) csrf_header = request.headers.get("x-csrf-token") # SameSite cookies plus same-origin validation; API clients may explicitly double-submit. origin = request.headers.get("origin") if origin and (not csrf_cookie or csrf_header != csrf_cookie): return JSONResponse({"detail": "CSRF 校验失败"}, status_code=403) response = await call_next(request) response.headers["X-Content-Type-Options"] = "nosniff" response.headers["X-Frame-Options"] = "DENY" response.headers["Referrer-Policy"] = "same-origin" response.headers["Content-Security-Policy"] = "default-src 'self'; img-src 'self' data:; style-src 'self' 'unsafe-inline'" return response @app.get("/api/docs", include_in_schema=False) async def docs(_: User = Depends(current_user)): return get_swagger_ui_html(openapi_url="/api/openapi.json", title="dodo API") @app.get("/api/openapi.json", include_in_schema=False) async def openapi(_: User = Depends(current_user)): return app.openapi() app.include_router(mvp_router) logger = logging.getLogger(__name__) _login_attempts: dict[tuple[str, str], deque[float]] = defaultdict(deque) @app.get("/health/live") async def live(): return {"status": "ok"} @app.get("/health/ready") async def ready(db: AsyncSession = Depends(get_db)): await db.execute(select(1)) return {"status": "ok"} @app.get("/api/v1/setup/status") async def setup_status(db: AsyncSession = Depends(get_db)): count = await db.scalar(select(func.count()).select_from(User)) return {"initialized": bool(count)} @app.post("/api/v1/setup/initialize", response_model=UserOut, status_code=201) async def initialize( payload: InitializeRequest, response: Response, request: Request, db: AsyncSession = Depends(get_db), ): count = await db.scalar(select(func.count()).select_from(User)) if count: raise HTTPException(status_code=409, detail="系统已经初始化") db.add(AppState(key="initialized")) user = User(username=payload.username, password_hash=hash_password(payload.password)) db.add(user) try: await db.flush() except IntegrityError as exc: await db.rollback() raise HTTPException(status_code=409, detail="系统已经初始化") from exc db.add(TaskList(user_id=user.id, name="收集箱", is_inbox=True)) await db.commit() await db.refresh(user) await issue_session(db, response, user, request) return user @app.post("/api/v1/auth/login", response_model=UserOut) async def login( payload: LoginRequest, response: Response, request: Request, db: AsyncSession = Depends(get_db), ): from .config import get_settings ip = request.client.host if request.client else "unknown" key = (payload.username.casefold(), ip) now = time.monotonic() attempts = _login_attempts[key] window = get_settings().login_window_seconds while attempts and attempts[0] < now - window: attempts.popleft() if len(attempts) >= get_settings().login_attempts: raise HTTPException(status_code=429, detail="登录尝试过多，请稍后再试", headers={"Retry-After": str(window)}) user = await db.scalar(select(User).where(User.username == payload.username)) if user is None or not verify_password(user.password_hash, payload.password): attempts.append(now) raise HTTPException(status_code=401, detail="用户名或密码错误") attempts.clear() await issue_session(db, response, user, request) return user @app.get("/api/v1/me", response_model=UserOut) async def me(user: User = Depends(current_user)): return user @app.post("/api/v1/auth/change-password", status_code=204) async def change_password( payload: ChangePasswordRequest, token: str = Depends(session_token), user: User = Depends(current_user), db: AsyncSession = Depends(get_db), ): if not verify_password(user.password_hash, payload.current_password): raise HTTPException(status_code=400, detail="当前密码不正确") user.password_hash = hash_password(payload.new_password) current_hash = hash_token(token) await db.execute( delete(Session).where(Session.user_id == user.id, Session.token_hash != current_hash) ) await db.commit() return Response(status_code=204) @app.get("/api/v1/bootstrap") async def bootstrap_data(user: User = Depends(current_user), db: AsyncSession = Depends(get_db)): folders = list((await db.scalars( select(Folder) .where(Folder.user_id == user.id, Folder.deleted_at.is_(None)) .order_by(Folder.position, Folder.created_at) )).all()) lists = list((await db.scalars( select(TaskList) .where(TaskList.user_id == user.id, TaskList.deleted_at.is_(None)) .order_by(TaskList.is_inbox.desc(), TaskList.position, TaskList.created_at) )).all()) inbox = next((item for item in lists if item.is_inbox), None) return { "user": UserOut.model_validate(user), "folders": [FolderOut.model_validate(item) for item in folders], "lists": [ListOut.model_validate(item) for item in lists], "inbox_id": inbox.id if inbox else None, } @app.post("/api/v1/auth/logout", status_code=204) async def logout( response: Response, token: str = Depends(session_token), db: AsyncSession = Depends(get_db), ): session = await db.scalar(select(Session).where(Session.token_hash == hash_token(token))) if session is not None: await db.delete(session) await db.commit() response.delete_cookie(COOKIE_NAME, path="/") response.delete_cookie(CSRF_COOKIE_NAME, path="/") return Response(status_code=204, headers=response.headers) @app.get("/api/v1/sessions", response_model=list[SessionOut]) async def list_sessions( token: str = Depends(session_token), user: User = Depends(current_user), db: AsyncSession = Depends(get_db), ): current_hash = hash_token(token) rows = (await db.scalars(select(Session).where(Session.user_id == user.id).order_by(Session.created_at.desc()))).all() return [{"id": row.id, "current": row.token_hash == current_hash, "created_at": row.created_at, "last_seen_at": row.last_seen_at, "expires_at": row.expires_at, "ip_address": row.ip_address, "user_agent": row.user_agent} for row in rows] @app.delete("/api/v1/sessions/{session_id}", status_code=204) async def revoke_session( session_id: UUID, response: Response, token: str = Depends(session_token), user: User = Depends(current_user), db: AsyncSession = Depends(get_db), ): row = await db.scalar(select(Session).where(Session.id == session_id, Session.user_id == user.id)) if not row: raise HTTPException(404, "会话不存在") current = row.token_hash == hash_token(token) await db.delete(row); await db.commit() if current: response.delete_cookie(COOKIE_NAME, path="/"); response.delete_cookie(CSRF_COOKIE_NAME, path="/") return Response(status_code=204, headers=response.headers) @app.post("/api/v1/folders", response_model=FolderOut, status_code=201) async def create_folder( payload: FolderCreate, user: User = Depends(current_user), db: AsyncSession = Depends(get_db), ): folder = Folder(user_id=user.id, name=payload.name) db.add(folder) await db.commit() await db.refresh(folder) return folder @app.get("/api/v1/folders", response_model=list[FolderOut]) async def list_folders(user: User = Depends(current_user), db: AsyncSession = Depends(get_db)): query = select(Folder).where(Folder.user_id == user.id, Folder.deleted_at.is_(None)) return list((await db.scalars(query.order_by(Folder.position, Folder.created_at))).all()) @app.patch("/api/v1/folders/{folder_id}", response_model=FolderOut) async def rename_folder( folder_id: UUID, payload: NameUpdate, user: User = Depends(current_user), db: AsyncSession = Depends(get_db), ): folder = await db.scalar( select(Folder).where( Folder.id == folder_id, Folder.user_id == user.id, Folder.deleted_at.is_(None) ) ) if folder is None: raise HTTPException(status_code=404, detail="文件夹不存在") folder.name = payload.name await db.flush() audit(db, user.id, "update", "folder", folder.id) await db.commit() return folder @app.delete("/api/v1/folders/{folder_id}", status_code=204) async def delete_folder( folder_id: UUID, user: User = Depends(current_user), db: AsyncSession = Depends(get_db), ): folder = await db.scalar( select(Folder).where( Folder.id == folder_id, Folder.user_id == user.id, Folder.deleted_at.is_(None) ) ) if folder is None: raise HTTPException(status_code=404, detail="文件夹不存在") folder.deleted_at = utcnow() audit(db, user.id, "delete", "folder", folder.id) await db.execute( update(TaskList) .where(TaskList.user_id == user.id, TaskList.folder_id == folder.id) .values(folder_id=None) ) await db.commit() return Response(status_code=204) async def _lock_list_order(db: AsyncSession, user_id: UUID) -> None: # Serialize list creation/moves/reorders per user so max-position allocation # and full-scope reorder validation cannot interleave. await db.scalar(select(User.id).where(User.id == user_id).with_for_update()) @app.post("/api/v1/lists", response_model=ListOut, status_code=201) async def create_list( payload: ListCreate, user: User = Depends(current_user), db: AsyncSession = Depends(get_db), ): if payload.folder_id and not await db.scalar( select(Folder.id).where( Folder.id == payload.folder_id, Folder.user_id == user.id, Folder.deleted_at.is_(None), ) ): raise HTTPException(status_code=404, detail="文件夹不存在") await _lock_list_order(db, user.id) folder_scope = ( TaskList.folder_id == payload.folder_id if payload.folder_id is not None else TaskList.folder_id.is_(None) ) max_position = await db.scalar( select(func.max(TaskList.position)).where( TaskList.user_id == user.id, folder_scope, TaskList.is_inbox.is_(False), TaskList.deleted_at.is_(None), ) ) item = TaskList( user_id=user.id, folder_id=payload.folder_id, name=payload.name, position=(max_position if max_position is not None else -1) + 1, ) db.add(item) await db.flush() audit(db, user.id, "create", "list", item.id) await db.commit() await db.refresh(item) return item @app.get("/api/v1/lists", response_model=list[ListOut]) async def list_lists( archived: bool = False, user: User = Depends(current_user), db: AsyncSession = Depends(get_db), ): deleted_condition = TaskList.deleted_at.is_not(None) if archived else TaskList.deleted_at.is_(None) query = select(TaskList).where(TaskList.user_id == user.id, deleted_condition) ordering = (TaskList.is_inbox.desc(), TaskList.position, TaskList.created_at) return list((await db.scalars(query.order_by(*ordering))).all()) async def _owned_list( db: AsyncSession, user_id: UUID, list_id: UUID, *, include_archived: bool = False, ) -> TaskList: conditions = [TaskList.id == list_id, TaskList.user_id == user_id] if not include_archived: conditions.append(TaskList.deleted_at.is_(None)) item = await db.scalar(select(TaskList).where(*conditions)) if item is None: raise HTTPException(status_code=404, detail="清单不存在") return item @app.put("/api/v1/lists/reorder", status_code=204) async def reorder_lists( payload: ListReorder, user: User = Depends(current_user), db: AsyncSession = Depends(get_db), ): if payload.folder_id is not None: folder = await db.scalar( select(Folder).where( Folder.id == payload.folder_id, Folder.user_id == user.id, Folder.deleted_at.is_(None), ) ) if folder is None: raise HTTPException(status_code=404, detail="文件夹不存在") await _lock_list_order(db, user.id) folder_scope = ( TaskList.folder_id == payload.folder_id if payload.folder_id is not None else TaskList.folder_id.is_(None) ) scope_rows = list( ( await db.scalars( select(TaskList) .where( TaskList.user_id == user.id, folder_scope, TaskList.is_inbox.is_(False), TaskList.deleted_at.is_(None), ) .order_by(TaskList.position, TaskList.created_at, TaskList.id) .with_for_update() ) ).all() ) if any(list_id not in {row.id for row in scope_rows} for list_id in payload.list_ids): inbox_requested = await db.scalar( select(TaskList.id).where( TaskList.id.in_(payload.list_ids), TaskList.user_id == user.id, TaskList.is_inbox.is_(True), ) ) if inbox_requested: raise HTTPException(status_code=409, detail="系统收集箱不能排序") raise HTTPException(status_code=409, detail="清单不属于指定作用域") if set(payload.list_ids) != {row.id for row in scope_rows}: raise HTTPException(status_code=409, detail="清单顺序已变化，请刷新后重试") rows_by_id = {row.id: row for row in scope_rows} for position, list_id in enumerate(payload.list_ids): rows_by_id[list_id].position = position await db.commit() return Response(status_code=204) @app.put("/api/v1/lists/{list_id}/move", response_model=ListOut) async def move_list_to_folder( list_id: UUID, payload: ListMove, user: User = Depends(current_user), db: AsyncSession = Depends(get_db), ): await _lock_list_order(db, user.id) item = await _owned_list(db, user.id, list_id) if item.is_inbox: raise HTTPException(status_code=409, detail="系统收集箱不能移动") if payload.folder_id is not None: folder = await db.scalar( select(Folder).where( Folder.id == payload.folder_id, Folder.user_id == user.id, Folder.deleted_at.is_(None), ) ) if folder is None: raise HTTPException(status_code=404, detail="文件夹不存在") target_scope = ( TaskList.folder_id == payload.folder_id if payload.folder_id is not None else TaskList.folder_id.is_(None) ) target_rows = list( ( await db.scalars( select(TaskList) .where( TaskList.user_id == user.id, target_scope, TaskList.is_inbox.is_(False), TaskList.deleted_at.is_(None), TaskList.id != item.id, ) .order_by(TaskList.position, TaskList.created_at, TaskList.id) .with_for_update() ) ).all() ) expected_ids = {row.id for row in target_rows} | {item.id} if set(payload.list_ids) != expected_ids: raise HTTPException(status_code=409, detail="目标清单顺序已变化，请刷新后重试") rows_by_id = {row.id: row for row in target_rows} rows_by_id[item.id] = item item.folder_id = payload.folder_id for position, ordered_id in enumerate(payload.list_ids): rows_by_id[ordered_id].position = position await db.flush() audit( db, user.id, "move", "list", item.id, folder_id=str(payload.folder_id) if payload.folder_id else None, ) await db.commit() await db.refresh(item) return item @app.patch("/api/v1/lists/{list_id}", response_model=ListOut) async def rename_list( list_id: UUID, payload: NameUpdate, user: User = Depends(current_user), db: AsyncSession = Depends(get_db), ): item = await _owned_list(db, user.id, list_id) if item.is_inbox: raise HTTPException(status_code=409, detail="系统收集箱不能重命名") item.name = payload.name await db.flush() audit(db, user.id, "update", "list", item.id) await db.commit() await db.refresh(item) return item @app.delete("/api/v1/lists/{list_id}", status_code=204) async def delete_list( list_id: UUID, user: User = Depends(current_user), db: AsyncSession = Depends(get_db), ): item = await _owned_list(db, user.id, list_id) if item.is_inbox: raise HTTPException(status_code=409, detail="系统收集箱不能删除") item.deleted_at = utcnow() await db.commit() return Response(status_code=204) @app.post("/api/v1/lists/{list_id}/restore", response_model=ListOut) async def restore_list( list_id: UUID, user: User = Depends(current_user), db: AsyncSession = Depends(get_db), ): item = await _owned_list(db, user.id, list_id, include_archived=True) if item.is_inbox or item.deleted_at is None: raise HTTPException(status_code=409, detail="清单未归档") item.deleted_at = None await db.flush() audit(db, user.id, "restore", "list", item.id) await db.commit() await db.refresh(item) return item def _attachment_path(attachment_root: Path, storage_name: str) -> Path: if not storage_name or "\x00" in storage_name: raise ValueError(storage_name) windows_path = PureWindowsPath(storage_name) if windows_path.is_absolute() or windows_path.drive: raise ValueError(storage_name) parts = storage_name.replace("\\", "/").split("/") if any(part in {"", ".", ".."} for part in parts): raise ValueError(storage_name) relative_path = Path(*parts) resolved_path = (attachment_root / relative_path).resolve() try: resolved_path.relative_to(attachment_root) except ValueError as exc: raise ValueError(storage_name) from exc return resolved_path def _create_unique_quarantine(attachment_root: Path) -> tuple[UUID, Path]: while True: operation_id = uuid4() trash_dir = attachment_root / ".purge-trash" / str(operation_id) try: trash_dir.mkdir(parents=True, exist_ok=False) except FileExistsError: continue return operation_id, trash_dir def _remove_owned_quarantine(trash_dir: Path) -> None: shutil.rmtree(trash_dir) try: trash_dir.parent.rmdir() except OSError: pass def _restore_quarantined_files( moved_files: list[tuple[Path, Path]], *, list_id: UUID, operation_id: UUID ) -> list[str]: restore_errors: list[str] = [] for source_path, trash_path in reversed(moved_files): try: source_path.parent.mkdir(parents=True, exist_ok=True) os.replace(trash_path, source_path) except OSError: restore_errors.append(str(trash_path)) logger.exception( "List purge attachment restore failed", extra={ "list_id": str(list_id), "operation_id": str(operation_id), "trash_path": str(trash_path), }, ) return restore_errors async def _retry_purge_cleanup(db: AsyncSession, operation: PurgeOperation) -> Response: trash_dir = Path(operation.trash_dir) try: if trash_dir.exists(): _remove_owned_quarantine(trash_dir) except OSError as exc: logger.exception( "List purge retry cleanup failed: %s", trash_dir, extra={ "list_id": str(operation.list_id), "operation_id": str(operation.id), "trash_path": str(trash_dir), }, ) raise HTTPException( status_code=500, detail=( "清单数据已删除，但附件清理未完成，" f"操作ID: {operation.id}, 隔离路径: {trash_dir}" ), ) from exc await db.delete(operation) await db.commit() return Response(status_code=204) @app.delete("/api/v1/lists/{list_id}/purge", status_code=204) async def purge_list( list_id: UUID, user: User = Depends(current_user), db: AsyncSession = Depends(get_db), ): from .config import get_settings item = await db.scalar( select(TaskList) .where(TaskList.id == list_id, TaskList.user_id == user.id) .with_for_update() ) if item is None: pending_operation = await db.scalar( select(PurgeOperation) .where(PurgeOperation.list_id == list_id, PurgeOperation.user_id == user.id) .with_for_update() ) if pending_operation is not None: return await _retry_purge_cleanup(db, pending_operation) raise HTTPException(status_code=404, detail="清单不存在") if item.is_inbox: raise HTTPException(status_code=409, detail="系统收集箱不能永久删除") if item.deleted_at is None: raise HTTPException(status_code=409, detail="请先归档再永久删除") task_ids = select(Task.id).where(Task.list_id == item.id) template_ids = select(RecurrenceTemplate.id).where(RecurrenceTemplate.task_id.in_(task_ids)) storage_names = list( (await db.scalars(select(Attachment.storage_name).where(Attachment.task_id.in_(task_ids)))).all() ) attachment_root = Path(get_settings().attachment_dir).resolve() try: attachment_paths = [_attachment_path(attachment_root, name) for name in storage_names] except ValueError as exc: await db.rollback() logger.error( "List purge rejected unsafe attachment path", extra={"list_id": str(list_id), "storage_name": str(exc)}, ) raise HTTPException( status_code=409, detail="附件存储路径无效，无法永久删除清单" ) from exc moved_files: list[tuple[Path, Path]] = [] owns_trash_dir = False operation_id = uuid4() trash_dir = attachment_root / ".purge-trash" / str(operation_id) try: operation_id, trash_dir = _create_unique_quarantine(attachment_root) owns_trash_dir = True for source_path in attachment_paths: if not source_path.is_file(): continue trash_path = trash_dir / source_path.relative_to(attachment_root) trash_path.parent.mkdir(parents=True, exist_ok=True) os.replace(source_path, trash_path) moved_files.append((source_path, trash_path)) except OSError as exc: restore_errors = _restore_quarantined_files( moved_files, list_id=list_id, operation_id=operation_id ) if owns_trash_dir and not restore_errors: _remove_owned_quarantine(trash_dir) await db.rollback() if restore_errors: raise HTTPException( status_code=500, detail=( "附件隔离失败且部分附件恢复失败，" f"操作ID: {operation_id}, 隔离路径: {trash_dir}, " f"未恢复: {', '.join(restore_errors)}" ), ) from exc raise HTTPException(status_code=500, detail="附件隔离失败，清单未删除") from exc operation = PurgeOperation( id=operation_id, list_id=list_id, user_id=user.id, trash_dir=str(trash_dir), status="cleanup_pending", ) try: await db.execute( delete(RecurrenceException).where(RecurrenceException.template_id.in_(template_ids)) ) await db.execute(delete(RecurrenceTemplate).where(RecurrenceTemplate.task_id.in_(task_ids))) await db.execute(delete(Attachment).where(Attachment.task_id.in_(task_ids))) await db.execute(delete(Task).where(Task.list_id == item.id)) await db.delete(item) db.add(operation) await db.commit() except Exception as exc: await db.rollback() restore_errors = _restore_quarantined_files( moved_files, list_id=list_id, operation_id=operation_id ) if not restore_errors: _remove_owned_quarantine(trash_dir) raise HTTPException(status_code=500, detail="数据库提交失败，清单未删除") from exc raise HTTPException( status_code=500, detail=( "数据库提交失败且部分附件恢复失败，" f"操作ID: {operation_id}, 隔离路径: {trash_dir}, " f"未恢复: {', '.join(restore_errors)}" ), ) from exc try: _remove_owned_quarantine(trash_dir) except OSError as exc: logger.exception( "List purge committed but quarantine cleanup failed: %s", trash_dir, extra={ "list_id": str(list_id), "operation_id": str(operation_id), "trash_path": str(trash_dir), }, ) raise HTTPException( status_code=500, detail=( "清单数据已删除，但附件清理未完成，" f"操作ID: {operation_id}, 隔离路径: {trash_dir}" ), ) from exc await db.delete(operation) try: await db.commit() except Exception as exc: await db.rollback() raise HTTPException( status_code=500, detail=( "清单数据及附件已删除，但清理操作状态未完成，" f"操作ID: {operation_id}；请重试永久删除" ), ) from exc return Response(status_code=204) def _task_ordering(): """Portable total ordering: open, dated, due time, then manual/stable keys.""" return ( case((Task.completed.is_(False), 0), else_=1), case((Task.due_at.is_(None), 1), else_=0), Task.due_at.asc(), Task.position.asc(), Task.created_at.asc(), Task.id.asc(), ) def _task_cursor_values(task: Task) -> list[object]: return [ 0 if not task.completed else 1, 1 if task.due_at is None else 0, task.due_at.isoformat() if task.due_at else None, task.position, task.created_at.isoformat(), str(task.id), ] def _encode_task_cursor(task: Task) -> str: raw = json.dumps({"v": 2, "keys": _task_cursor_values(task)}).encode() return base64.urlsafe_b64encode(raw).decode().rstrip("=") _CURSOR_MAX_POSITION = 2**63 - 1 _CURSOR_DATETIME = re.compile( r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d{1,6})?(?:Z|[+-]\d{2}:\d{2})$" ) def _cursor_rank(value: object) -> int: if type(value) is not int or value not in (0, 1): raise ValueError return value def _cursor_position(value: object) -> int: if type(value) is not int or not 0 <= value <= _CURSOR_MAX_POSITION: raise ValueError return value def _cursor_datetime(value: object) -> datetime: if not isinstance(value, str) or not _CURSOR_DATETIME.fullmatch(value): raise ValueError parsed = datetime.fromisoformat(value) if parsed.tzinfo is None or parsed.utcoffset() is None: raise ValueError return parsed.astimezone(UTC) def _decode_task_cursor(cursor: str) -> tuple[int, int, datetime | None, int, datetime, UUID]: try: raw = base64.urlsafe_b64decode(cursor + "=" * (-len(cursor) % 4)) payload = json.loads(raw) if not isinstance(payload, dict) or type(payload.get("v")) is not int or payload["v"] != 2: raise ValueError keys = payload["keys"] if not isinstance(keys, list) or len(keys) != 6: raise ValueError completed_rank, due_rank, due_at, position, created_at, task_id = keys completed_rank = _cursor_rank(completed_rank) due_rank = _cursor_rank(due_rank) if (due_rank == 0) != (due_at is not None): raise ValueError if not isinstance(task_id, str): raise TypeError return ( completed_rank, due_rank, _cursor_datetime(due_at) if due_at is not None else None, _cursor_position(position), _cursor_datetime(created_at), UUID(task_id), ) except (ValueError, TypeError, KeyError, UnicodeError, json.JSONDecodeError, OverflowError) as exc: raise HTTPException(status_code=422, detail="无效的游标") from exc def _task_after_cursor(values: tuple[int, int, datetime | None, int, datetime, UUID]): completed_rank, due_rank, due_at, position, created_at, task_id = values completed_expr = case((Task.completed.is_(False), 0), else_=1) due_expr = case((Task.due_at.is_(None), 1), else_=0) prefix = [completed_expr == completed_rank, due_expr == due_rank] alternatives = [completed_expr > completed_rank, and_(prefix[0], due_expr > due_rank)] if due_rank == 0 and due_at is not None: alternatives.append(and_(*prefix, Task.due_at > due_at)) prefix.append(Task.due_at == due_at) alternatives.extend( [ and_(*prefix, Task.position > position), and_(*prefix, Task.position == position, Task.created_at > created_at), and_( *prefix, Task.position == position, Task.created_at == created_at, Task.id > task_id, ), ] ) return or_(*alternatives) def _encode_trash_cursor(created_at: datetime, task_id: UUID) -> str: raw = json.dumps([created_at.isoformat(), str(task_id)]).encode() return base64.urlsafe_b64encode(raw).decode().rstrip("=") def _decode_trash_cursor(cursor: str) -> tuple[datetime, UUID]: try: raw = base64.urlsafe_b64decode(cursor + "=" * (-len(cursor) % 4)) timestamp, task_id = json.loads(raw) return datetime.fromisoformat(timestamp), UUID(task_id) except (ValueError, TypeError, json.JSONDecodeError) as exc: raise HTTPException(status_code=422, detail="无效的游标") from exc @app.post("/api/v1/tasks", response_model=TaskOut, status_code=201) async def create_task( payload: TaskCreate, user: User = Depends(current_user), db: AsyncSession = Depends(get_db), ): await _owned_list(db, user.id, payload.list_id) if payload.parent_id: parent = await db.scalar( select(Task).where( Task.id == payload.parent_id, Task.user_id == user.id, Task.list_id == payload.list_id, Task.parent_id.is_(None), Task.deleted_at.is_(None), ) ) if parent is None: raise HTTPException(status_code=400, detail="父任务必须是同一清单的顶层任务") if payload.rrule: if not payload.due_at: raise HTTPException(status_code=422, detail="重复任务需要截止时间") from .mvp import parse_rrule parse_rrule(payload.rrule) data = payload.model_dump(exclude={"rrule"}) parent_filter = Task.parent_id == payload.parent_id if payload.parent_id else Task.parent_id.is_(None) max_position = await db.scalar(select(func.max(Task.position)).where( Task.user_id == user.id, Task.list_id == payload.list_id, parent_filter, Task.deleted_at.is_(None), )) task = Task(user_id=user.id, position=(max_position if max_position is not None else -1) + 1, **data) db.add(task) await db.flush() if payload.rrule: db.add(RecurrenceTemplate(user_id=user.id, task_id=task.id, rrule=payload.rrule.upper(), starts_at=task.due_at)) audit(db, user.id, "create", "task", task.id) await db.commit() await db.refresh(task) return task @app.get("/api/v1/tasks", response_model=TaskPage) async def list_tasks( q: str | None = None, list_id: UUID | None = None, completed: bool | None = None, due_from: datetime | None = None, due_to: datetime | None = None, cursor: str | None = None, page: int | None = Query(default=None, ge=1), page_size: int | None = Query(default=None, ge=1, le=100), limit: int = Query(default=50, ge=1, le=100), user: User = Depends(current_user), db: AsyncSession = Depends(get_db), ): query = select(Task).where( Task.user_id == user.id, Task.deleted_at.is_(None), Task.parent_id.is_(None), exists( select(TaskList.id).where( TaskList.id == Task.list_id, TaskList.user_id == user.id, TaskList.deleted_at.is_(None), ) ), ) if list_id is not None: await _owned_list(db, user.id, list_id) query = query.where(Task.list_id == list_id) if completed is not None: query = query.where(Task.completed.is_(completed)) if due_from is not None: query = query.where(Task.due_at >= due_from) if due_to is not None: query = query.where(Task.due_at < due_to) if q: pattern = f"%{q}%" list_match = exists( select(TaskList.id).where( TaskList.id == Task.list_id, TaskList.user_id == user.id, TaskList.name.ilike(pattern), ) ) query = query.where( or_(Task.title.ilike(pattern), Task.description.ilike(pattern), list_match) ) total = await db.scalar(select(func.count()).select_from(query.order_by(None).subquery())) or 0 ordering = _task_ordering() if page is not None: size = page_size or limit items = list((await db.scalars(query.order_by(*ordering).offset((page - 1) * size).limit(size))).all()) return TaskPage(items=await _task_details(db, items), total=total, page=page, page_size=size) if cursor: query = query.where(_task_after_cursor(_decode_task_cursor(cursor))) rows = list((await db.scalars(query.order_by(*ordering).limit(limit + 1))).all()) has_more = len(rows) > limit items = rows[:limit] next_cursor = _encode_task_cursor(items[-1]) if has_more else None return TaskPage(items=await _task_details(db, items), next_cursor=next_cursor, total=total, page=1, page_size=limit) async def _task_details(db: AsyncSession, tasks: list[Task]) -> list[TaskDetailOut]: if not tasks: return [] allowed_scopes = {(task.id, task.user_id, task.list_id) for task in tasks} subtasks = list( ( await db.scalars( select(Task) .where( tuple_(Task.parent_id, Task.user_id, Task.list_id).in_(allowed_scopes), Task.deleted_at.is_(None), ) .order_by(*_task_ordering()) ) ).all() ) subtasks_by_task: dict[UUID, list[Task]] = defaultdict(list) for subtask in subtasks: if (subtask.parent_id, subtask.user_id, subtask.list_id) in allowed_scopes: subtasks_by_task[subtask.parent_id].append(subtask) return [ TaskDetailOut( **TaskOut.model_validate(task).model_dump(), subtasks=subtasks_by_task[task.id], ) for task in tasks ] async def _task_detail(db: AsyncSession, task: Task) -> TaskDetailOut: return (await _task_details(db, [task]))[0] @app.put("/api/v1/tasks/reorder", status_code=204) async def reorder_tasks( payload: TaskReorder, user: User = Depends(current_user), db: AsyncSession = Depends(get_db), ): rows = list((await db.scalars(select(Task).where( Task.id.in_(payload.task_ids), Task.user_id == user.id, Task.deleted_at.is_(None) ))).all()) if len(rows) != len(payload.task_ids): raise HTTPException(status_code=404, detail="任务不存在") parent_scopes = {row.parent_id for row in rows} if len(parent_scopes) != 1: raise HTTPException(status_code=400, detail="只能调整同一层级任务的顺序") list_scopes = {row.list_id for row in rows} if len(list_scopes) != 1: raise HTTPException(status_code=400, detail="只能调整同一清单内任务的顺序") sort_tiers = {(row.completed, row.due_at) for row in rows} if len(sort_tiers) != 1: raise HTTPException(status_code=400, detail="只能调整相同完成状态和截止时间档的任务顺序") parent_id = next(iter(parent_scopes)) scope_query = select(Task).where( Task.user_id == user.id, Task.list_id == next(iter(list_scopes)), Task.deleted_at.is_(None), Task.parent_id.is_(None) if parent_id is None else Task.parent_id == parent_id, ).order_by(Task.position, Task.created_at, Task.id) scope_rows = list((await db.scalars(scope_query)).all()) requested = set(payload.task_ids) ordered_rows = iter([next(row for row in rows if row.id == task_id) for task_id in payload.task_ids]) merged = [next(ordered_rows) if row.id in requested else row for row in scope_rows] for position, row in enumerate(merged): row.position = position await db.commit() return Response(status_code=204) @app.get("/api/v1/tasks/{task_id}", response_model=TaskDetailOut) async def get_task( task_id: UUID, user: User = Depends(current_user), db: AsyncSession = Depends(get_db), ): task = await db.scalar( select(Task) .join(TaskList, TaskList.id == Task.list_id) .where( Task.id == task_id, Task.user_id == user.id, Task.deleted_at.is_(None), TaskList.user_id == user.id, TaskList.deleted_at.is_(None), ) ) if task is None: raise HTTPException(status_code=404, detail="任务不存在") return await _task_detail(db, task) @app.patch("/api/v1/tasks/{task_id}", response_model=TaskDetailOut) async def update_task( task_id: UUID, payload: TaskUpdate, user: User = Depends(current_user), db: AsyncSession = Depends(get_db), ): task = await db.scalar(select(Task).where(Task.id == task_id, Task.user_id == user.id, Task.deleted_at.is_(None))) if task is None: raise HTTPException(status_code=404, detail="任务不存在") data = payload.model_dump(exclude_unset=True) expected_version = data.pop("version") recurrence = None reset_subtasks = False if data.get("completed") is True: recurrence = await db.scalar(select(RecurrenceTemplate).where( RecurrenceTemplate.task_id == task_id, RecurrenceTemplate.user_id == user.id )) if recurrence: next_items = occurrences( recurrence.rrule, recurrence.starts_at, recurrence.starts_at + timedelta(microseconds=1), recurrence.starts_at + timedelta(days=3660), recurrence.ends_at, ) if next_items: data["completed"] = False data["due_at"] = next_items[0] recurrence.starts_at = next_items[0] reset_subtasks = True if "list_id" in data: await _owned_list(db, user.id, data["list_id"]) if task.parent_id: parent_list = await db.scalar( select(Task.list_id).where(Task.id == task.parent_id, Task.user_id == user.id) ) if parent_list != data["list_id"]: raise HTTPException(status_code=400, detail="子任务必须与父任务属于同一清单") data["version"] = Task.version + 1 data["updated_at"] = utcnow() result = await db.execute( update(Task) .where( Task.id == task_id, Task.user_id == user.id, Task.deleted_at.is_(None), Task.version == expected_version, ) .values(**data) .returning(Task) ) task = result.scalar_one_or_none() if task is None: exists_id = await db.scalar( select(Task.id).where(Task.id == task_id, Task.user_id == user.id, Task.deleted_at.is_(None)) ) if exists_id: raise HTTPException(status_code=409, detail="任务已被更新，请刷新后重试") raise HTTPException(status_code=404, detail="任务不存在") if reset_subtasks: reset_at = utcnow() await db.execute( update(Task) .where( Task.parent_id == task.id, Task.user_id == user.id, Task.deleted_at.is_(None), Task.completed.is_(True), ) .values(completed=False, version=Task.version + 1, updated_at=reset_at) ) if "list_id" in data and task.parent_id is None: await db.execute( update(Task) .where(Task.parent_id == task.id, Task.user_id == user.id, Task.deleted_at.is_(None)) .values(list_id=task.list_id, version=Task.version + 1, updated_at=utcnow()) ) changed = {k for k in data if k not in {"version", "updated_at"}} if changed & {"title", "description", "priority", "due_at", "list_id", "completed"}: action = "complete" if data.get("completed") is True else "update" audit(db, user.id, action, "task", task.id, fields=sorted(changed)) await db.commit() return await _task_detail(db, task) @app.delete("/api/v1/tasks/{task_id}", status_code=204) async def delete_task( task_id: UUID, user: User = Depends(current_user), db: AsyncSession = Depends(get_db), ): task = await db.scalar( select(Task).where(Task.id == task_id, Task.user_id == user.id, Task.deleted_at.is_(None)) ) if task is None: raise HTTPException(status_code=404, detail="任务不存在") deleted_at = utcnow() await db.execute( update(Task) .where(or_(Task.id == task.id, Task.parent_id == task.id), Task.user_id == user.id) .values(deleted_at=deleted_at, version=Task.version + 1, updated_at=deleted_at) ) audit(db, user.id, "delete", "task", task.id) await db.commit() return Response(status_code=204) @app.get("/api/v1/trash", response_model=TaskPage) async def list_trash( cursor: str | None = None, page: int | None = Query(default=None, ge=1), page_size: int | None = Query(default=None, ge=1, le=100), limit: int = Query(default=50, ge=1, le=100), user: User = Depends(current_user), db: AsyncSession = Depends(get_db), ): query = select(Task).where( Task.user_id == user.id, Task.deleted_at.is_not(None), Task.parent_id.is_(None) ) total = await db.scalar(select(func.count()).select_from(query.order_by(None).subquery())) or 0 ordering = (Task.created_at, Task.id) if page is not None: size = page_size or limit items = list((await db.scalars(query.order_by(*ordering).offset((page - 1) * size).limit(size))).all()) return TaskPage(items=await _task_details(db, items), total=total, page=page, page_size=size) if cursor: created_at, task_id = _decode_trash_cursor(cursor) query = query.where( or_(Task.created_at > created_at, (Task.created_at == created_at) & (Task.id > task_id)) ) rows = list((await db.scalars(query.order_by(*ordering).limit(limit + 1))).all()) has_more = len(rows) > limit items = rows[:limit] next_cursor = _encode_trash_cursor(items[-1].created_at, items[-1].id) if has_more else None return TaskPage(items=await _task_details(db, items), next_cursor=next_cursor, total=total, page=1, page_size=limit) @app.post("/api/v1/tasks/{task_id}/restore", response_model=TaskDetailOut) async def restore_task( task_id: UUID, user: User = Depends(current_user), db: AsyncSession = Depends(get_db), ): task = await db.scalar( select(Task).where(Task.id == task_id, Task.user_id == user.id, Task.deleted_at.is_not(None)) ) if task is None: raise HTTPException(status_code=404, detail="回收站中不存在该任务") await _owned_list(db, user.id, task.list_id) now = utcnow() await db.execute( update(Task) .where(or_(Task.id == task.id, Task.parent_id == task.id), Task.user_id == user.id) .values(deleted_at=None, version=Task.version + 1, updated_at=now) ) audit(db, user.id, "restore", "task", task.id) await db.commit() await db.refresh(task) return await _task_detail(db, task) @app.delete("/api/v1/trash/{task_id}", status_code=204) async def permanently_delete_task( task_id: UUID, user: User = Depends(current_user), db: AsyncSession = Depends(get_db), ): task = await db.scalar( select(Task).where(Task.id == task_id, Task.user_id == user.id, Task.deleted_at.is_not(None)) ) if task is None: raise HTTPException(status_code=404, detail="回收站中不存在该任务") await db.execute(delete(Task).where(Task.parent_id == task.id, Task.user_id == user.id)) await db.delete(task) await db.commit() return Response(status_code=204) @app.post("/api/v1/tasks/batch", response_model=BatchResult) async def batch_update_tasks( payload: BatchTaskUpdate, user: User = Depends(current_user), db: AsyncSession = Depends(get_db), ): task_ids = list(dict.fromkeys(payload.task_ids)) tasks = list( ( await db.scalars( select(Task).where( Task.id.in_(task_ids), Task.user_id == user.id, Task.deleted_at.is_(None) ) ) ).all() ) if len(tasks) != len(task_ids): raise HTTPException(status_code=404, detail="一个或多个任务不存在") if payload.list_id is not None: await _owned_list(db, user.id, payload.list_id) standalone_children = [task for task in tasks if task.parent_id is not None] if standalone_children: raise HTTPException(status_code=400, detail="子任务不能脱离父任务单独移动") changes = payload.model_dump(exclude_unset=True, exclude={"task_ids", "soft_delete"}) if payload.soft_delete: changes["deleted_at"] = utcnow() if changes: changes["version"] = Task.version + 1 changes["updated_at"] = utcnow() target_ids = set(task_ids) if payload.soft_delete: parent_ids = [task.id for task in tasks if task.parent_id is None] if parent_ids: child_ids = list( ( await db.scalars( select(Task.id).where( Task.parent_id.in_(parent_ids), Task.user_id == user.id, Task.deleted_at.is_(None), ) ) ).all() ) target_ids.update(child_ids) await db.execute( update(Task) .where(Task.id.in_(target_ids), Task.user_id == user.id, Task.deleted_at.is_(None)) .values(**changes) ) if payload.list_id is not None: parent_ids = [task.id for task in tasks if task.parent_id is None] if parent_ids: await db.execute( update(Task) .where( Task.parent_id.in_(parent_ids), Task.user_id == user.id, Task.deleted_at.is_(None), ) .values( list_id=payload.list_id, version=Task.version + 1, updated_at=utcnow(), ) ) await db.commit() return BatchResult(updated=len(task_ids)) static_dir = Path(__file__).parent / "static" if static_dir.exists(): app.mount("/assets", StaticFiles(directory=static_dir / "assets"), name="assets") @app.get("/{path:path}", include_in_schema=False) async def spa(path: str): if path == "api" or path.startswith("api/"): raise HTTPException(status_code=404, detail="Not Found") root = static_dir.resolve() target = (root / path).resolve() headers = {"Cache-Control": "no-cache, no-store, must-revalidate, max-age=0"} if target.is_file() and target.is_relative_to(root): if target.name == "sw.js": return HTMLResponse( "self.addEventListener('install',()=>self.skipWaiting());" "self.addEventListener('activate',e=>e.waitUntil(" "caches.keys().then(k=>Promise.all(k.map(x=>caches.delete(x))))" ".then(()=>self.registration.unregister())" ".then(()=>self.clients.matchAll({type:'window'}))" ".then(cs=>cs.forEach(c=>c.navigate(c.url)))));", media_type="application/javascript", headers=headers, ) return FileResponse(target) return FileResponse(root / "index.html", headers=headers)