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>
43 lines
1.2 KiB
Python
43 lines
1.2 KiB
Python
from typing import List, Literal
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models import KolVideo
|
|
from app.config import settings
|
|
|
|
|
|
async def query_videos(
|
|
session: AsyncSession,
|
|
query_type: Literal["star_id", "unique_id", "nickname"],
|
|
values: List[str],
|
|
) -> List[KolVideo]:
|
|
"""
|
|
查询 KOL 视频数据.
|
|
|
|
Args:
|
|
session: 数据库会话
|
|
query_type: 查询类型 (star_id, unique_id, nickname)
|
|
values: 查询值列表
|
|
|
|
Returns:
|
|
匹配的视频列表
|
|
"""
|
|
stmt = select(KolVideo)
|
|
|
|
if query_type == "star_id":
|
|
# 精准匹配 star_id
|
|
stmt = stmt.where(KolVideo.star_id.in_(values))
|
|
elif query_type == "unique_id":
|
|
# 精准匹配 star_unique_id
|
|
stmt = stmt.where(KolVideo.star_unique_id.in_(values))
|
|
elif query_type == "nickname":
|
|
# 模糊匹配 star_nickname (使用第一个值)
|
|
if values:
|
|
stmt = stmt.where(KolVideo.star_nickname.like(f"%{values[0]}%"))
|
|
|
|
# 限制返回数量
|
|
stmt = stmt.limit(settings.MAX_QUERY_LIMIT)
|
|
|
|
result = await session.execute(stmt)
|
|
return list(result.scalars().all())
|