@@ -0,0 +1,255 @@
"""
Task 2 批量导入冒烟脚本 — httpx（幂等版，可重复跑）
确保后端已启动: py -3.12 -c " import sys; sys.path.insert(0, r ' E: \t ps-dashboard \b ackend ' ); 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，
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 ( )
simonkoson