增加垂直spider查库api

README.md

@@ -61,8 +61,8 @@ sh run_api.sh prod
 - `POST /api/v1/crawler/videos/query`
 - `GET /health` 和 `GET /ready` 仅用于本机健康检查，Nginx 示例默认禁止公网访问
 - 业务接口不校验 Token，公网部署必须在 Nginx 层配置访问控制
-- `start_time`、`end_time` 推荐传13位毫秒时间戳，API 会统一转换为东八区数据库查询时间
-- 未传时间时默认按 `create_time` 查询最近3天；传入 `keywords` 时按 `video_title` 模糊匹配
+- 创建时间通过 `filters` 的 `create_time` 字段传入，支持时间戳和日期字符串
+- 未传 `create_time` 时默认查询最近3天；传入 `keywords` 时按 `video_title` 模糊匹配
 - 响应通过 `has_more` 和 `next_cursor` 表示是否存在下一页；调用方下一页原样回传 `cursor`
 - 每次API调用都会通过有界队列批量向阿里云SLS上报URL、请求参数、SQL结果长度、查询耗时、请求总耗时、成功状态、失败阶段、异常类型、HTTP状态码、request_id和错误消息；异常时 `message` 包含脱敏后的原始堆栈；本地日志仅记录
   URL、请求参数、状态码，失败时额外记录错误原因

api/chui_zhi/videos.py

@@ -3,7 +3,7 @@ from datetime import datetime, timedelta, timezone
 from typing import Any, List, Literal, Optional, Tuple
 
 from fastapi import APIRouter, Request
-from pydantic import Field, field_validator, model_validator
+from pydantic import Field, PrivateAttr, field_validator, model_validator
 
 from api.base import ApiParams, BaseApi
 from api.errors import BusinessValidationError
@@ -21,6 +21,7 @@ FilterField = Literal[
     'comment_cnt',
     'duration',
     'publish_time',
+    'create_time',
 ]
 FilterOperator = Literal['>', '>=', '=', '<', '<=', 'between', 'in', 'not_in']
 SqlFragment = Tuple[str, List[Any]]
@@ -28,6 +29,33 @@ 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):
@@ -45,52 +73,33 @@ class FilterCondition(ApiParams):
 
 
 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,
     )
-    start_time: Optional[datetime] = None
-    end_time: Optional[datetime] = None
     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('start_time', 'end_time', mode='before')
-    @classmethod
-    def normalize_query_time(cls, value):
-        """毫秒时间戳统一转换为东八区的数据库查询时间，同时兼容日期字符串。"""
-        is_number = isinstance(value, (int, float)) and not isinstance(value, bool)
-        is_digit_string = isinstance(value, str) and value.strip().isdigit()
-        if is_number or is_digit_string:
-            try:
-                timestamp = float(value)
-                if not math.isfinite(timestamp):
-                    raise ValueError('时间戳必须是有限数字')
-                if abs(timestamp) >= 10_000_000_000:
-                    timestamp /= 1000
-                return datetime.fromtimestamp(timestamp, tz=CHINA_TIMEZONE).replace(tzinfo=None)
-            except (OverflowError, OSError, ValueError) as exc:
-                raise ValueError(f'时间格式错误: {value}') from exc
-        return value
-
-    @field_validator('start_time', 'end_time', mode='after')
-    @classmethod
-    def normalize_timezone(cls, value):
-        if value is not None and value.tzinfo is not None:
-            return value.astimezone(CHINA_TIMEZONE).replace(tzinfo=None)
-        return value
-
     @field_validator('platforms')
     @classmethod
     def validate_platforms(cls, values: List[str]) -> List[str]:
@@ -116,17 +125,11 @@ class VideoQueryParams(ApiParams):
         return normalized
 
     @model_validator(mode='after')
-    def fill_and_validate_time_range(self):
-        now = datetime.now(CHINA_TIMEZONE).replace(tzinfo=None)
-        if self.start_time is None and self.end_time is None:
-            self.end_time = now
-            self.start_time = self.end_time - timedelta(days=3)
-        elif self.start_time is None:
-            self.start_time = self.end_time - timedelta(days=3)
-        elif self.end_time is None:
-            self.end_time = now
-        if self.start_time >= self.end_time:
-            raise ValueError('start_time必须早于end_time')
+    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
 
 
@@ -143,23 +146,11 @@ COMPARISON_OPERATORS = frozenset({'>', '>=', '=', '<', '<='})
 
 def normalize_filter_value(field: FilterField, value: Any) -> Any:
     """将请求值转换为数据库可比较的数字或日期字符串。"""
-    if field == 'publish_time':
-        if isinstance(value, datetime):
-            if value.tzinfo is not None:
-                value = value.astimezone(CHINA_TIMEZONE).replace(tzinfo=None)
-            return value.strftime('%Y-%m-%d %H:%M:%S')
-        if isinstance(value, (int, float)) or (isinstance(value, str) and value.strip().isdigit()):
-            timestamp = float(value)
-            if timestamp >= 10_000_000_000:
-                timestamp /= 1000
-            return datetime.fromtimestamp(timestamp, tz=CHINA_TIMEZONE).strftime('%Y-%m-%d %H:%M:%S')
+    if field in ('publish_time', 'create_time'):
         try:
-            parsed = datetime.fromisoformat(str(value).strip())
-            if parsed.tzinfo is not None:
-                parsed = parsed.astimezone(CHINA_TIMEZONE).replace(tzinfo=None)
-            return parsed.strftime('%Y-%m-%d %H:%M:%S')
+            return normalize_datetime(value, field).strftime('%Y-%m-%d %H:%M:%S')
         except ValueError as exc:
-            raise BusinessValidationError(f'publish_time筛选值格式错误: {value}') from exc
+            raise BusinessValidationError(str(exc)) from exc
     if isinstance(value, bool):
         raise BusinessValidationError(f'{field}筛选值必须是数字: {value}')
     try:
@@ -217,25 +208,45 @@ def build_filter_scope(
     platform_placeholders = ', '.join(['%s'] * len(params.platforms))
     clauses = [
         f'{column("platform")} IN ({platform_placeholders})',
-        f'{column("create_time")} >= %s',
-        f'{column("create_time")} < %s',
         f"{column('video_url')} <> ''",
     ]
-    sql_params: List[Any] = [*params.platforms, params.start_time, params.end_time]
+    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])
 
-    filter_clauses, filter_params = [], []
+    grouped_filters = {}
     for condition in params.filters:
         clause, values = compile_filter_condition(condition, table_alias)
-        filter_clauses.append(clause)
-        filter_params.extend(values)
-    if filter_clauses:
+        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_clauses)})")
+        clauses.append(f"({joiner.join(filter_groups)})")
         sql_params.extend(filter_params)
     return ' AND '.join(clauses), sql_params
 
@@ -271,7 +282,9 @@ def build_query(params: VideoQueryParams) -> SqlFragment:
 
 # ==================== 接口实现 ====================
 
-def timestamp_ms(value: datetime) -> int:
+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:
@@ -292,13 +305,13 @@ class VideoQueryApi(BaseApi):
         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,
-            'start_time': timestamp_ms(params.start_time),
-            'end_time': timestamp_ms(params.end_time),
         }
 
 

api/fastapi_app.py

@@ -1,9 +1,11 @@
 import asyncio
+import inspect
 import json
 import re
 import time
 import uuid
 from contextlib import asynccontextmanager
+from pathlib import Path
 
 import uvicorn
 from fastapi import FastAPI, Request
@@ -24,6 +26,35 @@ API_PATH = '/api/v1/crawler/videos/query'
 HEALTH_PATH = '/health'
 READY_PATH = '/ready'
 MAX_REQUEST_BODY_SIZE = 1024 * 1024
+PROJECT_ROOT = Path(__file__).resolve().parents[1]
+
+
+def route_code_location(route) -> str:
+    """返回路由最终业务方法及其项目源码位置。"""
+    endpoint = inspect.unwrap(getattr(route, 'endpoint', None))
+    if endpoint is None:
+        return getattr(route, 'name', '')
+
+    module = getattr(endpoint, '__module__', '')
+    qualname = getattr(endpoint, '__qualname__', getattr(route, 'name', ''))
+    target = f'{module}.{qualname}'.strip('.')
+    source_file = inspect.getsourcefile(endpoint)
+    if not source_file:
+        return target
+
+    source_path = Path(source_file).resolve()
+    try:
+        display_path = source_path.relative_to(PROJECT_ROOT)
+    except ValueError:
+        # 第三方框架路由只展示模块方法，不打印虚拟环境内部路径。
+        return target
+    if display_path.parts and display_path.parts[0] == '.venv':
+        return target
+    try:
+        line_number = inspect.getsourcelines(endpoint)[1]
+    except (OSError, TypeError):
+        return f'{target} [{display_path}]'
+    return f'{target} [{display_path}:{line_number}]'
 
 
 def print_api_routes(app: FastAPI) -> None:
@@ -41,7 +72,7 @@ def print_api_routes(app: FastAPI) -> None:
         methods = ','.join(sorted(getattr(route, 'methods', None) or []))
         if path and methods:
             print(
-                f'  {methods:<12} {path:<36} -> {getattr(route, "name", "")}',
+                f'  {methods:<12} {path:<36} -> {route_code_location(route)}',
                 flush=True,
             )
     print('', flush=True)

docs/chui_zhi_video_api.md

@@ -52,17 +52,15 @@ X-Request-ID: scheduler-task-123-page-1
 | 参数 | 类型 | 必填 | 默认值 | 限制与说明 |
 |---|---|---:|---|---|
 | `platforms` | `string[]` | 否 | `xiaoniangao`、`xiaoniangaotuijianliu` | 只支持这两个平台；1～20 个；单项最长 50；自动去除首尾空格和重复值 |
-| `start_time` | 时间或时间戳 | 否 | 见时间规则 | 查询 `create_time >= start_time` |
-| `end_time` | 时间或时间戳 | 否 | 见时间规则 | 查询 `create_time < end_time`，结束时间不包含 |
 | `keywords` | `string[]` | 否 | `[]` | 最多 20 个；单项最长 200；按标题模糊匹配 |
 | `filter_match_mode` | `integer` | 否 | `2` | `1` 表示过滤条件 OR；`2` 表示过滤条件 AND |
 | `filters` | `object[]` | 否 | `[]` | 最多 50 个结构化过滤条件 |
 | `limit` | `integer` | 否 | `500` | 范围为 1～`API_MAX_LIMIT`；超出范围直接返回参数错误 |
 | `cursor` | `object` | 否 | `null` | 第一页不传；下一页原样传回响应的 `next_cursor` |
 
-### 3.1 时间参数规则
+### 3.1 创建时间规则
 
-`start_time` 和 `end_time` 支持：
+创建时间和其他筛选字段一样放在 `filters` 中，`field` 固定为 `create_time`。时间值支持：
 
 - 13 位毫秒时间戳，例如 `1786982400000`
 - 10 位秒时间戳，例如 `1786982400`
@@ -71,22 +69,16 @@ X-Request-ID: scheduler-task-123-page-1
 
 时间戳统一按东八区转换为数据库查询时间。
 
-缺省规则：
+单边或双边范围示例：
 
-| 传参情况 | 实际范围 |
-|---|---|
-| 开始和结束均未传 | 当前时间往前 3 天至当前时间 |
-| 只传 `end_time` | `end_time` 往前 3 天至 `end_time` |
-| 只传 `start_time` | `start_time` 至当前时间 |
-| 两者都传 | 使用调用方指定范围 |
-
-`start_time` 必须早于 `end_time`。查询边界为左闭右开：
-
-```sql
-create_time >= start_time AND create_time < end_time
+```json
+"filters": [
+  {"field": "create_time", "operator": ">=", "value": 1786896000000},
+  {"field": "create_time", "operator": "<", "value": 1786982400000}
+]
 ```
 
-第一页响应会返回实际使用的 `start_time` 和 `end_time`（毫秒时间戳）。调度端会固定该范围用于后续页面，避免默认“当前时间”随分页变化。
+完全没有 `create_time` 条件时，API 默认查询当前时间之前3天。多条创建时间条件固定使用 AND，不受 `filter_match_mode` 影响。默认时间范围的首屏时间锚点会写入 `next_cursor.query_time`，下一页原样回传游标即可保持范围不漂移。
 
 ### 3.2 关键词规则
 
@@ -150,6 +142,7 @@ video_title LIKE '%关键词%'
 | `comment_cnt` | 评论数 | 数字 |
 | `duration` | 视频时长，单位秒 | 数字 |
 | `publish_time` | 视频发布时间 | 时间或时间戳 |
+| `create_time` | 数据创建时间 | 时间或时间戳 |
 
 允许操作符：
 
@@ -176,8 +169,8 @@ video_title LIKE '%关键词%'
 "filter_match_mode": 2
 ```
 
-- `1`：所有 `filters` 使用 OR
-- `2`：所有 `filters` 使用 AND，默认值
+- `1`：除 `create_time` 外的过滤条件使用 OR
+- `2`：除 `create_time` 外的过滤条件使用 AND，默认值
 
 例如：
 
@@ -197,7 +190,9 @@ video_title LIKE '%关键词%'
 like_cnt > 3 AND duration BETWEEN 60 AND 300
 ```
 
-`filter_match_mode` 只控制 `filters` 数组内部，不改变关键词之间的 OR 规则。
+`filter_match_mode` 不改变关键词之间的 OR 规则。同一字段的多条条件始终使用 AND，
+因此 `play_cnt >= 100` 和 `play_cnt <= 1000` 会组成播放量区间；不同字段之间才应用
+`filter_match_mode`。多条 `create_time` 条件也始终使用 AND，并作为整个查询的时间范围。
 
 ### 3.5 游标参数
 
@@ -217,23 +212,24 @@ like_cnt > 3 AND duration BETWEEN 60 AND 300
   "keywords": ["早"],
   "limit": 500,
   "cursor": {
-    "id": 6815000
+    "id": 6815000,
+    "query_time": 1786982400000
   }
 }
 ```
 
-`cursor.id` 必须是大于 0 的整数。不要自行修改游标，也不要在分页过程中修改其他筛选参数。
+`cursor.id` 必须是大于 0 的整数。`query_time` 仅在使用默认近3天范围时返回。不要自行修改游标，也不要在分页过程中修改其他筛选参数。
 
 ## 4. 完整请求示例
 
 ```json
 {
   "platforms": ["xiaoniangao", "xiaoniangaotuijianliu"],
-  "start_time": 1786896000000,
-  "end_time": 1786982400000,
   "keywords": ["早", "养生"],
   "filter_match_mode": 2,
   "filters": [
+    {"field": "create_time", "operator": ">=", "value": 1786896000000},
+    {"field": "create_time", "operator": "<", "value": 1786982400000},
     {"field": "play_cnt", "operator": ">=", "value": 1000},
     {"field": "like_cnt", "operator": ">", "value": 3},
     {"field": "duration", "operator": "between", "value": [60, 300]},
@@ -294,7 +290,7 @@ like_cnt > 3 AND duration BETWEEN 60 AND 300
 查询依次应用以下规则：
 
 1. `platform IN (...)`
-2. `create_time >= start_time AND create_time < end_time`
+2. 应用显式的 `create_time` 条件；未提供时限定为最近3天
 3. `video_url <> ''`
 4. 有关键词时，执行标题模糊匹配
 5. 有结构化过滤条件时，按 `filter_match_mode` 组合
@@ -376,9 +372,7 @@ ORDER BY cv.id DESC;
     ],
     "count": 1,
     "has_more": true,
-    "next_cursor": {"id": 6815000},
-    "start_time": 1786723200000,
-    "end_time": 1786982400000
+    "next_cursor": {"id": 6815000, "query_time": 1786982400000}
   }
 }
 ```
@@ -393,8 +387,6 @@ ORDER BY cv.id DESC;
 | `data.count` | 当前页实际返回数量，不包含额外探测行 |
 | `data.has_more` | 是否存在下一页 |
 | `data.next_cursor` | 下一页游标；没有下一页时为 `null` |
-| `data.start_time` | API 实际使用的开始时间，13 位毫秒时间戳 |
-| `data.end_time` | API 实际使用的结束时间，13 位毫秒时间戳 |
 
 响应头包含：
 
@@ -550,10 +542,10 @@ curl --request POST "$CHUI_ZHI_API_URL" \
 curl --request POST "$CHUI_ZHI_API_URL" \
   --header "Content-Type: application/json" \
   --data '{
-    "start_time": 1786896000000,
-    "end_time": 1786982400000,
     "filter_match_mode": 2,
     "filters": [
+      {"field": "create_time", "operator": ">=", "value": 1786896000000},
+      {"field": "create_time", "operator": "<", "value": 1786982400000},
       {"field": "like_cnt", "operator": ">", "value": 3},
       {"field": "duration", "operator": "between", "value": [60, 300]}
     ],
@@ -568,14 +560,12 @@ curl --request POST "$CHUI_ZHI_API_URL" \
   --header "Content-Type: application/json" \
   --data '{
     "keywords": ["早"],
-    "start_time": 1786723200000,
-    "end_time": 1786982400000,
     "limit": 500,
-    "cursor": {"id": 6815000}
+    "cursor": {"id": 6815000, "query_time": 1786982400000}
   }'
 ```
 
-后续页必须继续使用第一页响应的时间范围、关键词、过滤条件和 `limit`，只替换 `cursor`。
+后续页必须继续使用第一页的关键词、过滤条件和 `limit`，并原样传递响应中的 `next_cursor`。
 
 ### 10.5 健康检查
 
@@ -715,10 +705,9 @@ export CHUI_ZHI_API_URL="https://api.example.com/api/v1/crawler/videos/query"
 
 1. 把抓取计划转换为 API 的结构化参数。
 2. 请求第一页。
-3. 固定 API 返回的实际 `start_time` 和 `end_time`。
-4. 上传并保存当前页视频。
-5. `has_more=true` 时传递 `next_cursor` 请求下一页。
-6. `has_more=false` 时结束。
+3. 上传并保存当前页视频。
+4. `has_more=true` 时原样传递 `next_cursor` 请求下一页。
+5. `has_more=false` 时结束。
 
 调度端复用 `requests.Session`，保持 TCP 连接；每一页处理完成后才请求下一页，避免一次性把全部视频加载到内存。
 

test/test_fastapi_video_query_api.py

@@ -4,7 +4,7 @@ from fastapi.testclient import TestClient
 
 from api.base import BaseApi
 from api.chui_zhi.videos import VideoQueryApi
-from api.fastapi_app import API_PATH, create_app
+from api.fastapi_app import API_PATH, create_app, print_api_routes
 
 
 class FakeLogger:
@@ -51,9 +51,11 @@ def test_fastapi_query_keeps_existing_contract_and_unified_logging():
     app = build_test_app(FakeMySQL())
     payload = {
         'platforms': ['xiaoniangao'],
-        'start_time': 1785772800000,
-        'end_time': 1785859200000,
-        'filters': [{'field': 'like_cnt', 'operator': '>', 'value': 3}],
+        'filters': [
+            {'field': 'create_time', 'operator': '>=', 'value': 1785772800000},
+            {'field': 'create_time', 'operator': '<', 'value': 1785859200000},
+            {'field': 'like_cnt', 'operator': '>', 'value': 3},
+        ],
         'limit': 1,
     }
     with TestClient(app, raise_server_exceptions=False) as client:
@@ -78,6 +80,22 @@ def test_fastapi_query_keeps_existing_contract_and_unified_logging():
     assert event['data']['query_result_count'] == 2
 
 
+def test_default_time_range_is_pinned_in_next_cursor():
+    class FakeMySQL:
+        async def fetch_all(self, sql, params):
+            return [{'id': 2}, {'id': 1}]
+
+    app = build_test_app(FakeMySQL())
+    with TestClient(app, raise_server_exceptions=False) as client:
+        first_page = client.post(API_PATH, json={'limit': 1}).json()['data']
+        cursor = first_page['next_cursor']
+        second_page = client.post(API_PATH, json={'limit': 1, 'cursor': cursor}).json()['data']
+
+    assert cursor['id'] == 2
+    assert isinstance(cursor['query_time'], int)
+    assert second_page['next_cursor']['query_time'] == cursor['query_time']
+
+
 def test_fastapi_validation_and_business_errors_use_same_exit_contract():
     class FakeMySQL:
         async def fetch_all(self, sql, params):
@@ -124,6 +142,18 @@ def test_fastapi_health_readiness_and_documentation_routes():
     assert API_PATH in docs.json()['paths']
 
 
+def test_startup_route_list_contains_business_source_location(capsys):
+    app = create_app(manage_resources=False)
+
+    print_api_routes(app)
+
+    output = capsys.readouterr().out
+    assert 'api.chui_zhi.videos.VideoQueryApi.logic' in output
+    assert 'api/chui_zhi/videos.py:' in output
+    assert 'api.fastapi_app.health' in output
+    assert 'api/fastapi_app.py:' in output
+
+
 def test_fastapi_sls_failure_does_not_change_business_response():
     class FakeMySQL:
         async def fetch_all(self, sql, params):

test/test_video_query_api.py

@@ -108,11 +108,14 @@ def test_existing_crawler_sls_format_is_unchanged():
 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='create_time',
+                operator='between',
+                value=['2026-08-04 00:00:00', '2026-08-05 00:00:00'],
+            ),
             FilterCondition(field='like_cnt', operator='>=', value=100),
             FilterCondition(field='duration', operator='between', value=[60, 300]),
         ],
@@ -137,9 +140,10 @@ def test_build_query_contains_parameterized_filters():
 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)],
+        filters=[
+            FilterCondition(field='create_time', operator='>=', value='2026-08-04 00:00:00'),
+            FilterCondition(field='like_cnt', operator='>', value=3),
+        ],
         cursor={'id': 100},
         limit=50,
     )
@@ -155,16 +159,13 @@ def test_deduplication_groups_ids_before_cursor_pagination():
 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),
+        filters=[FilterCondition(field='create_time', operator='>=', value='2026-08-04 00:00:00')],
         limit=100,
         cursor={'id': 123},
     )
@@ -178,15 +179,16 @@ def test_id_cursor_builds_stable_group_pagination_and_fetches_one_extra_row():
 
 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),
+    query_time = datetime(2026, 8, 4, tzinfo=china_timezone)
+    condition = FilterCondition(
+        field='create_time',
+        operator='>=',
+        value=int(query_time.timestamp() * 1000),
     )
+    sql, params = compile_filter_condition(condition)
 
-    assert request.start_time == datetime(2026, 8, 4)
-    assert request.end_time == datetime(2026, 8, 5)
+    assert sql == '`create_time` >= %s'
+    assert params == ['2026-08-04 00:00:00']
 
 
 def test_missing_time_defaults_to_latest_three_days():
@@ -194,14 +196,44 @@ def test_missing_time_defaults_to_latest_three_days():
     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)
+    assert before <= request._default_query_end <= after
 
     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
+    assert request._default_query_end - timedelta(days=3) in sql_params
+    assert request._default_query_end in sql_params
+
+
+def test_explicit_create_time_filter_disables_default_time_range():
+    request = VideoQueryParams(filters=[
+        FilterCondition(field='create_time', operator='>', value='2026-08-04 00:00:00'),
+    ])
+
+    sql, params = build_query(request)
+
+    assert request._default_query_end is None
+    assert '`source`.`create_time` > %s' in sql
+    assert '`source`.`create_time` < %s' not in sql
+    assert '2026-08-04 00:00:00' in params
+
+
+def test_same_field_bounds_are_and_grouped_in_or_mode():
+    request = VideoQueryParams(
+        filter_match_mode=1,
+        filters=[
+            FilterCondition(field='play_cnt', operator='>=', value=100),
+            FilterCondition(field='play_cnt', operator='<=', value=1000),
+            FilterCondition(field='like_cnt', operator='>', value=3),
+        ],
+    )
+
+    sql, params = build_query(request)
+
+    assert '(`source`.`play_cnt` >= %s AND `source`.`play_cnt` <= %s)' in sql
+    assert ' OR (`source`.`like_cnt` > %s)' in sql
+    assert 100 in params
+    assert 1000 in params
 
 
 def test_rejects_unsupported_filter_field():