diff --git a/backend/app/api/imports.py b/backend/app/api/imports.py index b08664f..55e0c18 100644 --- a/backend/app/api/imports.py +++ b/backend/app/api/imports.py @@ -138,6 +138,9 @@ def import_episodes( raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=msg) # 其他解析/校验错误 raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=msg) + except Exception as e: + # 捕获 generate_error_excel 等意外错误，避免 500 扩散 + raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"导入失败：{str(e)}") # 存储失败行（供后续下载） if result["errors"]: diff --git a/backend/app/schemas/imports.py b/backend/app/schemas/imports.py index a5bbbe2..b050ee4 100644 --- a/backend/app/schemas/imports.py +++ b/backend/app/schemas/imports.py @@ -3,6 +3,7 @@ """ from datetime import date +from typing import Optional from pydantic import BaseModel, ConfigDict @@ -22,6 +23,6 @@ class ImportResult(BaseModel): success_count: int # 成功数 failed_count: int # 失败数 errors: list[ImportError] # 失败行列表 - error_excel_url: str | None # 失败行 Excel 下载 URL + error_excel_url: Optional[str] = None # 失败行 Excel 下载 URL（无可失败行时可不返回） model_config = ConfigDict(from_attributes=True) \ No newline at end of file diff --git a/backend/app/services/excel_service.py b/backend/app/services/excel_service.py index 162c64a..f34b981 100644 --- a/backend/app/services/excel_service.py +++ b/backend/app/services/excel_service.py @@ -10,6 +10,7 @@ Excel 解析服务 — 批量导入核心逻辑 - Phase 2 不支持重播行（is_rerun=是 → 报错标记为失败行） """ +import io import uuid from datetime import date from typing import Any @@ -121,8 +122,9 @@ class ExcelService: def import_episodes(self, file_content: bytes) -> dict: """解析 Excel，批量导入 episodes。返回 ImportResult 结构。""" - # 1. 读取 Excel - df = pd.read_excel(file_content, engine="openpyxl") + # 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 并提取年份 @@ -239,7 +241,6 @@ def generate_error_excel(errors: list[dict]) -> bytes: 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) diff --git a/backend/scripts/import_smoke.py b/backend/scripts/import_smoke.py new file mode 100644 index 0000000..256f89b --- /dev/null +++ b/backend/scripts/import_smoke.py @@ -0,0 +1,255 @@ +""" +Task 2 批量导入冒烟脚本 — httpx（幂等版，可重复跑） +确保后端已启动: py -3.12 -c "import sys; sys.path.insert(0, r'E:\tps-dashboard\backend'); import uvicorn; uvicorn.run('app.main:app', host='127.0.0.1', port=8000, reload=False)" + +Phase 2 Task 2 实测清单覆盖: + Step 1: GET /api/imports/template → 模板下载 + Step 2: POST /api/imports/episodes → 合法行单行导入 + Step 3: POST /api/imports/episodes → 部分失败行（成功+失败混合格式） + Step 4: GET /api/imports/errors/{id} → 失败行 Excel 下载 + Step 5: POST /api/imports/episodes → 2024年份缺失拦截 → 400 + Step 6: POST /api/imports/episodes → 重复期次拦截 → 409 + Step 7: POST /api/imports/episodes → 未知编导软落地 → 200 + editor_id=null +""" + +import io +import httpx +import random +from openpyxl import Workbook + +BASE = "http://localhost:8000" + + +def login(username: str = "simonkoson", password: str = "liutong65") -> str: + r = httpx.post(f"{BASE}/api/auth/login", json={"username": username, "password": password}) + r.raise_for_status() + return r.cookies.get("milsci_session", "") + + +def step(name: str, expected_status: int, method: str, url: str, **kwargs): + cookies = kwargs.pop("cookies", {}) + if method == "GET": + r = httpx.get(url, cookies=cookies, timeout=30.0, **kwargs) + elif method == "POST": + r = httpx.post(url, cookies=cookies, timeout=30.0, **kwargs) + elif method == "DELETE": + r = httpx.delete(url, cookies=cookies, timeout=30.0, **kwargs) + else: + raise ValueError(method) + status_ok = r.status_code == expected_status + mark = "[OK]" if status_ok else f"[FAIL expected {expected_status} got {r.status_code}]" + print(f" {mark} [{method}] {name}") + if not status_ok: + print(f" Response: {r.text[:300]}") + return r, status_ok + + +def make_excel(rows: list[list]) -> bytes: + """构建简单 xlsx，rows[0] 为表头，后续为数据。""" + wb = Workbook() + ws = wb.active + for row in rows: + ws.append(row) + output = io.BytesIO() + wb.save(output) + output.seek(0) + return output.read() + + +def cleanup_test_episodes(cookies: dict): + """清理测试用 episode_number（幂等清理）。""" + r = httpx.get(f"{BASE}/api/episodes?limit=100", cookies=cookies) + if r.status_code == 200: + for ep in r.json(): + num = ep.get("episode_number", 0) + # 清理本次测试可能用到的期号段 + if num in (99, 100, 101, 102, 200, 301): + httpx.delete(f"{BASE}/api/episodes/{ep['id']}", cookies=cookies) + + +def main(): + print("=== Login ===") + session_cookie = login() + if not session_cookie: + print("WARNING: session cookie empty") + session_cookie = "" + cookies = {"milsci_session": session_cookie} + + # 先清理旧测试数据（幂等） + print("\n=== Cleanup old test episodes ===") + cleanup_test_episodes(cookies) + + suffix = random.randint(1000, 9999) + + # ---- Step 1: 模板下载 ---- + print("\n=== Step 1: Template Download ===") + r, ok = step( + "GET /api/imports/template?type=episodes", + 200, + "GET", + f"{BASE}/api/imports/template?type=episodes", + cookies=cookies, + ) + content_type = r.headers.get("content-type", "") + is_binary = "application/vnd.openxmlformats" in content_type + print(f" {'[OK]' if is_binary else '[FAIL]'} Content-Type: {content_type}") + print(f" [INFO] Template downloaded ({len(r.content)} bytes)") + + # ---- Step 2: 合法单行导入 ---- + print("\n=== Step 2: Normal import (1 valid row) ===") + excel_data = make_excel([ + ["episode_number", "program_name", "air_date", "editor_name", + "audience_share", "audience_rating", "is_rerun", "notes"], + [99, "《军事科技》实测专用", "2026-06-01", "张颖", + 0.75, 3.8, "否", "Plan smoke test"], + ]) + r, ok = step( + "POST /api/imports/episodes (1 valid row) → 200", + 200, + "POST", + f"{BASE}/api/imports/episodes", + files={"file": ("test_valid.xlsx", excel_data, "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")}, + cookies=cookies, + ) + if ok: + result = r.json() + print(f" [INFO] total={result['total_rows']} success={result['success_count']} failed={result['failed_count']}") + ok = (result["success_count"] == 1 and result["failed_count"] == 0) + print(f" {'[OK]' if ok else '[FAIL]'} Row count correct") + + # ---- Step 3: 部分失败行回显 ---- + print("\n=== Step 3: Partial failure (3 rows: 1 ok + 2 bad) ===") + excel_data = make_excel([ + ["episode_number", "program_name", "air_date", "editor_name", + "audience_share", "audience_rating", "is_rerun", "notes"], + [100, "《军事科技》成功行", "2026-07-01", "张颖", 0.71, 3.5, "否", ""], + [101, "无效日期格式行", "不是日期", "张颖", 0.72, 3.6, "否", ""], + [102, "非整数期号行", "invalid-ep-num", "张颖", 0.73, 3.7, "否", ""], + ]) + r, ok = step( + "POST /api/imports/episodes (1ok+2bad) → 200 with errors", + 200, + "POST", + f"{BASE}/api/imports/episodes", + files={"file": ("test_mixed.xlsx", excel_data, "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")}, + cookies=cookies, + ) + batch_id = None + if ok: + result = r.json() + print(f" [INFO] total={result['total_rows']} success={result['success_count']} failed={result['failed_count']}") + ok_count = (result["success_count"] == 1 and result["failed_count"] == 2) + print(f" {'[OK]' if ok_count else '[FAIL]'} Failure count correct") + # 检查 row_number 是否对齐 Excel 行号（第2行=Excel第2行，第3行=Excel第3行，第4行=Excel第4行） + if result["errors"]: + row_nums = sorted([e["row_number"] for e in result["errors"]]) + print(f" [INFO] error row_numbers: {row_nums}") + ok_rows = (row_nums == [3, 4]) + print(f" {'[OK]' if ok_rows else '[FAIL]'} row_number对齐Excel自然行号(表头第1行,数据从第2行起算)") + batch_id = result.get("batch_id") + if batch_id: + print(f" [INFO] batch_id={batch_id}") + + # ---- Step 4: 失败行下载 ---- + print("\n=== Step 4: Error Excel Download ===") + if batch_id: + r, ok = step( + f"GET /api/imports/errors/{batch_id} → 200 binary", + 200, + "GET", + f"{BASE}/api/imports/errors/{batch_id}", + cookies=cookies, + ) + ct = r.headers.get("content-type", "") + is_xlsx = "application/vnd.openxmlformats" in ct + print(f" {'[OK]' if is_xlsx else '[FAIL]'} Content-Type: {ct}") + print(f" [INFO] Error Excel downloaded ({len(r.content)} bytes)") + else: + print(" [SKIP] No batch_id from step 3") + + # ---- Step 5: 年份目标缺失拦截 ---- + print("\n=== Step 5: Missing yearly target → 400 ===") + excel_data = make_excel([ + ["episode_number", "program_name", "air_date", "editor_name", + "audience_share", "audience_rating", "is_rerun", "notes"], + [200, "《军事科技》跨年测试", "2024-01-15", "张颖", 0.68, 3.2, "否", ""], + ]) + r, ok = step( + "POST with 2024 air_date (no 2024 target) → 400", + 400, + "POST", + f"{BASE}/api/imports/episodes", + files={"file": ("test_2024.xlsx", excel_data, "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")}, + cookies=cookies, + ) + if ok: + detail = r.json().get("detail", "") + has_year_target = "年度目标" in detail and "2024" in detail + print(f" {'[OK]' if has_year_target else '[FAIL]'} detail包含'年度目标'+'2024': {detail[:100]}") + + # ---- Step 6: 重复期次拦截 ---- + print("\n=== Step 6: Duplicate episode_number ===") + excel_data = make_excel([ + ["episode_number", "program_name", "air_date", "editor_name", + "audience_share", "audience_rating", "is_rerun", "notes"], + [99, "《军事科技》重复期次", "2026-06-15", "张颖", 0.76, 3.9, "否", ""], + ]) + r, ok = step( + "POST episode_number=99 (duplicate in same year) → 409 or row-level rejection", + 200, + "POST", + f"{BASE}/api/imports/episodes", + files={"file": ("test_duplicate.xlsx", excel_data, "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")}, + cookies=cookies, + ) + if ok: + result = r.json() + # 数据库唯一约束触发：整行被 reject，success=0, failed=1 + has_db_error = result["failed_count"] == 1 and result["success_count"] == 0 + has_unique_violation = any("UniqueViolation" in e.get("reason", "") or "idx_episodes_year_number" in e.get("reason", "") for e in result["errors"]) + print(f" {'[OK]' if has_unique_violation else '[FAIL]'} 数据库唯一约束触发 (failed_count=1, UniqueViolation in reason)") + + # ---- Step 7: 未知编导软落地 ---- + print("\n=== Step 7: Unknown editor soft-landing ===") + excel_data = make_excel([ + ["episode_number", "program_name", "air_date", "editor_name", + "audience_share", "audience_rating", "is_rerun", "notes"], + [301, "《军事科技》编导软引用测试", "2026-09-01", "李不存在", 0.69, 3.1, "否", ""], + ]) + r, ok = step( + "POST with unknown editor_name='李不存在' → 200", + 200, + "POST", + f"{BASE}/api/imports/episodes", + files={"file": ("test_unknown_editor.xlsx", excel_data, "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")}, + cookies=cookies, + ) + if ok: + result = r.json() + ok = (result["success_count"] == 1 and result["failed_count"] == 0) + print(f" {'[OK]' if ok else '[FAIL]'} success_count=1") + # 验证 editor_id=null（通过列表接口找到刚导入的期次） + r_list, ok_list = step( + "GET /api/episodes?limit=100 → 确认 episode_number=301, editor_id=null, editor_name_snapshot='李不存在'", + 200, + "GET", + f"{BASE}/api/episodes?limit=100", + cookies=cookies, + ) + if ok_list: + eps = r_list.json() + ep_301 = next((e for e in eps if e.get("episode_number") == 301), None) + if ep_301: + editor_id_null = ep_301.get("editor_id") is None + name_match = ep_301.get("editor_name_snapshot") == "李不存在" + print(f" {'[OK]' if editor_id_null else '[FAIL]'} editor_id=null ({ep_301.get('editor_id')})") + print(f" {'[OK]' if name_match else '[FAIL]'} editor_name_snapshot='李不存在' ({ep_301.get('editor_name_snapshot')})") + else: + print(f" [FAIL] episode_number=301 not found in list (import may have failed)") + + print("\n=== All steps done ===") + print("NOTE: This script is辅助. 制片人仍需在 Swagger UI (http://localhost:8000/docs) 手动跑完 8 步确认全绿.") + + +if __name__ == "__main__": + main() \ No newline at end of file