AutoScraperX/test/test_video_query_api.py

from datetime import datetime, timedelta, timezone

import pytest

from api.chui_zhi.videos import (
    BusinessValidationError,
    FilterCondition,
    VideoQueryParams,
    build_query,
    compile_filter_condition,
)
from config import settings
from core.utils.log.aliyun_log import AliyunLogger
from core.utils.log.logger_manager import LoggerManager


def test_api_aliyun_logger_uses_independent_destination():
    logger = LoggerManager.get_aliyun_logger(
        platform='chui_zhi',
        mode='api',
        env='prod',
        project='crawler-log-prod',
        logstore='crawler-api-access',
        endpoint='cn-hangzhou.log.aliyuncs.com',
    )

    assert isinstance(logger, AliyunLogger)
    assert logger._resolve_destination() == (
        'crawler-log-prod',
        'crawler-api-access',
        'cn-hangzhou.log.aliyuncs.com',
    )


def test_default_crawler_aliyun_destination_is_unchanged():
    logger = AliyunLogger(platform='dou_yin', mode='crawler', env='prod')

    assert logger._resolve_destination() == (
        'crawler-log-prod',
        'crawler-fetch',
        'cn-hangzhou.log.aliyuncs.com',
    )


def test_api_sls_log_is_flat_and_omits_empty_fields():
    logger = AliyunLogger(platform='chui_zhi', mode='api')
    contents = dict(logger._build_log_contents({
        'code': '2000',
        'message': 'API请求成功',
        'trace_id': 'request-1',
        'account': None,
        'data': {
            'request_id': 'request-1',
            'path': '/api/v1/crawler/videos/query',
            'method': 'POST',
            'request_params': {'keywords': ['早'], 'filters': []},
            'platforms': ['xiaoniangao', 'xiaoniangaotuijianliu'],
            'status_code': 200,
            'success': True,
            'request_duration_ms': 4096.24,
            'failure_stage': '',
            'error_type': '',
            'query_result_count': 140,
            'query_duration_ms': 4091.31,
            'query_success': True,
        },
    }))

    assert contents['TraceId'] == 'request-1'
    assert contents['success'] == 'true'
    assert contents['query_success'] == 'true'
    assert contents['request_params'] == '{"keywords":["早"],"filters":[]}'
    assert contents['platforms'] == '["xiaoniangao","xiaoniangaotuijianliu"]'
    assert 'data' not in contents
    assert 'account' not in contents
    assert 'timestamp' not in contents
    assert 'code' not in contents
    assert 'failure_stage' not in contents
    assert 'error_type' not in contents


def test_existing_crawler_sls_format_is_unchanged():
    logger = AliyunLogger(platform='dou_yin', mode='crawler')
    contents = dict(logger._build_log_contents({
        'code': '2000',
        'message': '抓取成功',
        'trace_id': 'crawler-trace-1',
        'account': None,
        'data': {
            'url': 'https://example.com/video/1',
            'status_code': 200,
            'success': True,
        },
    }))

    assert contents['TraceId'] == 'crawler-trace-1'
    assert contents['code'] == '2000'
    assert contents['data'] == (
        '{"url": "https://example.com/video/1", "status_code": 200, "success": true}'
    )
    assert contents['account'] == 'None'
    assert contents['success'] == 'True'
    assert 'timestamp' in contents
    # 旧格式不会把普通data中的url提升为顶层字段。
    assert 'url' not in contents


def test_build_query_contains_parameterized_filters():
    request = VideoQueryParams(
        platforms=['xiaoniangao', 'xiaoniangaotuijianliu'],
        start_time=datetime(2026, 8, 4),
        end_time=datetime(2026, 8, 5),
        keywords=['养生'],
        filter_match_mode=2,
        filters=[
            FilterCondition(field='like_cnt', operator='>=', value=100),
            FilterCondition(field='duration', operator='between', value=[60, 300]),
        ],
        limit=20,
    )

    sql, params = build_query(request)

    assert '`platform` IN (%s, %s)' in sql
    assert '`video_title` LIKE %s' in sql
    assert '`like_cnt` >= %s' in sql
    assert '`duration` BETWEEN %s AND %s' in sql
    assert 'SELECT MAX(`source`.`id`) AS `selected_id`' in sql
    assert "CASE WHEN `source`.`out_video_id` = '' THEN `source`.`id` ELSE 0 END" in sql
    assert '`dedup`.`selected_id` = `cv`.`id`' in sql
    assert 'ORDER BY `cv`.`id` DESC' in sql
    assert 'OFFSET' not in sql
    assert params[-1] == 21
    assert '%养生%' in params


def test_deduplication_groups_ids_before_cursor_pagination():
    request = VideoQueryParams(
        platforms=['xiaoniangao'],
        start_time=datetime(2026, 8, 4),
        end_time=datetime(2026, 8, 5),
        filters=[FilterCondition(field='like_cnt', operator='>', value=3)],
        cursor={'id': 100},
        limit=50,
    )

    sql, params = build_query(request)

    # 筛选条件只出现一次；HAVING作用于MAX(id)，防止旧重复记录跨页再次出现。
    assert sql.count('`source`.`like_cnt` > %s') == 1
    assert 'HAVING MAX(`source`.`id`) < %s' in sql
    assert params[-2:] == [100, 51]


def test_query_limit_above_server_max_is_rejected():
    with pytest.raises(ValueError):
        VideoQueryParams(
            start_time=datetime(2026, 8, 4),
            end_time=datetime(2026, 8, 5),
            limit=settings.API_MAX_LIMIT + 1,
        )


def test_id_cursor_builds_stable_group_pagination_and_fetches_one_extra_row():
    request = VideoQueryParams(
        start_time=datetime(2026, 8, 4),
        end_time=datetime(2026, 8, 5),
        limit=100,
        cursor={'id': 123},
    )

    sql, params = build_query(request)

    assert 'HAVING MAX(`source`.`id`) < %s' in sql
    assert params[-2:] == [123, 101]
    assert 'OFFSET' not in sql


def test_millisecond_timestamps_are_converted_to_china_time():
    china_timezone = timezone(timedelta(hours=8))
    start_time = datetime(2026, 8, 4, tzinfo=china_timezone)
    end_time = datetime(2026, 8, 5, tzinfo=china_timezone)
    request = VideoQueryParams(
        start_time=int(start_time.timestamp() * 1000),
        end_time=int(end_time.timestamp() * 1000),
    )

    assert request.start_time == datetime(2026, 8, 4)
    assert request.end_time == datetime(2026, 8, 5)


def test_missing_time_defaults_to_latest_three_days():
    before = datetime.now(timezone(timedelta(hours=8))).replace(tzinfo=None)
    request = VideoQueryParams()
    after = datetime.now(timezone(timedelta(hours=8))).replace(tzinfo=None)

    assert before <= request.end_time <= after
    assert request.end_time - request.start_time == timedelta(days=3)

    sql, sql_params = build_query(request)
    assert '`create_time` >= %s' in sql
    assert '`create_time` < %s' in sql
    assert request.start_time in sql_params
    assert request.end_time in sql_params


def test_rejects_unsupported_filter_field():
    with pytest.raises(ValueError):
        FilterCondition(field='unknown_field', operator='>', value=1)


def test_rejects_unsupported_platform_and_unknown_filter_parameter():
    with pytest.raises(ValueError, match='不支持的平台'):
        VideoQueryParams(platforms=['douyin'])
    with pytest.raises(ValueError):
        FilterCondition(field='like_cnt', operator='>', value=1, unknown='value')


@pytest.mark.parametrize(
    'field',
    [
        'like_cnt',
        'collection_cnt',
        'comment_cnt',
        'share_cnt',
        'play_cnt',
        'duration',
    ],
)
def test_supported_numeric_filter_mapping(field):
    condition = FilterCondition(field=field, operator='>', value=10)
    sql, params = compile_filter_condition(condition)

    assert f'`{field}` > %s' == sql
    assert params == [10]


def test_empty_filters_are_not_added_to_query():
    request = VideoQueryParams(filters=[])
    sql, _ = build_query(request)
    where_sql = sql.split('WHERE', 1)[1]
    assert '`like_cnt` >' not in where_sql


def test_publish_time_filter_mapping():
    condition = FilterCondition(
        field='publish_time',
        operator='>=',
        value='2026-08-01 00:00:00',
    )
    sql, params = compile_filter_condition(condition)

    assert sql == '`publish_time` >= %s'
    assert params == ['2026-08-01 00:00:00']


def test_filter_range_and_set_are_parameterized():
    range_condition = FilterCondition(field='duration', operator='between', value=[60, 300])
    set_condition = FilterCondition(field='play_cnt', operator='in', value=[10, 20])
    range_sql, range_params = compile_filter_condition(range_condition)
    set_sql, set_params = compile_filter_condition(set_condition)

    assert range_sql == '`duration` BETWEEN %s AND %s'
    assert range_params == [60, 300]
    assert set_sql == '`play_cnt` IN (%s, %s)'
    assert set_params == [10, 20]


def test_rejects_invalid_filter_range_and_non_finite_number():
    with pytest.raises(BusinessValidationError, match='起始值不能大于结束值'):
        compile_filter_condition(
            FilterCondition(field='duration', operator='between', value=[300, 60])
        )
    with pytest.raises(BusinessValidationError, match='有限数字'):
        compile_filter_condition(
            FilterCondition(field='like_cnt', operator='>', value='NaN')
        )


def test_rejects_raw_sql_in_filter_value():
    with pytest.raises(BusinessValidationError, match='必须是数字'):
        condition = FilterCondition(field='like_cnt', operator='>', value='0 OR 1=1')
        compile_filter_condition(condition)