simonkoson
247 lines
8.4 KiB
Python
247 lines
8.4 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 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() |