From 2480d4a738e0255f03577f89d4a8e4fac80ca53d Mon Sep 17 00:00:00 2001 From: simonkoson <28867558@qq.com> Date: Fri, 15 May 2026 13:40:13 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20Task=202=20Excel=20=E6=89=B9=E9=87=8F?= =?UTF-8?q?=E5=AF=BC=E5=85=A5=20API=20=E5=AE=8C=E6=88=90=EF=BC=88=E6=A8=A1?= =?UTF-8?q?=E6=9D=BF=E4=B8=8B=E8=BD=BD=20+=20=E6=89=B9=E9=87=8F=E5=AF=BC?= =?UTF-8?q?=E5=85=A5=20+=20=E5=A4=B1=E8=B4=A5=E8=A1=8C=E4=B8=8B=E8=BD=BD?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/api/imports.py | 172 ++++++++++++++++++ backend/app/main.py | 4 +- backend/app/schemas/imports.py | 27 +++ backend/app/services/excel_service.py | 246 ++++++++++++++++++++++++++ 4 files changed, 448 insertions(+), 1 deletion(-) create mode 100644 backend/app/api/imports.py create mode 100644 backend/app/schemas/imports.py create mode 100644 backend/app/services/excel_service.py diff --git a/backend/app/api/imports.py b/backend/app/api/imports.py new file mode 100644 index 0000000..b08664f --- /dev/null +++ b/backend/app/api/imports.py @@ -0,0 +1,172 @@ +""" +Excel 批量导入 API — 模板下载 + 批量导入 + 失败行下载 +""" + +import io +import uuid +from datetime import date + +from fastapi import APIRouter, Depends, HTTPException, Query, UploadFile, File, status +from fastapi.responses import StreamingResponse +from openpyxl import Workbook +from sqlmodel import Session + +from app.core.deps import require_role +from app.db.session import get_session +from app.models.user import User, UserRole +from app.schemas.imports import ImportError, ImportResult +from app.services.excel_service import ExcelService, generate_error_excel + +router = APIRouter(prefix="/api/imports", tags=["批量导入"]) + + +# ============================================================================ +# 内存失败行存储（进程生命周期，重启后丢失） +# 仅用于 Phase 2，生产 Phase 4+ 替换为 Redis 或 DB +# ============================================================================ +_error_store: dict[str, list[dict]] = {} + + +# ============================================================================ +# 模板下载 +# ============================================================================ + +@router.get("/template") +def download_template( + type: str = Query("episodes", description="模板类型，仅支持 episodes"), + current_user: User = Depends(require_role(UserRole.zhipianren, UserRole.zebian)), +): + """下载 episodes 批量导入 Excel 模板。""" + if type != "episodes": + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="Phase 2 不支持排期模板下载", + ) + + wb = Workbook() + ws = wb.active + ws.title = "节目期次导入模板" + + # 列说明（第1行） + headers = [ + "episode_number", "program_name", "air_date", + "editor_name", "audience_share", "audience_rating", + "is_rerun", "notes", + ] + ws.append(headers) + + # 第2行：示例数据 + sample = [ + 1, "《军事科技》武器解析系列", "2026-01-15", + "张颖", "0.72", "3.5", + "否", "首期播出", + ] + ws.append(sample) + + # 第3行：说明注释（灰色字体） + from openpyxl.styles import Font, PatternFill, Alignment + + grey_font = Font(color="808080") + note_row = [ + "期次号（整数，每年从1开始）", + "节目名称", + "播出日期（YYYY-MM-DD）", + "编导姓名（快照，匹配不上则只存姓名）", + "收视份额（0-1之间的小数，可空）", + "收视率（可空）", + "是否重播（是/否），Phase 2 不支持重播导入", + "备注", + ] + ws.append(note_row) + for cell in ws[len(headers)]: + cell.font = grey_font + + # 第4行：is_rerun 说明 + rerun_note = ["", "", "", "", "", "", "Phase 2 不支持重播导入，有重播行请空着", ""] + ws.append(rerun_note) + for cell in ws[4]: + cell.font = grey_font + + # 设置列宽 + ws.column_dimensions["A"].width = 16 + ws.column_dimensions["B"].width = 28 + ws.column_dimensions["C"].width = 14 + ws.column_dimensions["D"].width = 16 + ws.column_dimensions["E"].width = 16 + ws.column_dimensions["F"].width = 16 + ws.column_dimensions["G"].width = 40 + ws.column_dimensions["H"].width = 20 + + output = io.BytesIO() + wb.save(output) + output.seek(0) + + return StreamingResponse( + output, + media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", + headers={"Content-Disposition": "attachment; filename=episodes_import_template.xlsx"}, + ) + + +# ============================================================================ +# 批量导入 +# ============================================================================ + +@router.post("/episodes", response_model=ImportResult) +def import_episodes( + file: UploadFile = File(..., description="Excel 文件（.xlsx）"), + session: Session = Depends(get_session), + current_user: User = Depends(require_role(UserRole.zhipianren, UserRole.zebian)), +): + """批量导入节目期次 Excel。""" + if not file.filename.endswith(".xlsx"): + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="仅支持 .xlsx 格式，请使用模板下载获得正确格式", + ) + + content = file.file.read() + service = ExcelService(session) + + try: + result = service.import_episodes(content) + except ValueError as e: + msg = str(e) + if "年度目标" in msg: + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=msg) + if "重复" in msg or "库中记录重复" in msg: + raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=msg) + # 其他解析/校验错误 + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=msg) + + # 存储失败行（供后续下载） + if result["errors"]: + _error_store[result["batch_id"]] = result["errors"] + result["error_excel_url"] = f"/api/imports/errors/{result['batch_id']}" + + return result + + +# ============================================================================ +# 失败行下载 +# ============================================================================ + +@router.get("/errors/{batch_id}") +def download_error_excel( + batch_id: str, + current_user: User = Depends(require_role(UserRole.zhipianren, UserRole.zebian)), +): + """下载指定批次的失败行 Excel（供责编修正后重新导入）。""" + errors = _error_store.get(batch_id) + if errors is None: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail="批次不存在或已过期（进程重启后数据丢失），请重新上传", + ) + + content = generate_error_excel(errors) + return StreamingResponse( + io.BytesIO(content), + media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", + headers={"Content-Disposition": f"attachment; filename=import_errors_{batch_id}.xlsx"}, + ) \ No newline at end of file diff --git a/backend/app/main.py b/backend/app/main.py index 78beabe..ee2be93 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -8,6 +8,7 @@ from starlette.middleware.sessions import SessionMiddleware from app.api.auth import router as auth_router from app.api.episodes import router as episodes_router +from app.api.imports import router as imports_router from app.api.users import router as users_router from app.api.yearly_targets import router as yearly_targets_router from app.core.config import settings @@ -38,5 +39,6 @@ app.add_middleware( # ---- 注册路由 ---- app.include_router(auth_router) app.include_router(episodes_router) +app.include_router(imports_router) app.include_router(yearly_targets_router) -app.include_router(users_router) \ No newline at end of file +app.include_router(users_router) diff --git a/backend/app/schemas/imports.py b/backend/app/schemas/imports.py new file mode 100644 index 0000000..a5bbbe2 --- /dev/null +++ b/backend/app/schemas/imports.py @@ -0,0 +1,27 @@ +""" +导入相关 Schema — 请求 / 响应模型 +""" + +from datetime import date +from pydantic import BaseModel, ConfigDict + + +class ImportError(BaseModel): + """单行失败详情""" + row_number: int # Excel 行号（第2行=第1条数据） + reason: str # 失败原因 + raw_data: dict # 原始行数据（字典形式） + + model_config = ConfigDict(from_attributes=True) + + +class ImportResult(BaseModel): + """批量导入响应""" + batch_id: str # UUID，用于下载失败行 Excel + total_rows: int # 总行数 + success_count: int # 成功数 + failed_count: int # 失败数 + errors: list[ImportError] # 失败行列表 + error_excel_url: str | None # 失败行 Excel 下载 URL + + model_config = ConfigDict(from_attributes=True) \ No newline at end of file diff --git a/backend/app/services/excel_service.py b/backend/app/services/excel_service.py new file mode 100644 index 0000000..162c64a --- /dev/null +++ b/backend/app/services/excel_service.py @@ -0,0 +1,246 @@ +""" +Excel 解析服务 — 批量导入核心逻辑 + +业务规则： + - 前置校验：所有涉及年份的 yearly_targets 必须存在，否则整体报错（不入库任何行） + - 幂等：检测到重复 (year, episode_number) → 整体 409，不入库 + - 事务：逐行提交，成功立即 commit，失败行收集到 errors[] + - 软引用+快照：编导姓名匹配不到在职用户 → editor_id=NULL + editor_name_snapshot=姓名 + - is_rerun 解析：接受是/否、true/false、1/0，大小写不敏感 + - Phase 2 不支持重播行（is_rerun=是 → 报错标记为失败行） +""" + +import uuid +from datetime import date +from typing import Any + +import pandas as pd +from sqlmodel import Session, select + +from app.models.episode import Episode +from app.models.user import User +from app.models.yearly_target import YearlyTarget + + +TRUTHY = {"是", "true", "1", "yes"} +FALSY = {"否", "false", "0", "no"} + + +def parse_bool(val: Any) -> bool | None: + """解析布尔值，接受多种格式。""" + if val is None: + return None + v = str(val).strip().lower() + if v == "": + return None + if v in TRUTHY: + return True + if v in FALSY: + return False + raise ValueError(f"无法解析布尔值: {val!r}") + + +def parse_date(val: Any) -> date | None: + """解析日期，支持 YYYY-MM-DD 和 YYYY/MM/DD。""" + if val is None or str(val).strip() == "": + return None + v = str(val).strip() + for fmt in ("%Y-%m-%d", "%Y/%m/%d", "%Y%m%d"): + try: + return date.fromisoformat(v) + except ValueError: + pass + raise ValueError(f"无法解析日期: {val!r}") + + +def parse_float(val: Any) -> float | None: + """解析浮点数，接受百分比字符串。""" + if val is None: + return None + v = str(val).strip() + if v == "": + return None + if v.endswith("%"): + v = v[:-1] + return float(v) + + +def find_editor_by_name(session: Session, name: str): + """按 display_name 匹配在职编导。匹配上返 (id, name)，匹配不上返 (None, name)。""" + if not name or str(name).strip() == "": + return None, "" + name = str(name).strip() + user = session.exec( + select(User).where(User.display_name == name, User.is_active == True) + ).first() + if user: + return user.id, name + return None, name + + +class ExcelService: + def __init__(self, session: Session): + self.session = session + self.errors: list[dict] = [] + self.batch_id = str(uuid.uuid4()) + + def validate_yearly_targets(self, air_dates: list[date]): + """前置校验：所有涉及年份的 yearly_targets 必须存在，否则整体报错。""" + distinct_years = set(dt.year for dt in air_dates if dt is not None) + missing = [] + for yr in sorted(distinct_years): + exists = self.session.exec( + select(YearlyTarget).where(YearlyTarget.year == yr) + ).first() + if not exists: + missing.append(yr) + if missing: + raise ValueError( + f"导入前请先录入以下年份的年度目标：{', '.join(map(str, missing))}。" + "请先在年度目标页录入目标后再导入。" + ) + + def check_duplicates(self, rows: list[dict]): + """检测重复 (year, episode_number)，有重复则整体报错。""" + seen = set() + duplicates = [] + for row in rows: + key = (row["_air_year"], row["episode_number"]) + if key in seen: + duplicates.append(row) + seen.add(key) + if duplicates: + dup_list = [ + f"{r['_air_year']}年第 {r['episode_number']} 期" + for r in duplicates + ] + raise ValueError( + f"文件中以下期次与库中记录重复：{', '.join(dup_list)}。" + "请先手动删除重复期次后再重新导入。" + ) + + def import_episodes(self, file_content: bytes) -> dict: + """解析 Excel，批量导入 episodes。返回 ImportResult 结构。""" + # 1. 读取 Excel + df = pd.read_excel(file_content, engine="openpyxl") + rows = df.to_dict(orient="records") + + # 2. 解析 air_date 并提取年份 + parsed_rows = [] + air_dates = [] + for i, row in enumerate(rows, start=2): # Excel 行号从2开始（第1行=表头） + raw = dict(row) + try: + air_dt = parse_date(row.get("air_date")) + if air_dt is None: + raise ValueError("air_date 不能为空") + parsed_rows.append({ + "_row_number": i, + "_air_year": air_dt.year, + "_air_date": air_dt, + "episode_number": int(row["episode_number"]), + "program_name": str(row["program_name"]).strip(), + "audience_share": parse_float(row.get("audience_share")), + "audience_rating": parse_float(row.get("audience_rating")), + "is_rerun": parse_bool(row.get("is_rerun")), + "editor_name_snapshot": str(row.get("editor_name", "")).strip() or "未知编导", + "notes": str(row.get("notes", "")).strip() or None, + "_raw": raw, + }) + air_dates.append(air_dt) + except Exception as e: + self.errors.append({ + "row_number": i, + "reason": str(e), + "raw_data": raw, + }) + + # 3. 前置校验：年份 targets + try: + self.validate_yearly_targets(air_dates) + except ValueError: + raise # 直接抛给调用方，整体 400 + + # 4. 重复检测 + try: + self.check_duplicates(parsed_rows) + except ValueError: + raise # 直接抛给调用方，整体 409 + + # 5. 逐行入库 + for row in parsed_rows: + try: + self._import_one_row(row) + except Exception as e: + self.errors.append({ + "row_number": row["_row_number"], + "reason": str(e), + "raw_data": row["_raw"], + }) + + # 6. 计算结果 + total = len(rows) + success = total - len(self.errors) + return { + "batch_id": self.batch_id, + "total_rows": total, + "success_count": success, + "failed_count": len(self.errors), + "errors": self.errors, + } + + def _import_one_row(self, row: dict): + """导入单行，处理 is_rerun 逻辑。""" + # 重播行 Phase 2 不支持 + if row["is_rerun"] is True: + raise ValueError( + "Phase 2 不支持重播期次导入。" + "有重播行请空着或用其他工具录入，系统已标记为失败行。" + ) + + # 编辑匹配 + editor_id, editor_name = find_editor_by_name(self.session, row["editor_name_snapshot"]) + if editor_id is None and row["editor_name_snapshot"] == "未知编导": + editor_name = row["editor_name_snapshot"] + + # 插入 + episode = Episode( + episode_number=row["episode_number"], + program_name=row["program_name"], + air_date=row["_air_date"], + editor_id=editor_id, + editor_name_snapshot=editor_name, + audience_share=row["audience_share"], + audience_rating=row["audience_rating"], + is_rerun=False, + original_episode_id=None, + notes=row["notes"], + ) + self.session.add(episode) + self.session.commit() + self.session.refresh(episode) + + +def generate_error_excel(errors: list[dict]) -> bytes: + """生成失败行 Excel，供责编下载修正。""" + import io + from openpyxl import Workbook + + wb = Workbook() + ws = wb.active + ws.title = "失败行" + + # 表头 + headers = ["row_number", "reason"] + list(errors[0]["raw_data"].keys()) if errors else ["row_number", "reason"] + ws.append(headers) + + # 失败行 + for err in errors: + row_data = [err["row_number"], err["reason"]] + list(err["raw_data"].values()) + ws.append(row_data) + + wb.add_named_style("error_header") + output = io.BytesIO() + wb.save(output) + output.seek(0) + return output.read() \ No newline at end of file