""" 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 io 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（pandas 2.x 要求 file-like object，用 BytesIO 包装 bytes） file_like = io.BytesIO(file_content) df = pd.read_excel(file_like, 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) output = io.BytesIO() wb.save(output) output.seek(0) return output.read()