95 lines
3.2 KiB
Python
95 lines
3.2 KiB
Python
|
"""
|
|||
|
|
一次性脚本：清除测试数据，导入 25 期真实收视数据 + 回填 AI 标签
|
|||
|
|
数据来源：
|
|||
|
|
- 收视数据：ai-labeling/example/2026收视update.xlsx（已导出为 _tmp_excel.json）
|
|||
|
|
- AI 标签：ai-labeling/benchmark-set/ground-truth.json（v0.6.0，制片人审定）
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
import json
|
|||
|
|
import sys
|
|||
|
|
import os
|
|||
|
|
from datetime import date
|
|||
|
|
|
|||
|
|
# 加 backend 到 path
|
|||
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'backend'))
|
|||
|
|
|
|||
|
|
from app.db.session import get_session
|
|||
|
|
from app.models.episode import Episode
|
|||
|
|
from app.models.user import User # 注册 User 表元数据，FK 解析需要
|
|||
|
|
from sqlmodel import select, delete
|
|||
|
|
|
|||
|
|
# ── 1. 读数据源 ──
|
|||
|
|
project_root = os.path.join(os.path.dirname(__file__), '..')
|
|||
|
|
|
|||
|
|
with open(os.path.join(project_root, '_tmp_excel.json'), 'r', encoding='utf-8') as f:
|
|||
|
|
excel_rows = json.load(f)
|
|||
|
|
|
|||
|
|
with open(os.path.join(project_root, 'ai-labeling', 'benchmark-set', 'ground-truth.json'), 'r', encoding='utf-8') as f:
|
|||
|
|
gt_data = json.load(f)
|
|||
|
|
|
|||
|
|
# ground-truth 按 ep 编号索引
|
|||
|
|
gt_map = {ep['ep']: ep for ep in gt_data['episodes']}
|
|||
|
|
|
|||
|
|
print(f"Excel: {len(excel_rows)} rows")
|
|||
|
|
print(f"Ground-truth: {len(gt_map)} episodes (v{gt_data['version']})")
|
|||
|
|
|
|||
|
|
# ── 2. 构造 Episode 对象 ──
|
|||
|
|
new_episodes = []
|
|||
|
|
for row in excel_rows:
|
|||
|
|
# 解析期号："第1期" -> 1
|
|||
|
|
ep_num_str = row['ep'].replace('第', '').replace('期', '')
|
|||
|
|
ep_num = int(ep_num_str)
|
|||
|
|
|
|||
|
|
# 解析日期："2026 01 06" -> date(2026, 1, 6)
|
|||
|
|
parts = row['date'].split()
|
|||
|
|
air = date(int(parts[0]), int(parts[1]), int(parts[2]))
|
|||
|
|
|
|||
|
|
# 编导名（去空格）
|
|||
|
|
editor_name = row['editor'].replace(' ', '').replace(' ', '')
|
|||
|
|
|
|||
|
|
# 从 ground-truth 取 AI 标签
|
|||
|
|
gt = gt_map.get(ep_num, {})
|
|||
|
|
|
|||
|
|
ep = Episode(
|
|||
|
|
episode_number=ep_num,
|
|||
|
|
program_name=row['title'],
|
|||
|
|
air_date=air,
|
|||
|
|
editor_id=None, # 软引用，暂不关联 user id
|
|||
|
|
editor_name_snapshot=editor_name,
|
|||
|
|
audience_share=row['share'],
|
|||
|
|
audience_rating=row['rating'],
|
|||
|
|
# AI 标签
|
|||
|
|
program_format=gt.get('program_format'),
|
|||
|
|
equipment_domain=gt.get('equipment_domain'),
|
|||
|
|
scene_tags=gt.get('scene_tags'),
|
|||
|
|
tech_tags=gt.get('tech_tags'),
|
|||
|
|
narrative_structure=gt.get('narrative_structure'),
|
|||
|
|
opening_hook=gt.get('opening_hook'),
|
|||
|
|
ai_label_confidence='reviewed' if gt else None,
|
|||
|
|
)
|
|||
|
|
new_episodes.append(ep)
|
|||
|
|
|
|||
|
|
# ── 3. 执行：清旧 + 插新 ──
|
|||
|
|
session = next(get_session())
|
|||
|
|
|
|||
|
|
# 删除所有旧数据
|
|||
|
|
old_count = len(session.exec(select(Episode)).all())
|
|||
|
|
session.exec(delete(Episode))
|
|||
|
|
print(f"Deleted {old_count} old test episodes")
|
|||
|
|
|
|||
|
|
# 插入新数据
|
|||
|
|
for ep in new_episodes:
|
|||
|
|
session.add(ep)
|
|||
|
|
|
|||
|
|
session.commit()
|
|||
|
|
print(f"Inserted {len(new_episodes)} real episodes")
|
|||
|
|
|
|||
|
|
# ── 4. 验证 ──
|
|||
|
|
all_eps = session.exec(select(Episode).order_by(Episode.air_date)).all()
|
|||
|
|
print(f"\nVerification: {len(all_eps)} episodes in DB")
|
|||
|
|
labeled = sum(1 for e in all_eps if e.program_format is not None)
|
|||
|
|
print(f"With program_format: {labeled}/{len(all_eps)}")
|
|||
|
|
print()
|
|||
|
|
for e in all_eps:
|
|||
|
|
print(f" ep{e.episode_number:02d} | {str(e.air_date)} | {e.audience_share} | {e.editor_name_snapshot} | {e.program_format or '-'}")
|