""" 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()