"""每周重点工作:单条录入、按周跟踪的短期任务。 和项目管理(Project + Update)分开:不用先建项目档案,一条就是一个活儿。 字段按需求——工作内容 / 说明 / 预计完成时间(日历日期)/ 执行状态 / 结果反馈 / 是否延期。 period_id 是「录入周」——新建时顶栏选中的那一周,和项目管理/定制项目的周期归属同一个意思; 预计完成时间只存 due_date,不再按日期反查周期。本周录一条两周后要干的事,本周照样看得见。 「是否延期」是人工开关(delayed):更早周次录入、标了延期又没做完的,结转进当周视图。 录入整理在「每周重点工作」页,汇总在「工作看板」。 """ from __future__ import annotations from datetime import date from fastapi import APIRouter, Depends, HTTPException, Query from sqlalchemy.orm import Session from .. import constants from ..db import get_db from ..models import Period, WeeklyFocus from ..schemas import FocusIn, FocusPatch router = APIRouter() def _week_period(db: Session, year: int, month: int, week: int) -> Period: """取 periods 里的这一周,没有就补一条。""" sort_key = year * 10000 + month * 100 + week target = ( db.query(Period) .filter(Period.kind == "week", Period.sort_key == sort_key) .one_or_none() ) if target is None: target = Period( label=f"{month}月第{week}周", kind="week", year=year, month=month, week=week, sort_key=sort_key, ) db.add(target) db.flush() return target def _entry_period(db: Session, period_id: int | None) -> Period: """录入周:前端传顶栏当前选中的那一周;没传(脚本/API 直接调)就按今天所在的周。""" if period_id: existed = db.get(Period, period_id) if existed: return existed raise HTTPException(400, "录入周不存在") return _week_period(db, *constants.week_of_date(date.today())) def _clean_date(value: str | None) -> str | None: """校验日历日期,存 ISO 字符串。""" raw = (value or "").strip() if not raw: return None try: return date.fromisoformat(raw[:10]).isoformat() except ValueError: raise HTTPException(400, f"预计完成时间格式不对:{raw}(应为 YYYY-MM-DD)") @router.get("/focus") def list_focus( archived: str = Query("false", pattern="^(true|false|all)$"), status: str | None = None, keyword: str | None = None, db: Session = Depends(get_db), ): q = db.query(WeeklyFocus) if archived == "false": q = q.filter(WeeklyFocus.archived.is_(False)) elif archived == "true": q = q.filter(WeeklyFocus.archived.is_(True)) if status: q = q.filter(WeeklyFocus.status == status) if keyword: q = q.filter(WeeklyFocus.content.contains(keyword) | WeeklyFocus.detail.contains(keyword)) return [f.to_dict() for f in q.order_by(WeeklyFocus.id.desc()).all()] @router.post("/focus") def create_focus(payload: FocusIn, db: Session = Depends(get_db)): content = (payload.content or "").strip() if not content: raise HTTPException(400, "请填写工作内容") if payload.status not in constants.STATUS_CODES: raise HTTPException(400, f"未知状态:{payload.status}") due = _clean_date(payload.due_date) item = WeeklyFocus( content=content, detail=(payload.detail or "").strip() or None, period_id=_entry_period(db, payload.period_id).id, due_date=due, result=(payload.result or "").strip() or None, delayed=bool(payload.delayed), status=payload.status, ) db.add(item) db.commit() db.refresh(item) return item.to_dict() @router.patch("/focus/{fid}") def patch_focus(fid: int, payload: FocusPatch, db: Session = Depends(get_db)): item = db.get(WeeklyFocus, fid) if not item: raise HTTPException(404, "条目不存在") data = payload.model_dump(exclude_unset=True) if "content" in data: content = (data["content"] or "").strip() if not content: raise HTTPException(400, "请填写工作内容") data["content"] = content if "detail" in data: data["detail"] = (data["detail"] or "").strip() or None if "status" in data and data["status"] not in constants.STATUS_CODES: raise HTTPException(400, f"未知状态:{data['status']}") if "result" in data: data["result"] = (data["result"] or "").strip() or None if "due_date" in data: data["due_date"] = _clean_date(data["due_date"]) # 改预计完成时间不再把条目搬去别的周,只有显式指定录入周才动 period_id if data.get("period_id"): data["period_id"] = _entry_period(db, data["period_id"]).id else: data.pop("period_id", None) for key, value in data.items(): setattr(item, key, value) db.commit() db.refresh(item) return item.to_dict() @router.delete("/focus/{fid}") def delete_focus(fid: int, db: Session = Depends(get_db)): item = db.get(WeeklyFocus, fid) if not item: raise HTTPException(404, "条目不存在") content = item.content db.delete(item) db.commit() return {"id": fid, "content": content}