feat: Task 2 Excel 批量导入 API 完成(模板下载 + 批量导入 + 失败行下载)

This commit is contained in:
simonkoson
2026-05-15 13:40:13 +08:00
parent 9439c72d30
commit 2480d4a738
4 changed files with 448 additions and 1 deletions
+172
View File
@@ -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"},
)
+3 -1
View File
@@ -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)
app.include_router(users_router)
+27
View File
@@ -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)
+246
View File
@@ -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()