246 lines
8.3 KiB
Python
246 lines
8.3 KiB
Python
|
"""
|
||
|
|
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()
|