9a7681112b
- 修改数据库配置将MYSQL_DB从mys_info_v3改为mys_info_test - 新增auth_users表wechat_openid字段并建立唯一索引 - 为plan_checkins表添加计划性能索引(idx_plan_period_student) - 将checkin_records_old表关联方式从student_id改为user_id - 将plan_checkins表关联方式从student_id改为user_id并更新附件路径 - 移除routes/auth.py中的uuid导入使用安全token生成函数 - 重构routes/checkin.py打卡逻辑去除course_type字段并优化返回结构 - 添加打卡时间限制验证功能防止过期打卡操作 - 实现小组统计功能支持老师查看各组打卡情况 - 更新主页index.html界面适配新的打卡和计划功能 - 调整计划提交时间规则为周五至周四周期 - 添加月度和年度计划提交状态验证机制
585 lines
20 KiB
Python
585 lines
20 KiB
Python
import json
|
||
import logging
|
||
import os
|
||
from datetime import date, datetime, time, timedelta
|
||
from typing import List
|
||
|
||
from fastapi import APIRouter, Depends, File, Form, HTTPException, UploadFile
|
||
from fastapi.responses import FileResponse
|
||
from pydantic import BaseModel
|
||
|
||
from .. import config, db
|
||
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 = "未完成"
|
||
MAX_CONTENT_LEN = 1000
|
||
MAX_FILE_SIZE = 10 * 1024 * 1024
|
||
MAX_FILES = 4
|
||
ALLOWED_IMAGE_EXT = {".png", ".jpg", ".jpeg", ".gif", ".webp", ".bmp"}
|
||
ATTACH_DIR = config.PLAN_ATTACH_DIR
|
||
SELECT_COLS = "id,user_id,plan_type,period_key,status,checked_at,content,attachments,created_at,updated_at"
|
||
|
||
CONTENT_TYPES = {
|
||
"pdf": "application/pdf", "png": "image/png", "jpg": "image/jpeg",
|
||
"jpeg": "image/jpeg", "gif": "image/gif", "webp": "image/webp",
|
||
"bmp": "image/bmp", "doc": "application/msword",
|
||
"docx": "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
|
||
}
|
||
|
||
|
||
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 monthly_submit_state(period_key):
|
||
"""月计划提交状态：当前月内可提交，历史月/未来月均不可提交。"""
|
||
if not _validate_month_key(period_key):
|
||
return False, "无效的月周期"
|
||
today = date.today()
|
||
year, month = period_key.split("-")
|
||
period_year, period_month = int(year), int(month)
|
||
if period_year < today.year or (period_year == today.year and period_month < today.month):
|
||
return False, "已过提交时间"
|
||
if period_year > today.year or (period_year == today.year and period_month > today.month):
|
||
return False, "未到提交时间"
|
||
return True, ""
|
||
|
||
|
||
def yearly_submit_state(period_key):
|
||
"""年计划提交状态：当前年内可提交，往年/未来年均不可提交。"""
|
||
if not _validate_year_key(period_key):
|
||
return False, "无效的年周期"
|
||
today = date.today()
|
||
period_year = int(period_key)
|
||
if period_year < today.year:
|
||
return False, "已过提交时间"
|
||
if period_year > today.year:
|
||
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
|
||
"""
|
||
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)
|
||
# 截止到当前 ISO 周
|
||
iso_today = today.isocalendar()
|
||
last_jan4 = date(iso_today[0], 1, 4)
|
||
last_monday = last_jan4 - timedelta(days=last_jan4.isoweekday() - 1)
|
||
end_monday = last_monday + 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 = []
|
||
for y in range(start_year, today.year + 1):
|
||
for m in range(1, 13):
|
||
if y == today.year and m > today.month:
|
||
break
|
||
months.append("%04d-%02d" % (y, m))
|
||
months.reverse()
|
||
return months
|
||
|
||
if plan_type == "yearly":
|
||
years = []
|
||
for y in range(start_year, today.year + 1):
|
||
years.append(str(y))
|
||
years.reverse()
|
||
return years
|
||
|
||
return []
|
||
|
||
|
||
def get_period_key(plan_type):
|
||
today = date.today()
|
||
if plan_type == "weekly":
|
||
return default_weekly_period(today)
|
||
if plan_type == "monthly":
|
||
return today.strftime("%Y-%m")
|
||
if plan_type == "yearly":
|
||
return today.strftime("%Y")
|
||
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 []
|
||
|
||
|
||
def fmt_rec(rec):
|
||
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": _parse_attachments(rec.get("attachments")),
|
||
"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="无效的计划类型")
|
||
period_list = generate_period_list(type)
|
||
current = get_period_key(type)
|
||
return {
|
||
"type": type,
|
||
"periods": period_list,
|
||
"current_period": current,
|
||
}
|
||
|
||
|
||
@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 _validate_current_period(plan_type, period_key):
|
||
"""Normalize to the current period for the given plan type.
|
||
|
||
Returns (selected, can_submit, message); raises 400 for invalid types/periods.
|
||
"""
|
||
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="无效的周周期")
|
||
if selected != default_weekly_period(date.today()):
|
||
raise HTTPException(status_code=400, detail="仅可查看当前周期")
|
||
can_submit, message = weekly_submit_state(selected)
|
||
else:
|
||
if selected != get_period_key(plan_type):
|
||
raise HTTPException(status_code=400, detail="仅可查看当前周期")
|
||
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, 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 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"])
|
||
result.append({
|
||
"user_id": m["user_id"],
|
||
"name": m["name"],
|
||
"filled": rec is not None,
|
||
"record": {
|
||
"status": rec["status"],
|
||
"checked_at": fmt_dt(rec.get("checked_at")),
|
||
} if rec else None,
|
||
})
|
||
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.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)
|
||
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 if isinstance(a, str) and a.strip()][: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}
|
||
|
||
|
||
@router.post("/upload")
|
||
def upload(
|
||
plan_type: str = Form(...),
|
||
period_key: str = Form(""),
|
||
files: List[UploadFile] = File(...),
|
||
user=Depends(get_current_user),
|
||
):
|
||
plan_type = plan_type.strip()
|
||
if plan_type not in PLAN_TYPES:
|
||
raise HTTPException(status_code=400, detail="无效的计划类型")
|
||
if len(files) > MAX_FILES:
|
||
raise HTTPException(status_code=400, detail="最多上传4个文件")
|
||
period_key = period_key.strip() or get_period_key(plan_type)
|
||
save_dir = os.path.join(ATTACH_DIR, str(user["id"]), plan_type, period_key)
|
||
os.makedirs(save_dir, exist_ok=True)
|
||
saved = []
|
||
for f in files:
|
||
data = f.file.read()
|
||
if len(data) > MAX_FILE_SIZE:
|
||
continue
|
||
safe_name = os.path.basename(f.filename or "")
|
||
if not safe_name:
|
||
continue
|
||
ext = os.path.splitext(safe_name)[1].lower()
|
||
if ext not in ALLOWED_IMAGE_EXT:
|
||
continue
|
||
save_path = os.path.join(save_dir, safe_name)
|
||
with open(save_path, "wb") as fh:
|
||
fh.write(data)
|
||
rel_path = "attachments/" + str(user["id"]) + "/" + plan_type + "/" + period_key + "/" + safe_name
|
||
saved.append(rel_path)
|
||
if not saved:
|
||
raise HTTPException(status_code=400, detail="文件保存失败")
|
||
return {"success": True, "attachments": saved}
|
||
|
||
|
||
@router.get("/file/{sid}/{plan_type}/{period}/{filename}")
|
||
def get_file(sid: str, plan_type: str, period: str, filename: str):
|
||
safe_name = os.path.basename(filename)
|
||
file_path = os.path.join(ATTACH_DIR, sid, plan_type, period, safe_name)
|
||
if not os.path.exists(file_path):
|
||
raise HTTPException(status_code=404, detail="not found")
|
||
ext = os.path.splitext(safe_name)[1].lstrip(".").lower()
|
||
media_type = CONTENT_TYPES.get(ext, "application/octet-stream")
|
||
return FileResponse(file_path, media_type=media_type)
|
||
|
||
|
||
@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)):
|
||
if str(sid) != str(user["id"]):
|
||
raise HTTPException(status_code=403, detail="无权限")
|
||
safe_name = os.path.basename(filename)
|
||
file_path = os.path.join(ATTACH_DIR, sid, plan_type, period, safe_name)
|
||
if os.path.exists(file_path):
|
||
os.remove(file_path)
|
||
return {"success": True}
|