8fbcb72a3f
- 实现查询API (query.py): 支持star_id/unique_id/nickname三种查询方式 - 实现计算模块 (calculator.py): CPM/自然搜索UV/搜索成本计算 - 实现品牌API集成 (brand_api.py): 批量并发调用，10并发限制 - 实现导出服务 (export_service.py): Excel/CSV导出 - 前端组件: QueryForm/ResultTable/ExportButton - 主页面集成: 支持6种页面状态 - 测试: 44个测试全部通过，覆盖率88% Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
98 lines
2.6 KiB
Python
98 lines
2.6 KiB
Python
import csv
|
|
from io import BytesIO, StringIO
|
|
from typing import List, Dict, Any, Tuple
|
|
from openpyxl import Workbook
|
|
|
|
# 列定义: (中文名, 字段名)
|
|
COLUMN_HEADERS: List[Tuple[str, str]] = [
|
|
("视频ID", "item_id"),
|
|
("视频标题", "title"),
|
|
("爆文类型", "viral_type"),
|
|
("视频链接", "video_url"),
|
|
("新增A3率", "new_a3_rate"),
|
|
("看后搜人数", "after_view_search_uv"),
|
|
("回搜次数", "return_search_cnt"),
|
|
("自然曝光数", "natural_play_cnt"),
|
|
("加热曝光数", "heated_play_cnt"),
|
|
("总曝光数", "total_play_cnt"),
|
|
("总互动", "total_interact"),
|
|
("点赞", "like_cnt"),
|
|
("转发", "share_cnt"),
|
|
("评论", "comment_cnt"),
|
|
("合作行业ID", "industry_id"),
|
|
("合作行业", "industry_name"),
|
|
("合作品牌ID", "brand_id"),
|
|
("合作品牌", "brand_name"),
|
|
("发布时间", "publish_time"),
|
|
("达人昵称", "star_nickname"),
|
|
("达人unique_id", "star_unique_id"),
|
|
("预估视频价格", "estimated_video_cost"),
|
|
("预估自然CPM", "estimated_natural_cpm"),
|
|
("预估自然看后搜人数", "estimated_natural_search_uv"),
|
|
("预估自然看后搜人数成本", "estimated_natural_search_cost"),
|
|
]
|
|
|
|
|
|
def format_value(value: Any) -> Any:
|
|
"""格式化导出值."""
|
|
if value is None:
|
|
return ""
|
|
return value
|
|
|
|
|
|
def generate_excel(data: List[Dict[str, Any]]) -> bytes:
|
|
"""
|
|
生成 Excel 文件.
|
|
|
|
Args:
|
|
data: 数据列表
|
|
|
|
Returns:
|
|
Excel 文件的字节内容
|
|
"""
|
|
wb = Workbook()
|
|
ws = wb.active
|
|
ws.title = "KOL数据"
|
|
|
|
# 写入表头
|
|
headers = [col[0] for col in COLUMN_HEADERS]
|
|
ws.append(headers)
|
|
|
|
# 写入数据
|
|
for row in data:
|
|
row_data = [format_value(row.get(col[1])) for col in COLUMN_HEADERS]
|
|
ws.append(row_data)
|
|
|
|
# 保存到内存
|
|
output = BytesIO()
|
|
wb.save(output)
|
|
output.seek(0)
|
|
return output.read()
|
|
|
|
|
|
def generate_csv(data: List[Dict[str, Any]]) -> bytes:
|
|
"""
|
|
生成 CSV 文件.
|
|
|
|
Args:
|
|
data: 数据列表
|
|
|
|
Returns:
|
|
CSV 文件的字节内容 (UTF-8 BOM 编码)
|
|
"""
|
|
output = StringIO()
|
|
writer = csv.writer(output, quoting=csv.QUOTE_MINIMAL)
|
|
|
|
# 写入表头
|
|
headers = [col[0] for col in COLUMN_HEADERS]
|
|
writer.writerow(headers)
|
|
|
|
# 写入数据
|
|
for row in data:
|
|
row_data = [format_value(row.get(col[1])) for col in COLUMN_HEADERS]
|
|
writer.writerow(row_data)
|
|
|
|
# 返回 UTF-8 BOM 编码的内容 (Excel 可正确识别中文)
|
|
content = output.getvalue()
|
|
return ("\ufeff" + content).encode("utf-8")
|