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)