import json import logging import os from datetime import date, datetime, time, timedelta from typing import List from fastapi import APIRouter, Depends, HTTPException from fastapi.responses import RedirectResponse from pydantic import BaseModel from .. import db, oss_store from .auth import get_current_user logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/v1/plan", tags=["plan"]) PLAN_TYPES = {"weekly", "monthly", "yearly"} PLAN_STATUSES = {"已完成", "未完成"} DEFAULT_STATUS = "未完成" USER_STATUS_NORMAL = "正常" # auth_users.status，名单只统计状态正常的学员 MAX_CONTENT_LEN = 1000 MAX_FILE_SIZE = 10 * 1024 * 1024 MAX_FILES = 4 ALLOWED_IMAGE_EXT = {".png", ".jpg", ".jpeg", ".gif", ".webp", ".bmp"} SELECT_COLS = "id,user_id,plan_type,period_key,status,checked_at,content,attachments,created_at,updated_at" def iso_week_key(day): iso = day.isocalendar() return "%d-W%02d" % (iso[0], iso[1]) def parse_iso_week(period_key): parts = period_key.split("-W") if len(parts) != 2: return None try: year = int(parts[0]) week = int(parts[1]) except ValueError as exc: logger.warning("解析 ISO 周周期失败: %s", exc) return None jan4 = date(year, 1, 4) monday = jan4 - timedelta(days=jan4.isoweekday() - 1) return monday + timedelta(weeks=week - 1) def default_weekly_period(today=None): today = today or date.today() # 周五及以后（周五、周六、周日）默认填写下周计划 # 周一到周四默认填写本周计划 if today.weekday() >= 4: return iso_week_key(today + timedelta(days=7)) return iso_week_key(today) def weekly_window(period_key): monday = parse_iso_week(period_key) if not monday: return None, None # 开启：目标周的上一周周五 00:00 open_at = datetime.combine(monday - timedelta(days=3), time.min) # 截止：目标周的周四 23:59:59 deadline = datetime.combine(monday + timedelta(days=3), time(23, 59, 59)) return open_at, deadline def weekly_submit_state(period_key): open_at, deadline = weekly_window(period_key) if not open_at or not deadline: return False, "无效的周周期" now = datetime.now() if now < open_at: return False, "周五开始填写" if now > deadline: return False, "周四截止，已过提交时间" return True, "" PERIOD_START_YEAR = 2022 def parse_month_key(period_key): """把月周期 key 解析为该月 1 日；非法返回 None。""" if not _validate_month_key(period_key): return None year, month = period_key.split("-") return date(int(year), int(month), 1) def monthly_window(period_key): """月计划填写窗口：上个月最后 5 天 00:00 ~ 本月第 10 天 23:59:59。 例：2026-09 的窗口 = 2026-08-27 00:00 ~ 2026-09-10 23:59:59。 """ first_day = parse_month_key(period_key) if not first_day: return None, None prev_month_last = first_day - timedelta(days=1) # 上个月最后一天 open_at = datetime.combine(prev_month_last - timedelta(days=4), time.min) deadline = datetime.combine(first_day + timedelta(days=9), time(23, 59, 59)) return open_at, deadline def yearly_window(period_key): """年计划填写窗口：上一年 12 月 1 日 00:00 ~ 本年 1 月 31 日 23:59:59。 例：2026 的窗口 = 2025-12-01 00:00 ~ 2026-01-31 23:59:59。 """ if not _validate_year_key(period_key): return None, None year = int(period_key) open_at = datetime(year - 1, 12, 1, 0, 0, 0) deadline = datetime(year, 1, 31, 23, 59, 59) return open_at, deadline def monthly_submit_state(period_key): """月计划提交状态：窗口内（上月最后 5 天 ~ 本月第 10 天）可提交，其余时间不可。""" open_at, deadline = monthly_window(period_key) if not open_at or not deadline: return False, "无效的月周期" now = datetime.now() if now < open_at: return False, "未到提交时间" if now > deadline: return False, "已过提交时间" return True, "" def yearly_submit_state(period_key): """年计划提交状态：窗口内（上一年 12 月 ~ 本年 1 月）可提交，其余时间不可。""" open_at, deadline = yearly_window(period_key) if not open_at or not deadline: return False, "无效的年周期" now = datetime.now() if now < open_at: return False, "未到提交时间" if now > deadline: return False, "已过提交时间" return True, "" def generate_period_list(plan_type, start_year: int = PERIOD_START_YEAR): """ 生成从 start_year 到当前的所有自然周期列表，倒序排列（最新在前）。 - weekly: ISO 周格式 YYYY-Www - monthly: YYYY-MM - yearly: YYYY 三种类型的终点统一取「当前应填周期」而非「今天所在周期」：周计划周五起填下一周， 月计划上月末 5 天起填下一月，年计划上年 12 月起填下一年。终点若不随之延伸， current_period 会落在列表之外，前端选不到、也就无法提交。 """ today = date.today() if plan_type == "weekly": weeks = [] # 从 start_year 第一个 ISO 周开始 jan4 = date(start_year, 1, 4) first_monday = jan4 - timedelta(days=jan4.isoweekday() - 1) # 截止到「当前应填周期」（周一~周四为本周期，周五~周日为下一周期） end_monday = parse_iso_week(default_weekly_period(today)) if end_monday is None: # 兜底：理论上不会发生 iso_today = today.isocalendar() base_jan4 = date(iso_today[0], 1, 4) end_monday = (base_jan4 - timedelta(days=base_jan4.isoweekday() - 1) + timedelta(weeks=iso_today[1] - 1)) current = first_monday while current <= end_monday: weeks.append(iso_week_key(current)) current += timedelta(weeks=1) weeks.reverse() return weeks if plan_type == "monthly": months = [] # 终点取「当前应填周期」（窗口开放月，可能是下个月） end_key = default_monthly_period(today) end_year, end_month = int(end_key[:4]), int(end_key[5:7]) for y in range(start_year, end_year + 1): for m in range(1, 13): if y == end_year and m > end_month: break months.append("%04d-%02d" % (y, m)) months.reverse() return months if plan_type == "yearly": years = [] # 终点取「当前应填周期」（窗口开放年，可能是下一年） end_year = int(default_yearly_period(today)) for y in range(start_year, end_year + 1): years.append(str(y)) years.reverse() return years return [] def _month_keys_around(today): """返回 (上上月 … 下下月) 的月周期 key，按时间升序。""" keys = [] for offset in (-2, -1, 0, 1, 2): total = today.year * 12 + (today.month - 1) + offset year, month = divmod(total, 12) keys.append("%04d-%02d" % (year, month + 1)) return keys def default_monthly_period(today=None): """月计划应填周期：窗口正在开放的那个月；都不开放时取最近将要开放的月。""" today = today or date.today() now = datetime.now() next_open = None for key in _month_keys_around(today): open_at, deadline = monthly_window(key) if not open_at: continue if open_at <= now <= deadline: return key if open_at > now and (next_open is None or open_at < next_open[0]): next_open = (open_at, key) if next_open: return next_open[1] return "%04d-%02d" % (today.year, today.month) def default_yearly_period(today=None): """年计划应填周期：窗口正在开放的那一年；都不开放时取最近将要开放的年。""" today = today or date.today() now = datetime.now() next_open = None for key in (str(today.year - 1), str(today.year), str(today.year + 1), str(today.year + 2)): open_at, deadline = yearly_window(key) if not open_at: continue if open_at <= now <= deadline: return key if open_at > now and (next_open is None or open_at < next_open[0]): next_open = (open_at, key) if next_open: return next_open[1] return str(today.year) def get_period_key(plan_type): today = date.today() if plan_type == "weekly": return default_weekly_period(today) if plan_type == "monthly": return default_monthly_period(today) if plan_type == "yearly": return default_yearly_period(today) return "" def fmt_dt(value): if isinstance(value, datetime): return value.strftime("%Y-%m-%d %H:%M:%S") return str(value) if value else "" def _parse_attachments(raw): if isinstance(raw, str): try: return json.loads(raw) except Exception: logger.exception("计划附件字段解析失败") return [] return raw or [] ATTACH_PREFIX = "attachments/" OSS_PLAN_PREFIX = "plan/" def oss_key(rel_path): """规范相对路径 -> OSS 对象 key。 `attachments/{user_id}/{plan_type}/{period_key}/{filename}` 对应的对象 key 是 `plan/{user_id}/{plan_type}/{period_key}/{filename}` —— 只差一个前缀， 所以从磁盘换成 OSS 时**数据库里的相对路径一个字都不用改**。 外链（http/https）不是本桶的对象，返回空串。 """ if not isinstance(rel_path, str) or not rel_path.startswith(ATTACH_PREFIX): return "" tail = rel_path[len(ATTACH_PREFIX):].strip("/") if not tail or ".." in tail: return "" return OSS_PLAN_PREFIX + tail def media_url(rel_path): """把规范相对路径换成可直接喂给 `` 的地址。 - 外链（http/https）：原样返回。库里确实存在外链图片（小鹅通导入那批），必须放行。 - OSS 未启用 / key 非法：空串。前端**已经没有静态地址可退**，那张图就是不显示 —— 这是刻意的：图片只存在于 OSS，退回 nginx 静态地址只会得到一张 404 的裂图。 - 其余：OSS 签名 URL（纯本地计算，不发网络请求，可放心循环批量签）。 """ if not isinstance(rel_path, str) or not rel_path.strip(): return "" text = rel_path.strip() if text.startswith(("http://", "https://")): return text key = oss_key(text) if not key or not oss_store.is_enabled(): return "" return oss_store.signed_url(key) def media_urls(paths): """批量签名，**长度与顺序和入参严格一致**，签不出来的位置留空串。 保持等长是有意的：前端按同一个下标去取 `attachment_urls[i]`， 一旦长度不等就会出现"缩略图和正文错位"这种很难查的问题。 """ return [media_url(p) for p in (paths or [])] def fmt_rec(rec): files = _parse_attachments(rec.get("attachments")) return { "id": rec["id"], "user_id": rec["user_id"], "plan_type": rec["plan_type"], "period_key": rec["period_key"], "status": rec.get("status", DEFAULT_STATUS), "checked_at": fmt_dt(rec.get("checked_at")), "content": rec.get("content", ""), # attachments 是**规范值**，会被 /plan/checkin 原样回写进库，格式不许变。 "attachments": files, # 展示专用的签名 URL，与 attachments 一一对应。 # 为什么另开一个字段而不是直接把 attachments 换成 URL：小程序把 # this.data.attachments 原样提交回 /plan/checkin，一旦这里返回签名 URL， # 用户保存一次打卡就会把带 Expires/Signature 的地址写进数据库。 "attachment_urls": media_urls(files), "created_at": fmt_dt(rec.get("created_at")), "updated_at": fmt_dt(rec.get("updated_at")), } class PlanCheckinRequest(BaseModel): plan_type: str = "" period_key: str = "" status: str = DEFAULT_STATUS content: str = "" attachments: List[str] = [] @router.get("/periods") def periods(type: str, user=Depends(get_current_user)): """ 返回指定计划类型下的全部自然周期列表（从 2022 年到当前，倒序）。 用于前端做周期切换选择器。 """ if type not in PLAN_TYPES: raise HTTPException(status_code=400, detail="无效的计划类型") today = date.today() period_list = generate_period_list(type) current = get_period_key(type) resp = { "type": type, "periods": period_list, "current_period": current, "period_labels": {}, } if type == "weekly": # 「本周 / 下周」：周五~周日填写的是下一周期 resp["current_week"] = iso_week_key(today) resp["period_labels"] = { iso_week_key(today): "本周", iso_week_key(today + timedelta(days=7)): "下周", } elif type == "monthly": # 「本月 / 下月」：上月末 5 天起填写的是下一周期 next_year, next_month = divmod(today.year * 12 + today.month, 12) resp["period_labels"] = { "%04d-%02d" % (today.year, today.month): "本月", "%04d-%02d" % (next_year, next_month + 1): "下月", } elif type == "yearly": # 「今年 / 明年」：上年 12 月起填写的是下一周期 resp["period_labels"] = {str(today.year): "今年", str(today.year + 1): "明年"} return resp @router.get("/status") def status(type: str, period_key: str = "", user=Depends(get_current_user)): if type not in PLAN_TYPES: raise HTTPException(status_code=400, detail="无效的计划类型") selected, can_submit, message = _normalize_period(type, period_key) rec = db.fetch_one( "SELECT " + SELECT_COLS + " FROM plan_checkins WHERE user_id=%s AND plan_type=%s AND period_key=%s", (user["id"], type, selected), ) return {"plan_type": type, "period_key": selected, "checked": rec is not None, "can_submit": can_submit, "message": message, "record": fmt_rec(rec) if rec else None} def _is_teacher(user): """判断用户是否为老师角色。""" return (user.get("role") or "").strip().lower() == "teacher" def _validate_month_key(period_key: str) -> bool: """校验月周期格式 YYYY-MM。""" if not period_key: return False parts = period_key.split("-") if len(parts) != 2: return False try: year = int(parts[0]) month = int(parts[1]) return 2020 <= year <= 2100 and 1 <= month <= 12 except ValueError: return False def _validate_year_key(period_key: str) -> bool: """校验年周期格式 YYYY。""" if not period_key or len(period_key) != 4: return False try: year = int(period_key) return 2020 <= year <= 2100 except ValueError: return False def _normalize_period(plan_type, period_key): """ 规范化周期 key，做基础校验（不限制必须是当前周期）。 用于查看类接口：允许查看任意历史周期。 Returns (selected, can_submit, message); 校验失败抛出 400。 """ selected = period_key.strip() or get_period_key(plan_type) if plan_type == "weekly": if parse_iso_week(selected) is None: raise HTTPException(status_code=400, detail="无效的周周期") can_submit, message = weekly_submit_state(selected) elif plan_type == "monthly": if not _validate_month_key(selected): raise HTTPException(status_code=400, detail="无效的月周期") can_submit, message = monthly_submit_state(selected) elif plan_type == "yearly": if not _validate_year_key(selected): raise HTTPException(status_code=400, detail="无效的年周期") can_submit, message = yearly_submit_state(selected) else: can_submit, message = (True, "") return selected, can_submit, message def _groups_stats(plan_type, selected, group_filter=None): """单条 JOIN 聚合 SQL：提交计划即已完成，未提交即未完成。 按小组统计总人数 / 已完成(有记录)/ 未完成(无记录)，避免 N+1 查询。 group_filter: 只统计指定小组 """ where_clauses = [] params = [plan_type, selected] if group_filter: where_clauses.append("u.`group` = %s") params.append(group_filter) else: where_clauses.append("u.`group` IS NOT NULL AND u.`group` <> ''") where_sql = ("WHERE " + " AND ".join(where_clauses)) if where_clauses else "" rows = db.fetch_all( "SELECT u.`group` AS group_name, " "COUNT(u.id) AS total, " "SUM(CASE WHEN pc.user_id IS NOT NULL THEN 1 ELSE 0 END) AS checked_count " "FROM auth_users u " "LEFT JOIN plan_checkins pc ON pc.user_id = u.id " "AND pc.plan_type = %s AND pc.period_key = %s " + where_sql + " " "GROUP BY u.`group` ORDER BY u.`group`", tuple(params), ) groups_list = [] for item in rows or []: total = item.get("total") or 0 checked = item.get("checked_count") or 0 groups_list.append({ "group_name": item.get("group_name") or "", "total": total, "checked_count": checked, "unchecked_count": total - checked, }) return groups_list @router.get("/groups") def groups(type: str, period_key: str = "", user=Depends(get_current_user)): if type not in PLAN_TYPES: raise HTTPException(status_code=400, detail="无效的计划类型") selected, can_submit, message = _normalize_period(type, period_key) my_rec = db.fetch_one( "SELECT " + SELECT_COLS + " FROM plan_checkins " "WHERE user_id=%s AND plan_type=%s AND period_key=%s", (user["id"], type, selected), ) my_checked = my_rec is not None my_status = (my_rec or {}).get("status", "") is_teacher = _is_teacher(user) my_group = (user.get("group") or "").strip() if is_teacher: # 老师：查看所有小组 groups_list = _groups_stats(type, selected) if my_group: mine = [g for g in groups_list if g["group_name"] == my_group] others = [g for g in groups_list if g["group_name"] != my_group] groups_list = mine + others else: # 学员：只看自己所在小组 if not my_group: groups_list = [] else: groups_list = _groups_stats(type, selected, group_filter=my_group) # 未分组人数仅老师可见 unassigned_count = 0 if is_teacher: unassigned_row = db.fetch_one( "SELECT COUNT(*) AS cnt FROM auth_users WHERE `group` IS NULL OR `group` = ''" ) unassigned_count = unassigned_row["cnt"] if unassigned_row else 0 return { "type": type, "period_key": selected, "can_submit": can_submit, "message": message, "my_checked": my_checked, "my_status": my_status, "my_record": fmt_rec(my_rec) if my_rec else None, "is_teacher": is_teacher, "my_group": my_group, "unassigned_count": unassigned_count, "groups": groups_list, } @router.get("/group") def group(type: str, group_name: str, period_key: str = "", user=Depends(get_current_user)): if type not in PLAN_TYPES: raise HTTPException(status_code=400, detail="无效的计划类型") selected, can_submit, message = _normalize_period(type, period_key) group_name = group_name.strip() if not group_name: raise HTTPException(status_code=400, detail="缺少小组名") group_row = db.fetch_one("SELECT `group` FROM auth_users WHERE `group`=%s LIMIT 1", (group_name,)) if not group_row: raise HTTPException(status_code=404, detail="小组不存在") members = db.fetch_all( "SELECT id AS user_id, student_id, name FROM auth_users WHERE `group`=%s ORDER BY id", (group_name,), ) uids = [m["user_id"] for m in members] rec_map = {} if uids: placeholders = ",".join(["%s"] * len(uids)) records = db.fetch_all( "SELECT user_id, status, checked_at, content, attachments FROM plan_checkins " "WHERE plan_type=%s AND period_key=%s AND user_id IN (" + placeholders + ")", tuple([type, selected] + uids), ) rec_map = {r["user_id"]: r for r in records} result = [] for m in members: rec = rec_map.get(m["user_id"]) record = None if rec: files = _parse_attachments(rec.get("attachments")) record = { "status": rec["status"], "checked_at": fmt_dt(rec.get("checked_at")), "content": rec.get("content") or "", "attachments": files, "attachment_urls": media_urls(files), } result.append({ "user_id": m["user_id"], "student_id": m.get("student_id") or "", "name": m["name"], "filled": rec is not None, "record": record, }) result.sort(key=lambda item: (0 if not item["filled"] else 1, item["user_id"])) return { "type": type, "period_key": selected, "group_name": group_name, "can_submit": can_submit, "message": message, "members": result, } @router.get("/records") def records(period_key: str = "", user=Depends(get_current_user)): """ 周计划打卡名单（仅老师可查）。 请求参数： - period_key 周（ISO 周，如 `2026-W38`）；不传取当前应填周（周五~周日为下一周） 返回：`user_id` / `student_id` / `name` / `group_name` / `checked` 判定口径：该周在 `plan_checkins` 中存在记录 = 已打卡，与 `/groups` 的小组统计口径一致。 仅统计 `auth_users.status = '正常'` 的学员（`暂停` 等状态不计入名单）。 学员（非 teacher）一律拒绝。 """ if not _is_teacher(user): raise HTTPException(status_code=403, detail="无权限") selected = period_key.strip() or get_period_key("weekly") if parse_iso_week(selected) is None: raise HTTPException(status_code=400, detail="无效的周周期") # 用户表 LEFT JOIN 周计划打卡记录表，一次取回全员（含未打卡）避免 N+1 rows = db.fetch_all( "SELECT u.id AS user_id, u.student_id, u.name, u.`group` AS group_name, " " pc.id AS record_id " "FROM auth_users u " "LEFT JOIN plan_checkins pc ON pc.user_id = u.id " " AND pc.plan_type = %s AND pc.period_key = %s " "WHERE u.`status` = %s " "ORDER BY u.`group`, u.student_id", ("weekly", selected, USER_STATUS_NORMAL), ) return { "period_key": selected, "list": [ { "user_id": r["user_id"], "student_id": r.get("student_id") or "", "name": r.get("name") or "", "group_name": r.get("group_name") or "", "checked": r.get("record_id") is not None, } for r in rows or [] ], } @router.post("/checkin") def checkin(req: PlanCheckinRequest, user=Depends(get_current_user)): plan_type = req.plan_type.strip() if plan_type not in PLAN_TYPES: raise HTTPException(status_code=400, detail="无效的计划类型") period_key = req.period_key.strip() or get_period_key(plan_type) if plan_type == "weekly": if parse_iso_week(period_key) is None: raise HTTPException(status_code=400, detail="无效的周周期") default_period = default_weekly_period(date.today()) if period_key != default_period: raise HTTPException(status_code=400, detail="仅可填写当前周期") can_submit, message = weekly_submit_state(default_period) if not can_submit: raise HTTPException(status_code=400, detail=message) elif plan_type == "monthly": first_day = parse_month_key(period_key) if first_day is None: raise HTTPException(status_code=400, detail="无效的月周期") # 归一化为规范格式，避免 2026-9 / 2026-09 在同一逻辑月产生两行 period_key = "%04d-%02d" % (first_day.year, first_day.month) if period_key != default_monthly_period(date.today()): raise HTTPException(status_code=400, detail="仅可填写当前周期") can_submit, message = monthly_submit_state(period_key) if not can_submit: raise HTTPException(status_code=400, detail=message) elif plan_type == "yearly": if not _validate_year_key(period_key): raise HTTPException(status_code=400, detail="无效的年周期") if period_key != default_yearly_period(date.today()): raise HTTPException(status_code=400, detail="仅可填写当前周期") can_submit, message = yearly_submit_state(period_key) if not can_submit: raise HTTPException(status_code=400, detail=message) status = req.status.strip() if status not in PLAN_STATUSES: status = DEFAULT_STATUS content = req.content.strip() if len(content) > MAX_CONTENT_LEN: content = content[:MAX_CONTENT_LEN] # 落库前的最后一道闸：直传模式下客户端能往自己目录塞任意后缀的文件， # 签名只能锁目录、锁不住扩展名，所以必须在这里再筛一遍。 attachments = [a for a in (req.attachments or []) if _is_allowed_attachment(a)][:MAX_FILES] attachments_json = json.dumps(attachments, ensure_ascii=False) now = datetime.now() db.execute( "INSERT INTO plan_checkins (user_id, plan_type, period_key, status, checked_at, content, attachments) " "VALUES (%s, %s, %s, %s, %s, %s, %s) " "ON DUPLICATE KEY UPDATE status=VALUES(status), checked_at=VALUES(checked_at), " "content=VALUES(content), attachments=VALUES(attachments)", (user["id"], plan_type, period_key, status, now, content, attachments_json), ) rec = db.fetch_one( "SELECT " + SELECT_COLS + " FROM plan_checkins WHERE user_id=%s AND plan_type=%s AND period_key=%s", (user["id"], plan_type, period_key), ) return {"success": True, "record": fmt_rec(rec)} @router.get("/history") def history(type: str, page: int = 1, size: int = 10, user=Depends(get_current_user)): if type not in PLAN_TYPES: raise HTTPException(status_code=400, detail="无效的计划类型") page = max(1, page) size = max(1, min(50, size)) offset = (page - 1) * size total_row = db.fetch_one( "SELECT COUNT(*) AS cnt FROM plan_checkins WHERE user_id=%s AND plan_type=%s", (user["id"], type), ) total = total_row["cnt"] if total_row else 0 rows = db.fetch_all( "SELECT " + SELECT_COLS + " FROM plan_checkins WHERE user_id=%s AND plan_type=%s " "ORDER BY period_key DESC LIMIT %s OFFSET %s", (user["id"], type, size, offset), ) return {"list": [fmt_rec(r) for r in rows], "total": total, "page": page, "size": size} def _safe_period_key(value): """period_key 会直接拼进 OSS 对象 key，必须挡住路径穿越。 传 `../../..` 就能把文件写到预期目录之外 —— 需求文档 V3.0 列为待修项。 非法值一律当作"没传"，回落到当前周期，而不是报错中断（保持旧调用方兼容）。 """ text = (value or "").strip() if not text or "/" in text or "\\" in text or ".." in text: return "" return text def _is_allowed_attachment(rel_path): """落库前的最后一道闸：只收「规范相对路径 + 白名单扩展名」。 ⚠️ 这道校验在直传模式下**不能省**。PostObject 的 policy 用 `starts-with` 锁住了目录，但**锁不住扩展名** —— 一个被改过的客户端可以往自己的目录里 POST 任意后缀的文件（比如 .html）。所以真正决定「什么能进 attachments 字段」的地方是这里，而不是前端、也不是签名接口。 """ if not isinstance(rel_path, str): return False text = rel_path.strip() if not text: return False if text.startswith(("http://", "https://")): return True # 库里确实存在外链图（小鹅通导入那批），必须放行 if not oss_key(text): return False # 非 attachments/ 前缀、含 `..` 的都在这里被挡掉 return os.path.splitext(text)[1].lower() in ALLOWED_IMAGE_EXT class UploadSignatureRequest(BaseModel): plan_type: str = "" period_key: str = "" @router.post("/upload-signature") def upload_signature(req: UploadSignatureRequest, user=Depends(get_current_user)): """给「小程序直传 OSS」签一份 PostObject 表单签名。 直传后图片字节只走「小程序 → OSS」，服务器带宽为零；后端只签发这一小段 policy。 安全要点（改前先读 `oss_store.post_form` 的注释）： 1. **key 前缀由服务端用 session 里的 user_id 拼，前端无法指定。** 若让前端传 key，它就能写到 `plan/{别人的 uid}/...` 覆盖他人附件。 2. 返回的是**目录级**签名（`plan/{uid}/{type}/{period}/`），前端在该目录内 自行命名。原因是小程序会先压缩再上传，而压缩会换掉文件名 —— 签完整 key 就必须提前知道最终文件名，做不到「一次签名、多张复用」。 3. 扩展名**不靠签名兜**，而靠 `/plan/checkin` 落库时再校验一次 （`_is_allowed_attachment`）。所以那道校验不能删。 未配置 OSS 时返回 `enabled=false`，前端据此提示「上传服务不可用」—— 这不是错误分支，而是设计好的提示路径（没有中转兜底，图片只走直传）。 """ plan_type = (req.plan_type or "").strip() if plan_type not in PLAN_TYPES: raise HTTPException(status_code=400, detail="无效的计划类型") if not oss_store.is_enabled(): return {"success": True, "enabled": False} period_key = _safe_period_key(req.period_key) or get_period_key(plan_type) tail = str(user["id"]) + "/" + plan_type + "/" + period_key + "/" form = oss_store.post_form(OSS_PLAN_PREFIX + tail, max_size=MAX_FILE_SIZE) return { "success": True, "enabled": True, "rel_prefix": ATTACH_PREFIX + tail, "key_prefix": form["key_prefix"], "host": form["host"], "expires_in": form["expires_in"], "form_data": form["form_data"], } @router.get("/file/{sid}/{plan_type}/{period}/{filename}") def get_file(sid: str, plan_type: str, period: str, filename: str, user=Depends(get_current_user)): """旧版取图接口：302 到 OSS 签名地址。 ⚠️ 2026-09-22 之前这个接口**没有任何鉴权**（紧邻的 DELETE 反而有）， 只要按 {sid}/{plan_type}/{period}/{filename} 猜路径就能下载任意学员的打卡图， 需求文档 V3.0 把它列为高风险项。同日补上了登录校验。 图片现在只在 OSS 上，所以这里不再读磁盘，直接 302 —— 只是转个地址， 图片字节不经过服务器。小程序取图实际不走这个接口，而是用读接口并排返回的 `attachment_urls`；保留它只为兼容已经发出去的老地址。未配 OSS 时返回 404。 """ safe_name = os.path.basename(filename) key = oss_key("attachments/" + sid + "/" + plan_type + "/" + period + "/" + safe_name) if key and oss_store.is_enabled(): return RedirectResponse(oss_store.signed_url(key), status_code=302) raise HTTPException(status_code=404, detail="not found") @router.delete("/file/{sid}/{plan_type}/{period}/{filename}") def delete_file(sid: str, plan_type: str, period: str, filename: str, user=Depends(get_current_user)): """删除附件对象。只能删自己的。 OSS 的 delete 是幂等的（删不存在的 key 也算成功），所以不需要先查存在性。 """ if str(sid) != str(user["id"]): raise HTTPException(status_code=403, detail="无权限") safe_name = os.path.basename(filename) key = oss_key("attachments/" + sid + "/" + plan_type + "/" + period + "/" + safe_name) if key and oss_store.is_enabled(): try: oss_store.delete(key) except Exception: logger.exception("删除 OSS 对象失败: %s", key) return {"success": True}