AutoScraperX/api/chui_zhi/videos.py

import math
from datetime import datetime, timedelta, timezone
from typing import Any, List, Literal, Optional, Tuple

from fastapi import APIRouter, Request
from pydantic import Field, PrivateAttr, field_validator, model_validator

from api.base import ApiParams, BaseApi
from api.errors import BusinessValidationError
from config import settings


router = APIRouter(prefix='/api/v1/crawler/videos', tags=['垂直视频'])
CHINA_TIMEZONE = timezone(timedelta(hours=8))

FilterField = Literal[
    'play_cnt',
    'like_cnt',
    'share_cnt',
    'collection_cnt',
    'comment_cnt',
    'duration',
    'publish_time',
    'create_time',
]
FilterOperator = Literal['>', '>=', '=', '<', '<=', 'between', 'in', 'not_in']
SqlFragment = Tuple[str, List[Any]]
SUPPORTED_PLATFORMS = frozenset({'xiaoniangao', 'xiaoniangaotuijianliu'})
MAX_FILTER_SET_VALUES = 100


def normalize_datetime(value: Any, field_name: str) -> datetime:
    """把毫秒时间戳或日期字符串统一转换为东八区无时区时间。"""
    if isinstance(value, bool):
        raise ValueError(f'{field_name}时间格式错误: {value}')
    if isinstance(value, datetime):
        parsed = value
    elif isinstance(value, (int, float)) or (isinstance(value, str) and value.strip().isdigit()):
        try:
            timestamp = float(value)
            if not math.isfinite(timestamp):
                raise ValueError
            if abs(timestamp) >= 10_000_000_000:
                timestamp /= 1000
            parsed = datetime.fromtimestamp(timestamp, tz=CHINA_TIMEZONE)
        except (OverflowError, OSError, ValueError) as exc:
            raise ValueError(f'{field_name}时间格式错误: {value}') from exc
    else:
        try:
            parsed = datetime.fromisoformat(str(value).strip())
        except ValueError as exc:
            raise ValueError(f'{field_name}时间格式错误: {value}') from exc

    if parsed.tzinfo is not None:
        return parsed.astimezone(CHINA_TIMEZONE).replace(tzinfo=None)
    return parsed


# ==================== 请求参数 ====================

class FilterCondition(ApiParams):
    """单个结构化筛选条件，字段和操作符均通过白名单限制。"""

    field: FilterField
    operator: FilterOperator
    value: Any

    @model_validator(mode='after')
    def validate_collection_size(self):
        if isinstance(self.value, list) and len(self.value) > MAX_FILTER_SET_VALUES:
            raise ValueError(f'筛选值最多允许{MAX_FILTER_SET_VALUES}项')
        return self


class PageCursor(ApiParams):
    """稳定翻页游标，并固定默认近3天查询的时间锚点。"""

    id: int = Field(gt=0)
    query_time: Optional[datetime] = None

    @field_validator('query_time', mode='before')
    @classmethod
    def normalize_query_time(cls, value):
        return normalize_datetime(value, 'cursor.query_time') if value is not None else None


class VideoQueryParams(ApiParams):
    """查询垂直视频的请求参数。"""

    _default_query_end: Optional[datetime] = PrivateAttr(default=None)

    platforms: List[str] = Field(
        default_factory=lambda: ['xiaoniangao', 'xiaoniangaotuijianliu'],
        min_length=1,
        max_length=20,
    )
    keywords: List[str] = Field(default_factory=list, max_length=20)
    filter_match_mode: Literal[1, 2] = 2  # 1=OR，2=AND
    filters: List[FilterCondition] = Field(default_factory=list, max_length=50)
    limit: int = Field(default=500, ge=1, le=settings.API_MAX_LIMIT)
    cursor: Optional[PageCursor] = None

    @field_validator('platforms')
    @classmethod
    def validate_platforms(cls, values: List[str]) -> List[str]:
        normalized = [value.strip() for value in values if value and value.strip()]
        if not normalized:
            raise ValueError('platforms不能为空')
        if any(len(value) > 50 for value in normalized):
            raise ValueError('platform长度不能超过50')
        normalized = list(dict.fromkeys(normalized))
        unsupported = sorted(set(normalized) - SUPPORTED_PLATFORMS)
        if unsupported:
            raise ValueError(f'不支持的平台: {", ".join(unsupported)}')
        return normalized

    @field_validator('keywords')
    @classmethod
    def validate_keywords(cls, values: List[str]) -> List[str]:
        normalized = [value.strip() for value in values if value and value.strip()]
        if values and not normalized:
            raise ValueError('keywords不能全部为空')
        if any(len(value) > 200 for value in normalized):
            raise ValueError('keyword长度不能超过200')
        return normalized

    @model_validator(mode='after')
    def set_default_query_time(self):
        """没有创建时间筛选时使用近3天；翻页时沿用首屏时间锚点。"""
        if not any(condition.field == 'create_time' for condition in self.filters):
            cursor_time = self.cursor.query_time if self.cursor else None
            self._default_query_end = cursor_time or datetime.now(CHINA_TIMEZONE).replace(tzinfo=None)
        return self


# ==================== 功能实现 ====================

SELECT_COLUMNS = (
    'video_id', 'user_id', 'out_user_id', 'platform', 'strategy',
    'out_video_id', 'video_title', 'cover_url', 'video_url', 'duration',
    'publish_time', 'play_cnt', 'like_cnt', 'share_cnt', 'collection_cnt',
    'comment_cnt', 'width', 'height', 'id', 'create_time',
)
COMPARISON_OPERATORS = frozenset({'>', '>=', '=', '<', '<='})


def normalize_filter_value(field: FilterField, value: Any) -> Any:
    """将请求值转换为数据库可比较的数字或日期字符串。"""
    if field in ('publish_time', 'create_time'):
        try:
            return normalize_datetime(value, field).strftime('%Y-%m-%d %H:%M:%S')
        except ValueError as exc:
            raise BusinessValidationError(str(exc)) from exc
    if isinstance(value, bool):
        raise BusinessValidationError(f'{field}筛选值必须是数字: {value}')
    try:
        number = float(value)
    except (TypeError, ValueError) as exc:
        raise BusinessValidationError(f'{field}筛选值必须是数字: {value}') from exc
    if not math.isfinite(number):
        raise BusinessValidationError(f'{field}筛选值必须是有限数字: {value}')
    return int(number) if number.is_integer() else number


def qualified_column(field: str, table_alias: Optional[str] = None) -> str:
    return f'`{table_alias}`.`{field}`' if table_alias else f'`{field}`'


def compile_filter_condition(
    condition: FilterCondition,
    table_alias: Optional[str] = None,
) -> SqlFragment:
    """将结构化筛选条件转换成参数化SQL。"""
    field, operator, value = condition.field, condition.operator, condition.value
    column = qualified_column(field, table_alias)

    if operator in COMPARISON_OPERATORS:
        if value in (None, '') or isinstance(value, list):
            raise BusinessValidationError(f'{field}的{operator}操作需要单个有效值')
        return f'{column} {operator} %s', [normalize_filter_value(field, value)]

    if not isinstance(value, list):
        raise BusinessValidationError(f'{field}的{operator}操作需要数组值')
    if operator == 'between':
        if len(value) != 2:
            raise BusinessValidationError(f'{field}的between操作必须传两个值')
        normalized_values = [normalize_filter_value(field, item) for item in value]
        if normalized_values[0] > normalized_values[1]:
            raise BusinessValidationError(f'{field}的between起始值不能大于结束值')
        return f'{column} BETWEEN %s AND %s', normalized_values
    if operator in ('in', 'not_in'):
        if not value:
            raise BusinessValidationError(f'{field}的{operator}操作不能为空')
        placeholders = ', '.join(['%s'] * len(value))
        sql_operator = 'IN' if operator == 'in' else 'NOT IN'
        return f'{column} {sql_operator} ({placeholders})', [
            normalize_filter_value(field, item) for item in value
        ]
    raise BusinessValidationError(f'不支持的筛选操作符: {operator}')


def build_filter_scope(
    params: VideoQueryParams,
    table_alias: Optional[str] = None,
) -> SqlFragment:
    """生成可复用的筛选范围，供主查询和去重子查询保持一致。"""
    column = lambda field: qualified_column(field, table_alias)
    platform_placeholders = ', '.join(['%s'] * len(params.platforms))
    clauses = [
        f'{column("platform")} IN ({platform_placeholders})',
        f"{column('video_url')} <> ''",
    ]
    sql_params: List[Any] = [*params.platforms]

    if params._default_query_end is not None:
        clauses.append(f'{column("create_time")} >= %s')
        clauses.append(f'{column("create_time")} < %s')
        sql_params.extend([
            params._default_query_end - timedelta(days=3),
            params._default_query_end,
        ])

    if params.keywords:
        keyword_clause = f'{column("video_title")} LIKE %s'
        clauses.append(f"({' OR '.join([keyword_clause] * len(params.keywords))})")
        sql_params.extend([f'%{keyword}%' for keyword in params.keywords])

    grouped_filters = {}
    for condition in params.filters:
        clause, values = compile_filter_condition(condition, table_alias)
        field_clauses, field_params = grouped_filters.setdefault(condition.field, ([], []))
        field_clauses.append(clause)
        field_params.extend(values)

    # 创建时间是查询范围，始终与其他条件使用AND。
    create_time_group = grouped_filters.pop('create_time', None)
    if create_time_group:
        field_clauses, field_params = create_time_group
        clauses.append(f"({' AND '.join(field_clauses)})")
        sql_params.extend(field_params)

    # 同一字段的上下界必须使用AND；不同字段之间才应用计划配置的AND/OR模式。
    if grouped_filters:
        filter_groups, filter_params = [], []
        for field_clauses, field_params in grouped_filters.values():
            filter_groups.append(f"({' AND '.join(field_clauses)})")
            filter_params.extend(field_params)
        joiner = ' OR ' if params.filter_match_mode == 1 else ' AND '
        clauses.append(f"({joiner.join(filter_groups)})")
        sql_params.extend(filter_params)
    return ' AND '.join(clauses), sql_params


def build_query(params: VideoQueryParams) -> SqlFragment:
    """先按out_video_id取最大id，再按主键回表读取当前页完整数据。"""
    filter_scope, sql_params = build_filter_scope(params, 'source')
    columns = ', '.join(f'`cv`.`{column}`' for column in SELECT_COLUMNS)
    page_size = min(params.limit, settings.API_MAX_LIMIT)
    having_clause = ''
    if params.cursor:
        # 游标必须作用在分组结果上，不能放入WHERE，否则旧重复记录会在下一页再次出现。
        having_clause = 'HAVING MAX(`source`.`id`) < %s'
        sql_params.append(params.cursor.id)
    sql = f'''SELECT {columns}
              FROM `crawler_video` AS `cv`
              INNER JOIN (
                  SELECT MAX(`source`.`id`) AS `selected_id`
                  FROM `crawler_video` AS `source`
                  WHERE {filter_scope}
                  GROUP BY
                      CASE WHEN `source`.`out_video_id` = '' THEN `source`.`id` ELSE 0 END,
                      `source`.`out_video_id`
                  {having_clause}
                  ORDER BY `selected_id` DESC
                  LIMIT %s
              ) AS `dedup` ON `dedup`.`selected_id` = `cv`.`id`
              ORDER BY `cv`.`id` DESC'''
    # 多取1条判断下一页，避免为每次请求额外执行COUNT(*)。
    sql_params.append(page_size + 1)
    return sql, sql_params


# ==================== 接口实现 ====================

def timestamp_ms(value: Optional[datetime]) -> Optional[int]:
    if value is None:
        return None
    if value.tzinfo is None:
        value = value.replace(tzinfo=CHINA_TIMEZONE)
    else:
        value = value.astimezone(CHINA_TIMEZONE)
    return int(value.timestamp() * 1000)


class VideoQueryApi(BaseApi):
    """管理视频查询的入参、SQL、分页和业务返回数据。"""

    path = '/query'

    async def logic(self, params: VideoQueryParams, request: Request):
        sql, sql_params = build_query(params)
        rows = await self.fetch_all(request, sql, sql_params)

        page_size = min(params.limit, settings.API_MAX_LIMIT)
        has_more = len(rows) > page_size
        rows = rows[:page_size]
        next_cursor = {'id': rows[-1]['id']} if has_more and rows else None
        if next_cursor is not None and params._default_query_end is not None:
            next_cursor['query_time'] = timestamp_ms(params._default_query_end)
        return {
            'data': rows,
            'count': len(rows),
            'has_more': has_more,
            'next_cursor': next_cursor,
        }


VideoQueryApi().register(router)