增加垂直spider查库api

.env

@@ -36,3 +36,13 @@ REDIS_PORT=6379
 REDIS_PASSWORD="Wqsd@2019"
 REDIS_DB=0
 REDIS_MAX_CONNECTIONS=50
+
+API_HOST=127.0.0.1
+API_PORT=8888
+CHUI_ZHI_API_TOKEN=a8790ee353a8ccb7d7489aace4e29d5b36e0a4e069f0f04f83f03f691b03d8d7
+# 垂直视频 API：数据库连接池、查询并发与超时保护
+DB_POOL_SIZE=12
+API_MAX_CONCURRENT_REQUESTS=8
+API_QUEUE_TIMEOUT=2
+API_QUERY_TIMEOUT=30
+API_MAX_LIMIT=500

README.md

@@ -51,25 +51,28 @@
 完整的接口参数、筛选规则、去重分页、响应、错误码、日志和部署说明见
 [`docs/chui_zhi_video_api.md`](docs/chui_zhi_video_api.md)。
 
-API 默认只监听 `127.0.0.1:8888`，并强制配置 Token。跨服务器访问时应绑定内网/VPN
-地址或使用网关反向代理，不建议直接暴露到公网。
+API 默认只监听 `127.0.0.1:8888`。跨服务器访问时应使用 Nginx 反向代理，
+并通过固定出口 IP 白名单和限流保护公网接口。
 
 ```bash
-export CHUI_ZHI_API_TOKEN='replace-with-a-strong-token'
 sh run_api.sh prod
 ```
 
 - `POST /api/v1/crawler/videos/query`
 - `GET /health` 和 `GET /ready` 仅用于本机健康检查，Nginx 示例默认禁止公网访问
-- 调用方必须通过 `X-API-Key` 传递 `CHUI_ZHI_API_TOKEN`
+- 业务接口不校验 Token，公网部署必须在 Nginx 层配置访问控制
 - `start_time`、`end_time` 推荐传13位毫秒时间戳，API 会统一转换为东八区数据库查询时间
 - 未传时间时默认按 `create_time` 查询最近3天；传入 `keywords` 时按 `video_title` 模糊匹配
 - 响应通过 `has_more` 和 `next_cursor` 表示是否存在下一页；调用方下一页原样回传 `cursor`
 - 每次API调用都会通过有界队列批量向阿里云SLS上报URL、请求参数、SQL结果长度、查询耗时、请求总耗时、成功状态、失败阶段、异常类型、HTTP状态码、request_id和错误消息；异常时 `message` 包含脱敏后的原始堆栈；本地日志仅记录
   URL、请求参数、状态码，失败时额外记录错误原因
+- API访问日志默认写入独立的 `crawler-log-prod / crawler-api-access`，不再与爬虫日志 `crawler-fetch` 混合
 - 可通过 `API_HOST`、`API_PORT`、`API_MAX_CONCURRENT_REQUESTS`、`DB_POOL_SIZE`、`API_QUERY_TIMEOUT`、`API_QUEUE_TIMEOUT`、`API_LOG_QUEUE_SIZE`、`API_MAX_LIMIT` 调整运行参数
-- 当前只提供垂直视频查询接口，路由在 `api/app.py` 中直接注册，参数校验、SQL 构造和查询逻辑集中在
-  `api/chui_zhi/videos.py`
+- API 使用 FastAPI/Uvicorn；稳定启动入口为 `api/app.py`，应用生命周期和请求上下文在
+  `api/fastapi_app.py`
+- `api/base.py` 统一封装路由注册、执行、响应、异常和数据库查询保护，`api/reporting.py` 统一访问日志；垂直视频的入参模型、SQL、查询和分页全部集中在
+  `api/chui_zhi/videos.py` 的 `VideoQueryApi` 子类中
+- 本机可通过 `/docs`、`/redoc` 和 `/openapi.json` 查看接口文档；Nginx 默认不向公网暴露这些路径
 
 筛选条件使用 `field/operator/value` 结构；没有筛选条件时传空数组或省略 `filters`：
 
@@ -104,7 +107,9 @@ API在筛选范围内按 `out_video_id` 分组并保留最大 `id` 的记录，
 # 1. 配置 API 环境变量
 API_HOST=127.0.0.1
 API_PORT=8888
-CHUI_ZHI_API_TOKEN=replace-with-a-strong-token
+API_ALIYUN_LOG_PROJECT=crawler-log-prod
+API_ALIYUN_LOGSTORE=crawler-api-access
+API_ALIYUN_LOG_ENDPOINT=cn-hangzhou.log.aliyuncs.com
 
 # 2. 使用现有部署脚本同时启动主爬虫和API
 bash deploy.sh

api/app.py

@@ -1,474 +1,13 @@
-import asyncio
-import hmac
-import json
-import re
-import time
-import traceback
-import uuid
-from typing import Any
+"""FastAPI稳定ASGI入口，部署和外部导入统一使用 api.app。"""
+from api.fastapi_app import API_PATH, app, create_app, main
 
-from aiohttp import web
-from pydantic import ValidationError
 
-from api.chui_zhi.videos import (
-    MYSQL_KEY,
-    QUERY_METRICS_KEY,
-    QUERY_SEMAPHORE_KEY,
-    BusinessValidationError,
-    DatabaseQueryError,
-    ServiceBusyError,
-    api_response,
-    query_videos,
-)
-from config import settings
-from core.base.async_mysql_client import AsyncMySQLClient
-from core.utils.log.logger_manager import LoggerManager
-
-
-LOGGER_KEY = web.AppKey('logger', object)
-ALIYUN_LOGGER_KEY = web.AppKey('aliyun_logger', object)
-CLOUD_LOG_QUEUE_KEY = web.AppKey('cloud_log_queue', asyncio.Queue)
-CLOUD_LOG_WORKER_KEY = web.AppKey('cloud_log_worker', asyncio.Task)
-ERROR_REASON_KEY = web.AppKey('error_reason', str)
-ERROR_TRACEBACK_KEY = web.AppKey('error_traceback', str)
-ERROR_TYPE_KEY = web.AppKey('error_type', str)
-FAILURE_STAGE_KEY = web.AppKey('failure_stage', str)
-REQUEST_ID_KEY = web.AppKey('request_id', str)
-REQUEST_PARAMS_KEY = web.AppKey('request_params', dict)
-REQUEST_STARTED_AT_KEY = web.AppKey('request_started_at', float)
-API_PATH = '/api/v1/crawler/videos/query'
-HEALTH_PATH = '/health'
-READY_PATH = '/ready'
-AUTH_EXEMPT_PATHS = frozenset({HEALTH_PATH, READY_PATH})
-MAX_CLOUD_LOG_VALUE_LENGTH = 64 * 1024
-MAX_LOG_BATCH_SIZE = 50
-MAX_LOG_RETRIES = 3
-SENSITIVE_FIELD_PARTS = ('authorization', 'api_key', 'apikey', 'token', 'password', 'secret')
-
-
-def print_api_routes(app: web.Application) -> None:
-    print('\n已注册 API：', flush=True)
-    for route in app.router.routes():
-        path = route.resource.canonical
-        handler_name = getattr(route.handler, '__name__', route.handler.__class__.__name__)
-        print(f'  {route.method:<6} {path:<36} -> {handler_name}', flush=True)
-    print('', flush=True)
-
-
-def parse_json_text(text: str):
-    """尽量将日志内容还原为JSON；非JSON内容保留原字符串。"""
-    if not text:
-        return None
-    try:
-        return json.loads(text)
-    except (TypeError, ValueError):
-        return text
-
-
-def format_validation_error(exc: ValidationError) -> str:
-    """把Pydantic错误转换为调用方可直接定位的简洁参数信息。"""
-    messages = []
-    for error in exc.errors(include_url=False):
-        location = '.'.join(str(item) for item in error.get('loc', ())) or 'body'
-        error_type = error.get('type', '')
-        input_value = str(error.get('input', ''))[:100]
-        context = error.get('ctx') or {}
-        if error_type == 'extra_forbidden':
-            message = f'不支持的参数: {location}'
-        elif error_type == 'literal_error':
-            message = f'{location}不支持值 {input_value}，允许值为{context.get("expected", "白名单值")}'
-        elif error_type == 'missing':
-            message = f'{location}不能为空'
-        elif error_type == 'list_too_long':
-            message = f'{location}最多允许{context.get("max_length")}项'
-        elif error_type == 'list_too_short':
-            message = f'{location}至少需要{context.get("min_length")}项'
-        elif error_type == 'less_than_equal':
-            message = f'{location}不能大于{context.get("le")}'
-        elif error_type == 'greater_than_equal':
-            message = f'{location}不能小于{context.get("ge")}'
-        elif error_type == 'value_error':
-            detail = error.get('msg', '').removeprefix('Value error, ')
-            message = f'{location}: {detail}'
-        else:
-            message = f'{location}: {error.get("msg", "参数格式错误")}'
-        messages.append(message)
-    return '参数校验失败: ' + '; '.join(messages)
-
-
-def redact_value(value: Any):
-    """递归脱敏凭证字段，避免未来扩展请求参数时意外泄露。"""
-    if isinstance(value, dict):
-        return {
-            key: '***' if any(part in str(key).lower() for part in SENSITIVE_FIELD_PARTS)
-            else redact_value(item)
-            for key, item in value.items()
-        }
-    if isinstance(value, list):
-        return [redact_value(item) for item in value]
-    return value
-
-
-def redact_text(value: str) -> str:
-    """保留异常堆栈，同时移除配置中已知的密钥内容。"""
-    result = value
-    for secret in (
-        settings.CHUI_ZHI_API_TOKEN,
-        settings.DB_PASSWORD,
-        settings.ALIYUN_ACCESS_KEY_ID,
-        settings.ALIYUN_ACCESS_KEY_SECRET,
-    ):
-        if secret:
-            result = result.replace(str(secret), '***')
-    return result
-
-
-def limit_cloud_log_value(value):
-    """限制单个日志字段大小，避免请求参数超过SLS单条日志限制。"""
-    value = redact_value(value)
-    text = json.dumps(value, ensure_ascii=False, default=str)
-    if len(text) <= MAX_CLOUD_LOG_VALUE_LENGTH:
-        return value
-    return {
-        'truncated': True,
-        'original_length': len(text),
-        'content': text[:MAX_CLOUD_LOG_VALUE_LENGTH],
-    }
-
-
-def get_request_url(request: web.Request) -> str:
-    scheme = request.headers.get('X-Forwarded-Proto', request.scheme).split(',', 1)[0].strip()
-    host = request.headers.get('X-Forwarded-Host', request.host).split(',', 1)[0].strip()
-    return f'{scheme}://{host}{request.rel_url}'
-
-
-async def send_cloud_log_batch(app: web.Application, events: list[dict]) -> None:
-    await asyncio.wait_for(
-        asyncio.to_thread(app[ALIYUN_LOGGER_KEY].logging_batch, events),
-        timeout=settings.API_LOG_FLUSH_TIMEOUT,
-    )
-
-
-async def cloud_log_worker(app: web.Application) -> None:
-    """后台批量上报SLS，日志服务异常不阻塞业务请求。"""
-    queue = app[CLOUD_LOG_QUEUE_KEY]
-    logger = app[LOGGER_KEY]
-    while True:
-        event = await queue.get()
-        if event is None:
-            queue.task_done()
-            return
-
-        events = [event]
-        while len(events) < MAX_LOG_BATCH_SIZE:
-            try:
-                next_event = queue.get_nowait()
-            except asyncio.QueueEmpty:
-                break
-            if next_event is None:
-                queue.task_done()
-                break
-            events.append(next_event)
-
-        try:
-            for attempt in range(MAX_LOG_RETRIES):
-                try:
-                    await send_cloud_log_batch(app, events)
-                    break
-                except Exception:
-                    if attempt + 1 >= MAX_LOG_RETRIES:
-                        logger.exception(f'阿里云API日志批量上报失败: count={len(events)}')
-                    else:
-                        await asyncio.sleep(0.5 * (2 ** attempt))
-        finally:
-            for _ in events:
-                queue.task_done()
-
-
-async def enqueue_cloud_log(app: web.Application, event: dict) -> None:
-    queue = app.get(CLOUD_LOG_QUEUE_KEY)
-    if queue is None:
-        # 单元测试或嵌入运行未启动cleanup context时，仍可验证日志行为。
-        try:
-            await send_cloud_log_batch(app, [event])
-        except Exception:
-            app[LOGGER_KEY].exception('阿里云API日志上报失败')
-        return
-    try:
-        queue.put_nowait(event)
-    except asyncio.QueueFull:
-        app[LOGGER_KEY].error(
-            f'阿里云API日志队列已满，丢弃日志: request_id={event.get("trace_id", "")}'
-        )
-
-
-async def report_api_request(
-    request: web.Request,
-    response: web.StreamResponse | None,
-    exception: Exception | None = None,
-) -> None:
-    """本地日志同步落地，SLS日志仅入队，不增加接口响应延迟。"""
-    logger = request.app[LOGGER_KEY]
-    status_code = response.status if response is not None else getattr(exception, 'status', 500)
-    request_duration_ms = round(
-        (time.perf_counter() - request.get(REQUEST_STARTED_AT_KEY, time.perf_counter())) * 1000,
-        2,
-    )
-
-    request_params = request.get(REQUEST_PARAMS_KEY, {'query': dict(request.query), 'body': None})
-    safe_request_params = limit_cloud_log_value(request_params)
-    request_id = request[REQUEST_ID_KEY]
-    error_message = request.get(ERROR_TRACEBACK_KEY, '') or request.get(ERROR_REASON_KEY, '')
-    if not error_message and exception is not None:
-        error_message = f'{type(exception).__name__}: {exception}'
-    error_message = redact_text(error_message)
-    error_type = request.get(ERROR_TYPE_KEY, '')
-    if not error_type and exception is not None:
-        error_type = type(exception).__name__
-    failure_stage = request.get(FAILURE_STAGE_KEY, '')
-
-    request_url = get_request_url(request)
-    local_message = (
-        f'API请求 request_id={request_id} method={request.method} url={request_url} '
-        f'params={json.dumps(safe_request_params, ensure_ascii=False, default=str)} '
-        f'status={status_code} duration_ms={request_duration_ms}'
-    )
-    if error_message:
-        logger.error(f'{local_message} error={error_message}')
-    else:
-        logger.info(local_message)
-
-    query_metrics = request.get(QUERY_METRICS_KEY, {})
-    request_body = request_params.get('body') if isinstance(request_params, dict) else None
-    request_body = request_body if isinstance(request_body, dict) else {}
-    cloud_data = {
-        'url': request_url,
-        'path': request.path,
-        'method': request.method,
-        'request_id': request_id,
-        'request_params': safe_request_params,
-        'platforms': request_body.get('platforms'),
-        'status_code': status_code,
-        'success': status_code < 400,
-        'request_duration_ms': request_duration_ms,
-        'failure_stage': failure_stage,
-        'error_type': error_type,
-        'message': error_message,
-        'query_result_count': query_metrics.get('result_count'),
-        'query_duration_ms': query_metrics.get('duration_ms'),
-        'query_success': query_metrics.get('success'),
-    }
-    await enqueue_cloud_log(request.app, {
-        'code': '2000' if status_code < 400 else '9000',
-        'message': 'API请求成功' if status_code < 400 else error_message,
-        'data': cloud_data,
-        'trace_id': request_id,
-    })
-
-
-@web.middleware
-async def request_context_middleware(request: web.Request, handler):
-    """建立请求上下文，确保成功、失败和未鉴权请求都有访问日志。"""
-    incoming_request_id = request.headers.get('X-Request-ID', '').strip()
-    if not re.fullmatch(r'[A-Za-z0-9._-]{1,128}', incoming_request_id):
-        incoming_request_id = uuid.uuid4().hex
-    request[REQUEST_ID_KEY] = incoming_request_id
-    request[REQUEST_STARTED_AT_KEY] = time.perf_counter()
-    request[REQUEST_PARAMS_KEY] = {
-        'query': dict(request.query),
-        'body': None,
-    }
-    response = None
-    exception = None
-    try:
-        response = await handler(request)
-        response.headers['X-Request-ID'] = request[REQUEST_ID_KEY]
-        return response
-    except Exception as exc:
-        exception = exc
-        request[ERROR_TRACEBACK_KEY] = traceback.format_exc()
-        raise
-    finally:
-        try:
-            await report_api_request(request, response, exception)
-        except Exception:
-            # 日志异常不能覆盖API原始响应或异常。
-            request.app[LOGGER_KEY].exception('API访问日志记录失败')
-
-
-@web.middleware
-async def request_body_middleware(request: web.Request, handler):
-    """鉴权通过后再读取请求体，避免未授权请求消耗JSON解析资源。"""
-    raw_body = await request.read()
-    request[REQUEST_PARAMS_KEY]['body'] = parse_json_text(
-        raw_body.decode('utf-8', errors='replace')
-    )
-    return await handler(request)
-
-
-@web.middleware
-async def error_middleware(request: web.Request, handler):
-    try:
-        return await handler(request)
-    except json.JSONDecodeError as exc:
-        request[ERROR_REASON_KEY] = str(exc)
-        request[ERROR_TRACEBACK_KEY] = traceback.format_exc()
-        request[ERROR_TYPE_KEY] = type(exc).__name__
-        request[FAILURE_STAGE_KEY] = 'request_validation'
-        return api_response(code=400, msg=f'请求JSON格式错误: {exc.msg}')
-    except ValidationError as exc:
-        message = format_validation_error(exc)
-        request[ERROR_REASON_KEY] = message
-        request[ERROR_TRACEBACK_KEY] = traceback.format_exc()
-        request[ERROR_TYPE_KEY] = type(exc).__name__
-        request[FAILURE_STAGE_KEY] = 'request_validation'
-        return api_response(code=400, msg=message, data=exc.errors(include_url=False))
-    except BusinessValidationError as exc:
-        request[ERROR_REASON_KEY] = str(exc)
-        request[ERROR_TRACEBACK_KEY] = traceback.format_exc()
-        request[ERROR_TYPE_KEY] = type(exc).__name__
-        request[FAILURE_STAGE_KEY] = 'business_validation'
-        return api_response(code=422, msg=str(exc))
-    except ServiceBusyError as exc:
-        request[ERROR_REASON_KEY] = str(exc) or 'server busy'
-        request[ERROR_TRACEBACK_KEY] = traceback.format_exc()
-        request[ERROR_TYPE_KEY] = type(exc).__name__
-        request[FAILURE_STAGE_KEY] = 'concurrency_limit'
-        return api_response(code=503, msg='server busy')
-    except asyncio.TimeoutError as exc:
-        request[ERROR_REASON_KEY] = str(exc) or 'query timeout'
-        request[ERROR_TRACEBACK_KEY] = traceback.format_exc()
-        request[ERROR_TYPE_KEY] = type(exc).__name__
-        request[FAILURE_STAGE_KEY] = 'database_query'
-        return api_response(code=504, msg='query timeout')
-    except DatabaseQueryError as exc:
-        request[ERROR_REASON_KEY] = str(exc)
-        request[ERROR_TRACEBACK_KEY] = traceback.format_exc()
-        request[ERROR_TYPE_KEY] = type(exc).__name__
-        request[FAILURE_STAGE_KEY] = 'database_query'
-        request.app[LOGGER_KEY].exception(str(exc))
-        return api_response(code=500, msg='database query failed')
-    except web.HTTPException:
-        raise
-    except Exception as exc:
-        request[ERROR_REASON_KEY] = f'{type(exc).__name__}: {exc}'
-        request[ERROR_TRACEBACK_KEY] = traceback.format_exc()
-        request[ERROR_TYPE_KEY] = type(exc).__name__
-        request[FAILURE_STAGE_KEY] = 'internal'
-        request.app[LOGGER_KEY].exception(f'API request failed: {exc}')
-        return api_response(code=500, msg='internal server error')
-
-
-@web.middleware
-async def auth_middleware(request: web.Request, handler):
-    if request.path in AUTH_EXEMPT_PATHS:
-        return await handler(request)
-    supplied_token = request.headers.get('X-API-Key', '')
-    if not hmac.compare_digest(supplied_token, settings.CHUI_ZHI_API_TOKEN):
-        request[ERROR_REASON_KEY] = 'unauthorized'
-        request[ERROR_TYPE_KEY] = 'AuthenticationError'
-        request[FAILURE_STAGE_KEY] = 'authentication'
-        return api_response(code=401, msg='unauthorized')
-    return await handler(request)
-
-
-async def health(_request: web.Request) -> web.Response:
-    return api_response({'status': 'ok'})
-
-
-async def ready(request: web.Request) -> web.Response:
-    try:
-        async with asyncio.timeout(min(3, settings.API_QUERY_TIMEOUT)):
-            row = await request.app[MYSQL_KEY].fetch_one('SELECT 1 AS ok')
-        if not row or row.get('ok') != 1:
-            raise DatabaseQueryError('database readiness check failed')
-        return api_response({'status': 'ready'})
-    except Exception as exc:
-        request[ERROR_REASON_KEY] = f'{type(exc).__name__}: {exc}'
-        request[ERROR_TRACEBACK_KEY] = traceback.format_exc()
-        request[ERROR_TYPE_KEY] = type(exc).__name__
-        request[FAILURE_STAGE_KEY] = 'readiness'
-        return api_response(code=503, msg='not ready')
-
-
-async def app_context(app: web.Application):
-    if not settings.CHUI_ZHI_API_TOKEN:
-        raise RuntimeError('CHUI_ZHI_API_TOKEN未配置，拒绝启动垂直spider API')
-
-    logger = LoggerManager.get_logger(platform='chui_zhi', mode='api')
-    aliyun_logger = LoggerManager.get_aliyun_logger(platform='chui_zhi', mode='api')
-    mysql = AsyncMySQLClient(
-        host=settings.DB_HOST,
-        port=settings.DB_PORT,
-        user=settings.DB_USER,
-        password=settings.DB_PASSWORD,
-        db=settings.DB_NAME,
-        charset=settings.DB_CHARSET,
-        minsize=min(5, settings.DB_POOL_SIZE),
-        maxsize=settings.DB_POOL_SIZE,
-        pool_recycle=settings.DB_POOL_RECYCLE,
-        logger=logger,
-        # API中间件统一上报异常，避免数据库客户端同步调用SLS。
-        aliyun_logr=None,
-    )
-    await mysql.init_pool()
-    query_concurrency = min(settings.API_MAX_CONCURRENT_REQUESTS, settings.DB_POOL_SIZE)
-    app[MYSQL_KEY] = mysql
-    app[QUERY_SEMAPHORE_KEY] = asyncio.Semaphore(query_concurrency)
-    app[LOGGER_KEY] = logger
-    app[ALIYUN_LOGGER_KEY] = aliyun_logger
-    app[CLOUD_LOG_QUEUE_KEY] = asyncio.Queue(maxsize=settings.API_LOG_QUEUE_SIZE)
-    app[CLOUD_LOG_WORKER_KEY] = asyncio.create_task(cloud_log_worker(app))
-    logger.info(
-        f'垂直spider API启动: db_pool={settings.DB_POOL_SIZE}, '
-        f'query_concurrency={query_concurrency}'
-    )
-    print_api_routes(app)
-    yield
-
-    queue = app[CLOUD_LOG_QUEUE_KEY]
-    try:
-        await asyncio.wait_for(queue.join(), timeout=settings.API_LOG_FLUSH_TIMEOUT)
-    except asyncio.TimeoutError:
-        logger.error(f'API关闭时日志队列未完全清空: remaining={queue.qsize()}')
-        app[CLOUD_LOG_WORKER_KEY].cancel()
-        try:
-            await app[CLOUD_LOG_WORKER_KEY]
-        except asyncio.CancelledError:
-            pass
-    else:
-        await queue.put(None)
-        await app[CLOUD_LOG_WORKER_KEY]
-    await mysql.close()
-    logger.info('垂直spider API已关闭')
-
-
-def create_app() -> web.Application:
-    app = web.Application(
-        middlewares=[
-            request_context_middleware,
-            error_middleware,
-            auth_middleware,
-            request_body_middleware,
-        ],
-        client_max_size=1024 * 1024,
-    )
-    app.cleanup_ctx.append(app_context)
-    app.add_routes([
-        web.post(API_PATH, query_videos, name='chui_zhi_videos'),
-        web.get(HEALTH_PATH, health, name='health'),
-        web.get(READY_PATH, ready, name='ready'),
-    ])
-    return app
-
-
-def main():
-    web.run_app(
-        create_app(),
-        host=settings.API_HOST,
-        port=settings.API_PORT,
-        access_log=None,
-    )
+__all__ = [
+    'API_PATH',
+    'app',
+    'create_app',
+    'main',
+]
 
 
 if __name__ == '__main__':

api/base.py

@@ -0,0 +1,148 @@
+import asyncio
+import time
+from abc import ABC, abstractmethod
+from functools import wraps
+from typing import Any
+
+from fastapi import APIRouter, Request
+from fastapi.exceptions import RequestValidationError
+from pydantic import BaseModel, ConfigDict
+from starlette.exceptions import HTTPException as StarletteHTTPException
+from starlette.responses import Response
+
+from api.common import api_response
+from api.errors import BusinessValidationError, DatabaseQueryError, ServiceBusyError
+from api.reporting import format_validation_errors, report_api_request, set_failure
+from config import settings
+
+
+class ApiParams(BaseModel):
+    """所有API请求参数的公共基类，默认拒绝未声明字段。"""
+
+    model_config = ConfigDict(extra='forbid')
+
+
+def unified_endpoint(api):
+    """保留logic签名供FastAPI解析，并由基类统一完成执行和出口处理。"""
+    @wraps(api.logic)
+    async def wrapper(*args, **kwargs):
+        return await api.dispatch(*args, **kwargs)
+
+    return wrapper
+
+
+class BaseApi(ABC):
+    """API模板：统一注册、执行、响应、异常和日志，子类只实现业务。"""
+
+    path: str
+    methods: tuple[str, ...] = ('POST',)
+
+    def register(self, router: APIRouter) -> None:
+        router.add_api_route(
+            self.path,
+            unified_endpoint(self),
+            methods=list(self.methods),
+            name=self.__class__.__name__,
+            summary=(self.__doc__ or '').strip().splitlines()[0] or None,
+        )
+
+    async def dispatch(self, *args, **kwargs) -> Response:
+        request = self._find_request(args, kwargs)
+        try:
+            data = await self.logic(*args, **kwargs)
+            response = data if isinstance(data, Response) else api_response(data)
+        except Exception as exc:
+            response = self.exception_response(request, exc)
+
+        if request is not None:
+            response.headers['X-Request-ID'] = request.state.request_id
+            await self.report_once(request, response)
+        return response
+
+    async def fetch_all(self, request: Request, sql: str, params: list[Any]) -> list[dict]:
+        """统一执行列表查询，并记录并发、超时、耗时和结果数量。"""
+        semaphore = request.app.state.query_semaphore
+        try:
+            await asyncio.wait_for(semaphore.acquire(), timeout=settings.API_QUEUE_TIMEOUT)
+        except asyncio.TimeoutError as exc:
+            raise ServiceBusyError from exc
+
+        query_started_at = time.perf_counter()
+        try:
+            async with asyncio.timeout(settings.API_QUERY_TIMEOUT):
+                rows = await request.app.state.mysql.fetch_all(sql, params)
+            request.state.query_metrics = {
+                'result_count': len(rows),
+                'duration_ms': round((time.perf_counter() - query_started_at) * 1000, 2),
+                'success': True,
+            }
+            return rows
+        except asyncio.TimeoutError:
+            request.state.query_metrics = {
+                'result_count': None,
+                'duration_ms': round((time.perf_counter() - query_started_at) * 1000, 2),
+                'success': False,
+            }
+            raise
+        except Exception as exc:
+            request.state.query_metrics = {
+                'result_count': None,
+                'duration_ms': round((time.perf_counter() - query_started_at) * 1000, 2),
+                'success': False,
+            }
+            raise DatabaseQueryError('database query failed') from exc
+        finally:
+            semaphore.release()
+
+    @staticmethod
+    def _find_request(args, kwargs) -> Request | None:
+        candidates = (*args, *kwargs.values())
+        return next((item for item in candidates if isinstance(item, Request)), None)
+
+    @staticmethod
+    def exception_response(request: Request | None, exc: Exception) -> Response:
+        """所有API共用的异常到响应映射。"""
+        if request is None:
+            raise exc
+        if isinstance(exc, RequestValidationError):
+            errors = exc.errors()
+            message = format_validation_errors(errors)
+            set_failure(request, 'request_validation', exc, message)
+            return api_response(code=400, msg=message, data=errors)
+        if isinstance(exc, BusinessValidationError):
+            set_failure(request, 'business_validation', exc)
+            return api_response(code=422, msg=str(exc))
+        if isinstance(exc, ServiceBusyError):
+            set_failure(request, 'concurrency_limit', exc, str(exc) or 'server busy')
+            return api_response(code=503, msg='server busy')
+        if isinstance(exc, asyncio.TimeoutError):
+            set_failure(request, 'database_query', exc, str(exc) or 'query timeout')
+            return api_response(code=504, msg='query timeout')
+        if isinstance(exc, DatabaseQueryError):
+            set_failure(request, 'database_query', exc)
+            request.app.state.logger.exception(str(exc))
+            return api_response(code=500, msg='database query failed')
+        if isinstance(exc, StarletteHTTPException):
+            set_failure(request, 'routing', exc, str(exc.detail))
+            return api_response(code=exc.status_code, msg=str(exc.detail))
+
+        set_failure(request, 'internal', exc, f'{type(exc).__name__}: {exc}')
+        request.app.state.logger.exception(f'API request failed: {exc}')
+        return api_response(code=500, msg='internal server error')
+
+    @staticmethod
+    async def report_once(request: Request, response: Response) -> None:
+        """保证一个请求最多记录和上报一次访问日志。"""
+        if getattr(request.state, 'api_reported', False):
+            return
+        try:
+            await report_api_request(request, response)
+        except Exception:
+            request.app.state.logger.exception('API访问日志记录失败')
+        finally:
+            request.state.api_reported = True
+
+    @abstractmethod
+    async def logic(self, *args, **kwargs):
+        """只实现参数对应的业务逻辑，并直接返回业务data。"""
+        raise NotImplementedError

api/chui_zhi/videos.py

@@ -1,20 +1,16 @@
-import asyncio
-import json
 import math
-import time
 from datetime import datetime, timedelta, timezone
 from typing import Any, List, Literal, Optional, Tuple
 
-from aiohttp import web
-from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
+from fastapi import APIRouter, Request
+from pydantic import Field, field_validator, model_validator
 
+from api.base import ApiParams, BaseApi
+from api.errors import BusinessValidationError
 from config import settings
-from core.base.async_mysql_client import AsyncMySQLClient
 
 
-MYSQL_KEY = web.AppKey('mysql', AsyncMySQLClient)
-QUERY_SEMAPHORE_KEY = web.AppKey('query_semaphore', asyncio.Semaphore)
-QUERY_METRICS_KEY = web.AppKey('query_metrics', dict)
+router = APIRouter(prefix='/api/v1/crawler/videos', tags=['垂直视频'])
 CHINA_TIMEZONE = timezone(timedelta(hours=8))
 
 FilterField = Literal[
@@ -32,25 +28,11 @@ SUPPORTED_PLATFORMS = frozenset({'xiaoniangao', 'xiaoniangaotuijianliu'})
 MAX_FILTER_SET_VALUES = 100
 
 
-class ServiceBusyError(Exception):
-    """API并发已满，暂时无法执行查询。"""
-
-
-class BusinessValidationError(Exception):
-    """请求格式正确，但业务筛选值无法执行。"""
-
-
-class DatabaseQueryError(Exception):
-    """数据库查询执行失败。"""
-
-
 # ==================== 请求参数 ====================
 
-class FilterCondition(BaseModel):
+class FilterCondition(ApiParams):
     """单个结构化筛选条件，字段和操作符均通过白名单限制。"""
 
-    model_config = ConfigDict(extra='forbid')
-
     field: FilterField
     operator: FilterOperator
     value: Any
@@ -62,16 +44,14 @@ class FilterCondition(BaseModel):
         return self
 
 
-class PageCursor(BaseModel):
+class PageCursor(ApiParams):
     """稳定翻页游标，对应上一页最后一条数据的自增主键。"""
 
-    model_config = ConfigDict(extra='forbid')
-
     id: int = Field(gt=0)
 
 
-class VideoQueryRequest(BaseModel):
-    model_config = ConfigDict(extra='forbid')
+class VideoQueryParams(ApiParams):
+    """查询垂直视频的请求参数。"""
 
     platforms: List[str] = Field(
         default_factory=lambda: ['xiaoniangao', 'xiaoniangaotuijianliu'],
@@ -229,7 +209,7 @@ def compile_filter_condition(
 
 
 def build_filter_scope(
-    params: VideoQueryRequest,
+    params: VideoQueryParams,
     table_alias: Optional[str] = None,
 ) -> SqlFragment:
     """生成可复用的筛选范围，供主查询和去重子查询保持一致。"""
@@ -260,7 +240,7 @@ def build_filter_scope(
     return ' AND '.join(clauses), sql_params
 
 
-def build_query(params: VideoQueryRequest) -> SqlFragment:
+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)
@@ -289,17 +269,7 @@ def build_query(params: VideoQueryRequest) -> SqlFragment:
     return sql, sql_params
 
 
-# ==================== 接口处理与返回 ====================
-
-def api_response(data=None, code: int = 0, msg: str = '') -> web.Response:
-    response = web.json_response(
-        {'code': code, 'msg': msg, 'data': data},
-        status=200 if code == 0 else code,
-        dumps=lambda value: json.dumps(value, ensure_ascii=False, default=str),
-    )
-    response.headers['Cache-Control'] = 'no-store'
-    return response
-
+# ==================== 接口实现 ====================
 
 def timestamp_ms(value: datetime) -> int:
     if value.tzinfo is None:
@@ -309,65 +279,27 @@ def timestamp_ms(value: datetime) -> int:
     return int(value.timestamp() * 1000)
 
 
-async def execute_query(request: web.Request, sql: str, sql_params: List[Any]) -> List[dict]:
-    """在并发限制和查询超时保护下执行数据库查询。"""
-    semaphore = request.app[QUERY_SEMAPHORE_KEY]
-    try:
-        await asyncio.wait_for(semaphore.acquire(), timeout=settings.API_QUEUE_TIMEOUT)
-    except asyncio.TimeoutError as exc:
-        raise ServiceBusyError from exc
+class VideoQueryApi(BaseApi):
+    """管理视频查询的入参、SQL、分页和业务返回数据。"""
 
-    query_started_at = time.perf_counter()
-    try:
-        async with asyncio.timeout(settings.API_QUERY_TIMEOUT):
-            rows = await request.app[MYSQL_KEY].fetch_all(sql, sql_params)
-        request[QUERY_METRICS_KEY] = {
-            'result_count': len(rows),
-            'duration_ms': round((time.perf_counter() - query_started_at) * 1000, 2),
-            'success': True,
-        }
-        return rows
-    except asyncio.TimeoutError:
-        request[QUERY_METRICS_KEY] = {
-            'result_count': None,
-            'duration_ms': round((time.perf_counter() - query_started_at) * 1000, 2),
-            'success': False,
-        }
-        raise
-    except Exception as exc:
-        request[QUERY_METRICS_KEY] = {
-            'result_count': None,
-            'duration_ms': round((time.perf_counter() - query_started_at) * 1000, 2),
-            'success': False,
-        }
-        raise DatabaseQueryError('database query failed') from exc
-    finally:
-        semaphore.release()
+    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)
 
-async def query_videos(request: web.Request) -> web.Response:
-    """查询一页垂直视频，并返回下一页信息。"""
-    # 1. 校验请求参数，并补齐默认三天时间范围。
-    params = VideoQueryRequest.model_validate(await request.json())
+        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
+        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),
+        }
 
-    # 2. 生成参数化SQL并查询数据库。
-    sql, sql_params = build_query(params)
-    rows = await execute_query(request, sql, sql_params)
 
-    # 3. 截取当前页，并用最后一条数据的自增主键生成稳定游标。
-    page_size = min(params.limit, settings.API_MAX_LIMIT)
-    has_more = len(rows) > page_size
-    rows = rows[:page_size]
-    next_cursor = None
-    if has_more and rows:
-        last_row = rows[-1]
-        next_cursor = {'id': last_row['id']}
-    response_data = {
-        '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),
-    }
-    return api_response(response_data)
+VideoQueryApi().register(router)

api/common.py

@@ -0,0 +1,17 @@
+import json
+
+from starlette.responses import Response
+
+
+def api_response(data=None, code: int = 0, msg: str = '') -> Response:
+    """统一生成API响应，保持既有code/msg/data协议。"""
+    return Response(
+        content=json.dumps(
+            {'code': code, 'msg': msg, 'data': data},
+            ensure_ascii=False,
+            default=str,
+        ),
+        status_code=200 if code == 0 else code,
+        media_type='application/json',
+        headers={'Cache-Control': 'no-store'},
+    )

api/errors.py

@@ -0,0 +1,10 @@
+class ServiceBusyError(Exception):
+    """API并发已满，暂时无法执行请求。"""
+
+
+class BusinessValidationError(Exception):
+    """请求结构正确，但业务参数无法执行。"""
+
+
+class DatabaseQueryError(Exception):
+    """数据库查询执行失败。"""

api/fastapi_app.py

@@ -0,0 +1,200 @@
+import asyncio
+import json
+import re
+import time
+import uuid
+from contextlib import asynccontextmanager
+
+import uvicorn
+from fastapi import FastAPI, Request
+from fastapi.exceptions import RequestValidationError
+from starlette.exceptions import HTTPException as StarletteHTTPException
+
+from api.base import BaseApi
+from api.common import api_response
+from api.errors import BusinessValidationError, DatabaseQueryError, ServiceBusyError
+from api.reporting import cloud_log_worker, set_failure
+from api.routes import api_router
+from config import settings
+from core.base.async_mysql_client import AsyncMySQLClient
+from core.utils.log.logger_manager import LoggerManager
+
+
+API_PATH = '/api/v1/crawler/videos/query'
+HEALTH_PATH = '/health'
+READY_PATH = '/ready'
+MAX_REQUEST_BODY_SIZE = 1024 * 1024
+
+
+def print_api_routes(app: FastAPI) -> None:
+    def iter_routes(routes):
+        for route in routes:
+            included_router = getattr(route, 'original_router', None)
+            if included_router is not None:
+                yield from iter_routes(included_router.routes)
+            else:
+                yield route
+
+    print('\n已注册 API：', flush=True)
+    for route in iter_routes(app.routes):
+        path = getattr(route, 'path', None)
+        methods = ','.join(sorted(getattr(route, 'methods', None) or []))
+        if path and methods:
+            print(
+                f'  {methods:<12} {path:<36} -> {getattr(route, "name", "")}',
+                flush=True,
+            )
+    print('', flush=True)
+
+
+def parse_json_text(text: str):
+    if not text:
+        return None
+    try:
+        return json.loads(text)
+    except (TypeError, ValueError):
+        return text
+
+
+async def access_middleware(request: Request, call_next):
+    """统一入口和出口：上下文、Body、异常响应、请求ID和访问日志全部在此闭环。"""
+    incoming_request_id = request.headers.get('X-Request-ID', '').strip()
+    if not re.fullmatch(r'[A-Za-z0-9._-]{1,128}', incoming_request_id):
+        incoming_request_id = uuid.uuid4().hex
+    request.state.request_id = incoming_request_id
+    request.state.started_at = time.perf_counter()
+    request.state.request_params = {'query': dict(request.query_params), 'body': None}
+
+    response = None
+    try:
+        raw_body = await request.body()
+        request.state.request_params['body'] = parse_json_text(
+            raw_body.decode('utf-8', errors='replace')
+        )
+        if len(raw_body) > MAX_REQUEST_BODY_SIZE:
+            exc = ValueError('请求体不能超过1MB')
+            set_failure(request, 'request_validation', exc)
+            response = api_response(code=413, msg=str(exc))
+        else:
+            response = await call_next(request)
+    except Exception as exc:
+        response = BaseApi.exception_response(request, exc)
+
+    response.headers['X-Request-ID'] = request.state.request_id
+    # BaseApi负责业务接口上报；未进入BaseApi的校验错误、404/405及健康检查在此兜底。
+    if not getattr(request.state, 'api_reported', False):
+        await BaseApi.report_once(request, response)
+    return response
+
+
+async def api_exception_handler(request: Request, exc: Exception):
+    return BaseApi.exception_response(request, exc)
+
+
+async def health():
+    return api_response({'status': 'ok'})
+
+
+async def ready(request: Request):
+    try:
+        async with asyncio.timeout(min(3, settings.API_QUERY_TIMEOUT)):
+            row = await request.app.state.mysql.fetch_one('SELECT 1 AS ok')
+        if not row or row.get('ok') != 1:
+            raise DatabaseQueryError('database readiness check failed')
+        return api_response({'status': 'ready'})
+    except Exception as exc:
+        set_failure(request, 'readiness', exc, f'{type(exc).__name__}: {exc}')
+        return api_response(code=503, msg='not ready')
+
+
+@asynccontextmanager
+async def app_lifespan(app: FastAPI):
+    logger = LoggerManager.get_logger(platform='chui_zhi', mode='api')
+    aliyun_logger = LoggerManager.get_aliyun_logger(
+        platform='chui_zhi',
+        mode='api',
+        env=settings.ENV,
+        project=settings.API_ALIYUN_LOG_PROJECT,
+        logstore=settings.API_ALIYUN_LOGSTORE,
+        endpoint=settings.API_ALIYUN_LOG_ENDPOINT,
+    )
+    mysql = AsyncMySQLClient(
+        host=settings.DB_HOST,
+        port=settings.DB_PORT,
+        user=settings.DB_USER,
+        password=settings.DB_PASSWORD,
+        db=settings.DB_NAME,
+        charset=settings.DB_CHARSET,
+        minsize=min(5, settings.DB_POOL_SIZE),
+        maxsize=settings.DB_POOL_SIZE,
+        pool_recycle=settings.DB_POOL_RECYCLE,
+        logger=logger,
+        aliyun_logr=None,
+    )
+    await mysql.init_pool()
+    concurrency = min(settings.API_MAX_CONCURRENT_REQUESTS, settings.DB_POOL_SIZE)
+    app.state.mysql = mysql
+    app.state.query_semaphore = asyncio.Semaphore(concurrency)
+    app.state.logger = logger
+    app.state.aliyun_logger = aliyun_logger
+    app.state.cloud_log_queue = asyncio.Queue(maxsize=settings.API_LOG_QUEUE_SIZE)
+    app.state.cloud_log_worker = asyncio.create_task(cloud_log_worker(app))
+    logger.info(
+        f'垂直spider API启动: framework=fastapi, db_pool={settings.DB_POOL_SIZE}, '
+        f'query_concurrency={concurrency}, '
+        f'sls={settings.API_ALIYUN_LOG_PROJECT}/{settings.API_ALIYUN_LOGSTORE}'
+    )
+    print_api_routes(app)
+    yield
+
+    queue = app.state.cloud_log_queue
+    try:
+        await asyncio.wait_for(queue.join(), timeout=settings.API_LOG_FLUSH_TIMEOUT)
+    except asyncio.TimeoutError:
+        logger.error(f'API关闭时日志队列未完全清空: remaining={queue.qsize()}')
+        app.state.cloud_log_worker.cancel()
+        try:
+            await app.state.cloud_log_worker
+        except asyncio.CancelledError:
+            pass
+    else:
+        await queue.put(None)
+        await app.state.cloud_log_worker
+    await mysql.close()
+    logger.info('垂直spider API已关闭')
+
+
+def create_app(manage_resources: bool = True) -> FastAPI:
+    app = FastAPI(
+        title='AutoScraperX API',
+        version='1.0.0',
+        lifespan=app_lifespan if manage_resources else None,
+    )
+    app.middleware('http')(access_middleware)
+    app.add_exception_handler(RequestValidationError, api_exception_handler)
+    app.add_exception_handler(BusinessValidationError, api_exception_handler)
+    app.add_exception_handler(ServiceBusyError, api_exception_handler)
+    app.add_exception_handler(asyncio.TimeoutError, api_exception_handler)
+    app.add_exception_handler(DatabaseQueryError, api_exception_handler)
+    app.add_exception_handler(StarletteHTTPException, api_exception_handler)
+    app.add_exception_handler(Exception, api_exception_handler)
+    app.include_router(api_router)
+    app.add_api_route(HEALTH_PATH, health, methods=['GET'], name='health', include_in_schema=False)
+    app.add_api_route(READY_PATH, ready, methods=['GET'], name='ready', include_in_schema=False)
+    return app
+
+
+app = create_app()
+
+
+def main():
+    uvicorn.run(
+        app,
+        host=settings.API_HOST,
+        port=settings.API_PORT,
+        access_log=False,
+    )
+
+
+if __name__ == '__main__':
+    main()

api/reporting.py

@@ -0,0 +1,217 @@
+import asyncio
+import json
+import time
+import traceback
+from typing import Any
+
+from fastapi import Request
+from starlette.responses import Response
+
+from config import settings
+
+
+MAX_CLOUD_LOG_VALUE_LENGTH = 64 * 1024
+MAX_LOG_BATCH_SIZE = 50
+MAX_LOG_RETRIES = 3
+SENSITIVE_FIELD_PARTS = ('authorization', 'api_key', 'apikey', 'token', 'password', 'secret')
+
+
+def set_failure(request: Request, stage: str, exc: Exception, message: str | None = None) -> None:
+    request.state.failure_stage = stage
+    request.state.error_type = type(exc).__name__
+    request.state.error_reason = message if message is not None else str(exc)
+    request.state.error_traceback = traceback.format_exc()
+
+
+def format_validation_errors(errors: list[dict]) -> str:
+    """把FastAPI/Pydantic错误转换为稳定且可定位的中文信息。"""
+    messages = []
+    for error in errors:
+        location_items = list(error.get('loc', ()))
+        if location_items and location_items[0] == 'body':
+            location_items.pop(0)
+        location = '.'.join(str(item) for item in location_items) or 'body'
+        error_type = error.get('type', '')
+        input_value = str(error.get('input', ''))[:100]
+        context = error.get('ctx') or {}
+        if error_type == 'extra_forbidden':
+            message = f'不支持的参数: {location}'
+        elif error_type == 'literal_error':
+            message = f'{location}不支持值 {input_value}，允许值为{context.get("expected", "白名单值")}'
+        elif error_type == 'missing':
+            message = f'{location}不能为空'
+        elif error_type == 'list_too_long':
+            message = f'{location}最多允许{context.get("max_length")}项'
+        elif error_type == 'list_too_short':
+            message = f'{location}至少需要{context.get("min_length")}项'
+        elif error_type == 'less_than_equal':
+            message = f'{location}不能大于{context.get("le")}'
+        elif error_type == 'greater_than_equal':
+            message = f'{location}不能小于{context.get("ge")}'
+        elif error_type == 'value_error':
+            detail = error.get('msg', '').removeprefix('Value error, ')
+            message = f'{location}: {detail}'
+        else:
+            message = f'{location}: {error.get("msg", "参数格式错误")}'
+        messages.append(message)
+    return '参数校验失败: ' + '; '.join(messages)
+
+
+def _redact_value(value: Any):
+    if isinstance(value, dict):
+        return {
+            key: '***' if any(part in str(key).lower() for part in SENSITIVE_FIELD_PARTS)
+            else _redact_value(item)
+            for key, item in value.items()
+        }
+    if isinstance(value, list):
+        return [_redact_value(item) for item in value]
+    return value
+
+
+def _redact_text(value: str) -> str:
+    result = value
+    for secret in (
+        settings.DB_PASSWORD,
+        settings.ALIYUN_ACCESS_KEY_ID,
+        settings.ALIYUN_ACCESS_KEY_SECRET,
+    ):
+        if secret:
+            result = result.replace(str(secret), '***')
+    return result
+
+
+def _limit_log_value(value):
+    value = _redact_value(value)
+    text = json.dumps(value, ensure_ascii=False, default=str)
+    if len(text) <= MAX_CLOUD_LOG_VALUE_LENGTH:
+        return value
+    return {
+        'truncated': True,
+        'original_length': len(text),
+        'content': text[:MAX_CLOUD_LOG_VALUE_LENGTH],
+    }
+
+
+def _request_url(request: Request) -> str:
+    scheme = request.headers.get('X-Forwarded-Proto', request.url.scheme).split(',', 1)[0].strip()
+    host = request.headers.get('X-Forwarded-Host', request.headers.get('host', '')).split(',', 1)[0].strip()
+    return f'{scheme}://{host}{request.url.path}' + (f'?{request.url.query}' if request.url.query else '')
+
+
+async def _send_cloud_log_batch(app, events: list[dict]) -> None:
+    await asyncio.wait_for(
+        asyncio.to_thread(app.state.aliyun_logger.logging_batch, events),
+        timeout=settings.API_LOG_FLUSH_TIMEOUT,
+    )
+
+
+async def cloud_log_worker(app) -> None:
+    """异步批量上报访问日志，不阻塞业务请求。"""
+    queue = app.state.cloud_log_queue
+    logger = app.state.logger
+    while True:
+        first_event = await queue.get()
+        if first_event is None:
+            queue.task_done()
+            return
+
+        events = [first_event]
+        while len(events) < MAX_LOG_BATCH_SIZE:
+            try:
+                event = queue.get_nowait()
+            except asyncio.QueueEmpty:
+                break
+            if event is None:
+                queue.task_done()
+                break
+            events.append(event)
+
+        try:
+            for attempt in range(MAX_LOG_RETRIES):
+                try:
+                    await _send_cloud_log_batch(app, events)
+                    break
+                except Exception as exc:
+                    if attempt + 1 >= MAX_LOG_RETRIES:
+                        logger.exception(
+                            f'阿里云API日志批量上报失败: count={len(events)}, '
+                            f'destination={settings.API_ALIYUN_LOG_PROJECT}/'
+                            f'{settings.API_ALIYUN_LOGSTORE}, '
+                            f'error={type(exc).__name__}: {exc}'
+                        )
+                    else:
+                        await asyncio.sleep(0.5 * (2 ** attempt))
+        finally:
+            for _ in events:
+                queue.task_done()
+
+
+async def _enqueue_cloud_log(app, event: dict) -> None:
+    queue = getattr(app.state, 'cloud_log_queue', None)
+    if queue is None:
+        try:
+            await _send_cloud_log_batch(app, [event])
+        except Exception as exc:
+            app.state.logger.exception(
+                f'阿里云API日志上报失败: '
+                f'destination={settings.API_ALIYUN_LOG_PROJECT}/'
+                f'{settings.API_ALIYUN_LOGSTORE}, '
+                f'error={type(exc).__name__}: {exc}'
+            )
+        return
+    try:
+        queue.put_nowait(event)
+    except asyncio.QueueFull:
+        app.state.logger.error(
+            f'阿里云API日志队列已满，丢弃日志: request_id={event.get("trace_id", "")}'
+        )
+
+
+async def report_api_request(request: Request, response: Response) -> None:
+    """统一记录本地访问日志，并异步上报阿里云日志。"""
+    logger = request.app.state.logger
+    duration_ms = round((time.perf_counter() - request.state.started_at) * 1000, 2)
+    request_params = getattr(request.state, 'request_params', {'query': {}, 'body': None})
+    safe_request_params = _limit_log_value(request_params)
+    error_message = _redact_text(
+        getattr(request.state, 'error_traceback', '')
+        or getattr(request.state, 'error_reason', '')
+    )
+    request_url = _request_url(request)
+    local_message = (
+        f'API请求 request_id={request.state.request_id} method={request.method} url={request_url} '
+        f'params={json.dumps(safe_request_params, ensure_ascii=False, default=str)} '
+        f'status={response.status_code} duration_ms={duration_ms}'
+    )
+    if error_message:
+        logger.error(f'{local_message} error={error_message}')
+    else:
+        logger.info(local_message)
+
+    request_body = request_params.get('body') if isinstance(request_params, dict) else None
+    request_body = request_body if isinstance(request_body, dict) else {}
+    query_metrics = getattr(request.state, 'query_metrics', {})
+    cloud_data = {
+        'url': request_url,
+        'path': request.url.path,
+        'method': request.method,
+        'request_id': request.state.request_id,
+        'request_params': _limit_log_value(request_body),
+        'platforms': request_body.get('platforms'),
+        'status_code': response.status_code,
+        'success': response.status_code < 400,
+        'request_duration_ms': duration_ms,
+        'failure_stage': getattr(request.state, 'failure_stage', ''),
+        'error_type': getattr(request.state, 'error_type', ''),
+        'message': error_message,
+        'query_result_count': query_metrics.get('result_count'),
+        'query_duration_ms': query_metrics.get('duration_ms'),
+        'query_success': query_metrics.get('success'),
+    }
+    await _enqueue_cloud_log(request.app, {
+        'code': '2000' if response.status_code < 400 else '9000',
+        'message': 'API请求成功' if response.status_code < 400 else error_message,
+        'data': cloud_data,
+        'trace_id': request.state.request_id,
+    })

api/routes.py

@@ -0,0 +1,8 @@
+from fastapi import APIRouter
+
+from api.chui_zhi.videos import router as chui_zhi_video_router
+
+
+# 所有业务Router只在这里集中注册；应用入口不感知具体业务模块。
+api_router = APIRouter()
+api_router.include_router(chui_zhi_video_router)

config/base.py

@@ -32,18 +32,22 @@ class Settings(BaseSettings):
     DB_PASSWORD: str = Field(..., validation_alias="DB_PASSWORD")
     DB_NAME: str = Field(..., validation_alias="DB_NAME")
     DB_CHARSET: str = Field(..., validation_alias="DB_CHARSET")
-    DB_POOL_SIZE: int = 20
+    DB_POOL_SIZE: int = 12
     DB_POOL_RECYCLE: int = 3600
 
     # 垂直 spider 查询 API
     API_HOST: str = "127.0.0.1"
     API_PORT: int = 8888
-    API_MAX_LIMIT: int = 1000
-    API_MAX_CONCURRENT_REQUESTS: int = 20
+    API_MAX_LIMIT: int = 500
+    API_MAX_CONCURRENT_REQUESTS: int = 8
     API_QUERY_TIMEOUT: int = 30
     API_QUEUE_TIMEOUT: float = 2.0
     API_LOG_QUEUE_SIZE: int = 1000
     API_LOG_FLUSH_TIMEOUT: float = 5.0
+    API_ALIYUN_LOG_PROJECT: str = "crawler-log-prod"
+    API_ALIYUN_LOGSTORE: str = "crawler-api-access"
+    API_ALIYUN_LOG_ENDPOINT: str = "cn-hangzhou.log.aliyuncs.com"
+    # 兼容旧部署环境变量；API已不再读取或校验该值。
     CHUI_ZHI_API_TOKEN: str = ""
 
     # 阿里云RocketMQ配置

core/utils/log/aliyun_log.py

@@ -12,7 +12,7 @@ proxies = {"http": None, "https": None}
 from core.utils.trace_utils import get_current_trace_id  # 导入工具函数
 
 
-API_INDEXED_FIELDS = (
+LEGACY_PROMOTED_FIELDS = (
     "request_id",
     "path",
     "method",
@@ -26,16 +26,107 @@ API_INDEXED_FIELDS = (
     "query_success",
 )
 
+API_LOG_FIELDS = (
+    "request_id",
+    "url",
+    "path",
+    "method",
+    "request_params",
+    "platforms",
+    "status_code",
+    "success",
+    "request_duration_ms",
+    "failure_stage",
+    "error_type",
+    "query_result_count",
+    "query_duration_ms",
+    "query_success",
+)
+
+
+def _stringify_sls_value(value):
+    """SLS字段只能写字符串；复合值保留为紧凑JSON，布尔值统一小写。"""
+    if isinstance(value, bool):
+        return "true" if value else "false"
+    if isinstance(value, (dict, list, tuple)):
+        return json.dumps(value, ensure_ascii=False, separators=(",", ":"), default=str)
+    return str(value)
+
 
 class AliyunLogger(object):
     """
     阿里云日志方法
     """
 
-    def __init__(self, platform, mode, env="prod"):
+    def __init__(
+            self,
+            platform,
+            mode,
+            env="prod",
+            project=None,
+            logstore=None,
+            endpoint=None,
+    ):
         self.platform = platform
         self.mode = mode
         self.env = env
+        # 未指定时保持原有爬虫日志目的地；API可传入独立Logstore。
+        self.project = project
+        self.logstore = logstore
+        self.endpoint = endpoint
+
+    def _resolve_destination(self):
+        if self.env == "dev":
+            default_project = "crawler-log-dev"
+            default_logstore = "crawler-log-dev"
+        else:
+            default_project = "crawler-log-prod"
+            default_logstore = "crawler-fetch"
+        return (
+            self.project or default_project,
+            self.logstore or default_logstore,
+            self.endpoint or "cn-hangzhou.log.aliyuncs.com",
+        )
+
+    def _build_log_contents(self, event):
+        message = str(event.get("message") or "").replace("\r", " ").replace("\n", " ")
+        trace_id = event.get("trace_id") or get_current_trace_id() or ""
+        data = event.get("data") or {}
+
+        if self.mode == "api":
+            # API使用独立Logstore，字段直接平铺，避免data重复嵌套和无意义空字段。
+            contents = [
+                ("TraceId", str(trace_id)),
+                ("platform", str(self.platform)),
+                ("mode", str(self.mode)),
+                ("message", message),
+            ]
+            if isinstance(data, dict):
+                contents.extend(
+                    (field, _stringify_sls_value(data[field]))
+                    for field in API_LOG_FIELDS
+                    if data.get(field) not in (None, "")
+                )
+            return contents
+
+        # 原有爬虫日志格式保持不变。
+        contents = [
+            ("TraceId", str(trace_id)),
+            ("code", str(event.get("code", ""))),
+            ("platform", str(self.platform)),
+            ("mode", str(self.mode)),
+            ("message", message),
+            ("data", json.dumps(data, ensure_ascii=False, default=str) if data else ""),
+            ("account", str(event.get("account"))),
+            ("timestamp", str(int(time.time()))),
+        ]
+        if isinstance(data, dict):
+            contents.extend(
+                (field, str(data[field]))
+                for field in LEGACY_PROMOTED_FIELDS
+                if data.get(field) is not None
+            )
+        return contents
 
     # 写入阿里云日志
     def logging(
@@ -58,40 +149,13 @@ class AliyunLogger(object):
         """
         accessKeyId = settings.ALIYUN_ACCESS_KEY_ID
         accessKey = settings.ALIYUN_ACCESS_KEY_SECRET
-        if self.env == "dev":
-            project = "crawler-log-dev"
-            logstore = "crawler-log-dev"
-            endpoint = "cn-hangzhou.log.aliyuncs.com"
-        else:
-            project = "crawler-log-prod"
-            logstore = "crawler-fetch"
-            endpoint = "cn-hangzhou.log.aliyuncs.com"
+        project, logstore, endpoint = self._resolve_destination()
 
         client = LogClient(endpoint, accessKeyId, accessKey)
         log_group = []
         for event in events:
-            message = str(event.get("message") or "").replace("\r", " ").replace("\n", " ")
-            trace_id = event.get("trace_id") or get_current_trace_id() or ""
-            data = event.get("data") or {}
             log_item = LogItem()
-            contents = [
-                ("TraceId", str(trace_id)),
-                ("code", str(event.get("code", ""))),
-                ("platform", str(self.platform)),
-                ("mode", str(self.mode)),
-                ("message", message),
-                ("data", json.dumps(data, ensure_ascii=False, default=str) if data else ""),
-                ("account", str(event.get("account"))),
-                ("timestamp", str(int(time.time()))),
-            ]
-            # API核心指标额外提升为SLS顶层字段，便于直接统计请求量和成功率。
-            if isinstance(data, dict):
-                contents.extend(
-                    (field, str(data[field]))
-                    for field in API_INDEXED_FIELDS
-                    if data.get(field) is not None
-                )
-            log_item.set_contents(contents)
+            log_item.set_contents(self._build_log_contents(event))
             log_group.append(log_item)
 
         if not log_group:

core/utils/log/logger_manager.py

@@ -32,17 +32,24 @@ class LoggerManager:
     def get_aliyun_logger(
         platform: str = "system",
         mode: str = "crawler",
-        env: str = "prod"
+        env: str = "prod",
+        project: str | None = None,
+        logstore: str | None = None,
+        endpoint: str | None = None,
     ) -> AliyunLogger:
         """
 
         :rtype: AliyunLogger
         """
-        key = f"{platform}_{mode}"
+        # 目的地属于缓存键，避免同一platform/mode的不同业务误用同一个Logstore。
+        key = f"{platform}_{mode}_{env}_{project}_{logstore}_{endpoint}"
         if key not in LoggerManager._aliyun_loggers:
             LoggerManager._aliyun_loggers[key] = AliyunLogger(
                 platform=platform,
                 mode=mode,
-                env=env
+                env=env,
+                project=project,
+                logstore=logstore,
+                endpoint=endpoint,
             )
-        return LoggerManager._aliyun_loggers[key]
+        return LoggerManager._aliyun_loggers[key]

deploy.sh

@@ -127,13 +127,15 @@ main() {
     "$PIP" install -r "$REQUIREMENTS" || handle_error "安装依赖失败"
 
     # 在停止旧服务前验证全部配置，避免配置错误扩大停机时间。
-    log "检查API配置和Token..."
-    "$PYTHON" -c "from config import settings; assert len(settings.CHUI_ZHI_API_TOKEN) >= 32, 'CHUI_ZHI_API_TOKEN至少需要32个字符'; print(f'API监听地址: {settings.API_HOST}:{settings.API_PORT}')" >> "$LOG_FILE" \
+    log "检查API配置..."
+    "$PYTHON" -c "from config import settings; print(f'API监听地址: {settings.API_HOST}:{settings.API_PORT}')" >> "$LOG_FILE" \
         || handle_error "API配置检查失败，请检查${APP_DIR}/.env"
     API_PORT="$("$PYTHON" -c "from config import settings; print(settings.API_PORT)")" \
         || handle_error "无法读取API_PORT"
 
     log "停止现有API和主服务..."
+    # 兼容清理迁移期间直接从fastapi_app启动的进程，正式入口统一为api.app。
+    stop_service "迁移期垂直视频API" "$API_PID_FILE" "${PYTHON} -m api.fastapi_app"
     stop_service "垂直视频API" "$API_PID_FILE" "${PYTHON} -m api.app"
     stop_service "主爬虫服务" "$MAIN_PID_FILE" "${PYTHON} main.py"
 

deploy/nginx/chui_zhi_api.conf.example

@@ -54,7 +54,6 @@ server {
         proxy_set_header X-Real-IP $remote_addr;
         proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
         proxy_set_header X-Forwarded-Proto $scheme;
-        proxy_set_header X-API-Key $http_x_api_key;
         proxy_set_header X-Request-ID $http_x_request_id;
 
         proxy_connect_timeout 5s;

docs/chui_zhi_video_api.md

@@ -8,11 +8,11 @@
 |---|---|
 | 业务接口 | `POST /api/v1/crawler/videos/query` |
 | 请求格式 | `application/json` |
-| 鉴权 | 请求头 `X-API-Key` |
+| 鉴权 | 接口自身不校验 Token；公网访问由 Nginx 控制 |
 | 时间标准 | 东八区（UTC+8） |
 | 默认时间范围 | 最近 3 天 |
 | 默认页大小 | 500 条 |
-| 最大页大小 | 由 `API_MAX_LIMIT` 控制，默认 1000 条 |
+| 最大页大小 | 由 `API_MAX_LIMIT` 控制，默认 500 条 |
 | 分页方式 | 基于去重结果最大 `id` 的游标分页 |
 | 去重方式 | 相同 `out_video_id` 保留最大 `id` 对应的记录 |
 | 数据库操作 | 只读，不修改数据库结构和数据 |
@@ -24,25 +24,14 @@
 | `GET /health` | 进程存活检查，不访问数据库 |
 | `GET /ready` | 就绪检查，执行 `SELECT 1 AS ok` 验证数据库连接 |
 
-探针不要求 `X-API-Key`，但 Nginx 示例默认只允许本机访问，不应暴露为公网业务接口。
+探针由 Nginx 示例限制为仅本机访问，不应暴露为公网业务接口。
 
-## 2. 鉴权与请求追踪
+## 2. 访问控制与请求追踪
 
-### 2.1 API Token
+### 2.1 公网访问控制
 
-业务请求必须携带：
-
-```http
-X-API-Key: <CHUI_ZHI_API_TOKEN>
-```
-
-Token 只能通过环境变量或服务器密钥管理系统注入，不应写入源码：
-
-```bash
-export CHUI_ZHI_API_TOKEN="$(openssl rand -hex 32)"
-```
-
-Token 为空时，API 拒绝启动。Token 错误时返回 HTTP 401。
+业务接口自身不校验 Token。公网部署时应在 Nginx 层限制新加坡调度服务器的固定出口 IP，
+并启用请求速率和连接数限制；不要将后端 `8888` 端口直接暴露到公网。
 
 ### 2.2 Request ID
 
@@ -445,12 +434,12 @@ Cache-Control: no-store
 |---|---:|---|
 | `API_HOST` | `127.0.0.1` | 只监听本机，由 Nginx 对外代理 |
 | `API_PORT` | `8888` | API 本地监听端口 |
-| `DB_POOL_SIZE` | `20` | 单进程数据库连接池上限 |
+| `DB_POOL_SIZE` | `12` | 单进程数据库连接池上限 |
 | `DB_POOL_RECYCLE` | `3600` | 数据库连接回收秒数 |
-| `API_MAX_CONCURRENT_REQUESTS` | `20` | 查询并发配置上限 |
+| `API_MAX_CONCURRENT_REQUESTS` | `8` | 查询并发配置上限 |
 | `API_QUEUE_TIMEOUT` | `2.0` | 等待查询并发槽位的最长秒数 |
 | `API_QUERY_TIMEOUT` | `30` | 单条查询最长秒数 |
-| `API_MAX_LIMIT` | `1000` | 单页最大返回条数 |
+| `API_MAX_LIMIT` | `500` | 单页最大返回条数 |
 | `API_LOG_QUEUE_SIZE` | `1000` | SLS 异步日志队列长度 |
 | `API_LOG_FLUSH_TIMEOUT` | `5.0` | 单批 SLS 上报及关闭刷新超时 |
 
@@ -510,7 +499,22 @@ min(API_MAX_CONCURRENT_REQUESTS, DB_POOL_SIZE)
 - `message`
 - 脱敏后的 `request_params`
 
-其中请求量、状态、耗时和错误相关字段会提升为 SLS 顶层字段，方便直接统计每日请求量、成功率、P95/P99 耗时和错误分布。
+API日志全部使用SLS顶层字段，不再写入重复的 `data` 字符串。字典和数组字段使用紧凑JSON字符串，
+布尔值统一为小写 `true`/`false`；值为空的 `failure_stage`、`error_type` 等字段不会写入。
+`account=None` 和自定义 `timestamp` 也不会出现在API日志中，时间以SLS自带的 `__time__` 为准。
+这样可以直接统计每日请求量、成功率、P95/P99耗时和错误分布。
+
+API访问日志使用独立的SLS目的地，默认配置为：
+
+```dotenv
+API_ALIYUN_LOG_PROJECT=crawler-log-prod
+API_ALIYUN_LOGSTORE=crawler-api-access
+API_ALIYUN_LOG_ENDPOINT=cn-hangzhou.log.aliyuncs.com
+```
+
+部署前需要在阿里云日志服务中创建 `crawler-api-access` Logstore，并为上面的核心指标建立索引。
+原有爬虫日志继续写入 `crawler-fetch`，两者不会混在一起。如果需要使用其他Project或Logstore，
+只修改环境变量并重启API即可。
 
 ## 10. curl 示例
 
@@ -518,7 +522,6 @@ min(API_MAX_CONCURRENT_REQUESTS, DB_POOL_SIZE)
 
 ```bash
 export CHUI_ZHI_API_URL="https://api.example.com/api/v1/crawler/videos/query"
-export CHUI_ZHI_API_TOKEN="replace-with-real-token"
 ```
 
 ### 10.1 默认最近 3 天
@@ -526,7 +529,6 @@ export CHUI_ZHI_API_TOKEN="replace-with-real-token"
 ```bash
 curl --request POST "$CHUI_ZHI_API_URL" \
   --header "Content-Type: application/json" \
-  --header "X-API-Key: $CHUI_ZHI_API_TOKEN" \
   --header "X-Request-ID: manual-test-001" \
   --data '{}'
 ```
@@ -536,7 +538,6 @@ curl --request POST "$CHUI_ZHI_API_URL" \
 ```bash
 curl --request POST "$CHUI_ZHI_API_URL" \
   --header "Content-Type: application/json" \
-  --header "X-API-Key: $CHUI_ZHI_API_TOKEN" \
   --data '{
     "keywords": ["早"],
     "limit": 500
@@ -548,7 +549,6 @@ curl --request POST "$CHUI_ZHI_API_URL" \
 ```bash
 curl --request POST "$CHUI_ZHI_API_URL" \
   --header "Content-Type: application/json" \
-  --header "X-API-Key: $CHUI_ZHI_API_TOKEN" \
   --data '{
     "start_time": 1786896000000,
     "end_time": 1786982400000,
@@ -566,7 +566,6 @@ curl --request POST "$CHUI_ZHI_API_URL" \
 ```bash
 curl --request POST "$CHUI_ZHI_API_URL" \
   --header "Content-Type: application/json" \
-  --header "X-API-Key: $CHUI_ZHI_API_TOKEN" \
   --data '{
     "keywords": ["早"],
     "start_time": 1786723200000,
@@ -596,17 +595,11 @@ curl --fail http://127.0.0.1:8888/ready
 ```dotenv
 API_HOST=127.0.0.1
 API_PORT=8888
-CHUI_ZHI_API_TOKEN=至少32个字符的随机Token
-```
-
-生成 Token：
-
-```bash
-openssl rand -hex 32
+API_ALIYUN_LOG_PROJECT=crawler-log-prod
+API_ALIYUN_LOGSTORE=crawler-api-access
+API_ALIYUN_LOG_ENDPOINT=cn-hangzhou.log.aliyuncs.com
 ```
 
-Token 需要分别配置在 API 服务器和新加坡调度服务中，不能提交到 Git。
-
 执行完整部署：
 
 ```bash
@@ -619,9 +612,9 @@ bash deploy.sh
 1. 拉取 `master` 最新代码。
 2. 创建或更新 `/root/AutoScraperX/venv`。
 3. 安装 `requirements.txt`。
-4. 在停止旧服务前检查全部配置，并要求 Token 长度至少 32。
+4. 在停止旧服务前检查 API 配置。
 5. 根据 PID 文件优雅停止旧 API 和主爬虫；首次升级时兼容清理旧版无 PID 进程。
-6. 启动 `python -m api.app`。
+6. 启动 `python -m api.app`（Uvicorn单Worker）。
 7. 最多等待 15 秒调用 `/ready`，确认进程和数据库都可用。
 8. API 就绪后启动 `main.py`。
 9. 主进程启动检查失败时，同时停止本次启动的 API，避免留下半部署状态。
@@ -685,7 +678,7 @@ deploy/nginx/chui_zhi_api.conf.example
 - 请求体最大 1 MB
 - 开启 JSON gzip
 - 配置请求速率和单 IP 连接限制
-- 透传 `X-API-Key`、`X-Request-ID`、真实 IP 和协议
+- 透传 `X-Request-ID`、真实 IP 和协议
 - `/health`、`/ready` 只允许本机访问
 - 如果新加坡调度服务器有固定出口 IP，建议启用 IP 白名单
 
@@ -716,7 +709,6 @@ systemd 配置了进程异常自动重启、SIGTERM 优雅退出和文件句柄
 
 ```bash
 export CHUI_ZHI_API_URL="https://api.example.com/api/v1/crawler/videos/query"
-export CHUI_ZHI_API_TOKEN="same-token-as-api"
 ```
 
 调度流程：
@@ -760,8 +752,13 @@ export CHUI_ZHI_API_TOKEN="same-token-as-api"
 
 | 功能 | 文件 |
 |---|---|
-| aiohttp 应用、路由、中间件、鉴权、日志和生命周期 | `api/app.py` |
-| 请求模型、筛选、SQL、查询和响应 | `api/chui_zhi/videos.py` |
+| 稳定ASGI启动入口 | `api/app.py` |
+| FastAPI应用、请求上下文、校验错误兜底和生命周期 | `api/fastapi_app.py` |
+| API参数基类、注册、执行、响应、异常及数据库查询保护 | `api/base.py` |
+| 本地访问日志、SLS上报、脱敏及日志队列 | `api/reporting.py` |
+| 业务Router集中注册 | `api/routes.py` |
+| API公共异常 | `api/errors.py` |
+| 垂直视频入参、SQL、查询、分页及Router | `api/chui_zhi/videos.py` |
 | API 和数据库配置 | `config/base.py` |
 | 异步 MySQL 连接池 | `core/base/async_mysql_client.py` |
 | 阿里云 SLS 批量日志 | `core/utils/log/aliyun_log.py` |
@@ -769,3 +766,37 @@ export CHUI_ZHI_API_TOKEN="same-token-as-api"
 | Nginx 示例 | `deploy/nginx/chui_zhi_api.conf.example` |
 | systemd 示例 | `deploy/systemd/autoscraperx-api.service.example` |
 | API 测试 | `test/test_video_query_api.py` |
+
+## 15. 新增接口约定
+
+每个接口模块只管理请求模型和业务实现。请求模型继承 `ApiParams`，接口继承 `BaseApi`；
+子类只需声明必填的 `path` 并实现 `logic`，直接返回业务数据。请求方法默认是 `POST`，路由名称和说明自动生成。
+路由注册、`code/msg/data` 响应、业务异常、请求ID、本地访问日志和阿里云SLS上报均由基类处理。
+数据库列表查询统一使用基类的 `fetch_all()`，并发限制、查询超时、耗时和结果数量也会自动处理。
+参数在进入子类前校验失败、404/405 和健康检查由全局中间件兜底，使用同一套响应和上报逻辑，且不会重复上报。
+
+```python
+from fastapi import APIRouter, Request
+from api.base import ApiParams, BaseApi
+
+router = APIRouter(prefix='/api/v1/example', tags=['示例'])
+
+
+class ExampleParams(ApiParams):
+    value: str
+
+
+class ExampleApi(BaseApi):
+    """执行示例功能。"""
+
+    path = '/run'
+
+    async def logic(self, params: ExampleParams, request: Request):
+        return {'value': params.value}
+
+
+ExampleApi().register(router)
+```
+
+最后只需在 `api/routes.py` 中 `include_router`。子类不要捕获公共异常、不要重复封装
+`code/msg/data`，也不要自行记录或上报访问日志。

requirements.txt

@@ -1,5 +1,8 @@
 aiohappyeyeballs==2.6.1
 aiohttp==3.12.13
+fastapi==0.141.1
+httpx2==2.9.1
+uvicorn==0.52.3
 aiosignal==1.3.2
 aliyun-log-python-sdk==0.9.24
 aliyun-python-sdk-core==2.13.36
@@ -73,4 +76,4 @@ yarl==1.20.1
 zipp==3.23.0
 
 mq_http_sdk~=1.0.3
-APScheduler~=3.11.2
+APScheduler~=3.11.2

scripts/office/wx_getDomainInfo.py

@@ -64,7 +64,7 @@ async def main():
             insert_data = list(map(lambda x: [app_name, x], bizdomain))
             print(insert_data)
             async with FeishuDataAsync() as feishu_data:
-                    await feishu_data.insert_values("TxA2wpGZHiuLl2kMMokcaU9Mnlb", "d3a349", "A2:B", insert_data)
+                    await feishu_data.insert_values("TxA2wpGZHiuLl2kMMokcaU9Mnlb", "20A24E", "A2:B", insert_data)
         else:
             print(f"小程序 {app_name} 无法获取域名信息")
 

test/test_fastapi_video_query_api.py

@@ -0,0 +1,143 @@
+import asyncio
+
+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
+
+
+class FakeLogger:
+    def __init__(self):
+        self.records = []
+
+    def info(self, message):
+        self.records.append(('info', message))
+
+    def error(self, message):
+        self.records.append(('error', message))
+
+    def exception(self, message):
+        self.records.append(('exception', message))
+
+
+class FakeAliyunLogger:
+    def __init__(self):
+        self.events = []
+
+    def logging_batch(self, events):
+        self.events.extend(events)
+
+
+def build_test_app(mysql):
+    app = create_app(manage_resources=False)
+    app.state.mysql = mysql
+    app.state.query_semaphore = asyncio.Semaphore(10)
+    app.state.logger = FakeLogger()
+    app.state.aliyun_logger = FakeAliyunLogger()
+    return app
+
+
+def test_fastapi_query_keeps_existing_contract_and_unified_logging():
+    assert issubclass(VideoQueryApi, BaseApi)
+
+    class FakeMySQL:
+        async def fetch_all(self, sql, params):
+            return [
+                {'id': 1, 'platform': params[0], 'create_time': '2026-08-04 12:00:00'},
+                {'id': 2, 'platform': params[0], 'create_time': '2026-08-04 11:00:00'},
+            ]
+
+    app = build_test_app(FakeMySQL())
+    payload = {
+        'platforms': ['xiaoniangao'],
+        'start_time': 1785772800000,
+        'end_time': 1785859200000,
+        'filters': [{'field': 'like_cnt', 'operator': '>', 'value': 3}],
+        'limit': 1,
+    }
+    with TestClient(app, raise_server_exceptions=False) as client:
+        response = client.post(
+            API_PATH,
+            json=payload,
+            headers={'X-Request-ID': 'scheduler-request-1'},
+        )
+
+    body = response.json()
+    assert response.status_code == 200
+    assert response.headers['X-Request-ID'] == 'scheduler-request-1'
+    assert response.headers['Cache-Control'] == 'no-store'
+    assert body['code'] == 0
+    assert body['data']['count'] == 1
+    assert body['data']['has_more'] is True
+    assert body['data']['next_cursor'] == {'id': 1}
+    event = app.state.aliyun_logger.events[0]
+    assert event['trace_id'] == 'scheduler-request-1'
+    assert event['data']['request_params'] == payload
+    assert event['data']['status_code'] == 200
+    assert event['data']['query_result_count'] == 2
+
+
+def test_fastapi_validation_and_business_errors_use_same_exit_contract():
+    class FakeMySQL:
+        async def fetch_all(self, sql, params):
+            raise AssertionError('参数错误时不应查询数据库')
+
+    app = build_test_app(FakeMySQL())
+    with TestClient(app, raise_server_exceptions=False) as client:
+        validation = client.post(API_PATH, json={'unknown_parameter': 1})
+        business = client.post(
+            API_PATH,
+            json={'filters': [{'field': 'like_cnt', 'operator': '>', 'value': 'invalid'}]},
+        )
+        wrong_method = client.get(API_PATH)
+
+    assert validation.status_code == 400
+    assert '不支持的参数: unknown_parameter' in validation.json()['msg']
+    assert business.status_code == 422
+    assert '必须是数字' in business.json()['msg']
+    assert wrong_method.status_code == 405
+    assert wrong_method.json() == {'code': 405, 'msg': 'Method Not Allowed', 'data': None}
+    assert len(app.state.aliyun_logger.events) == 3
+    business_event = app.state.aliyun_logger.events[1]
+    assert business_event['data']['failure_stage'] == 'business_validation'
+    assert business_event['data']['error_type'] == 'BusinessValidationError'
+    assert 'Traceback (most recent call last)' in business_event['data']['message']
+    assert '必须是数字' in business_event['data']['message']
+    assert all('X-Request-ID' in response.headers for response in (validation, business, wrong_method))
+
+
+def test_fastapi_health_readiness_and_documentation_routes():
+    class FakeMySQL:
+        async def fetch_one(self, sql):
+            assert sql == 'SELECT 1 AS ok'
+            return {'ok': 1}
+
+    app = build_test_app(FakeMySQL())
+    with TestClient(app, raise_server_exceptions=False) as client:
+        health = client.get('/health')
+        ready = client.get('/ready')
+        docs = client.get('/openapi.json')
+
+    assert health.json()['data']['status'] == 'ok'
+    assert ready.json()['data']['status'] == 'ready'
+    assert API_PATH in docs.json()['paths']
+
+
+def test_fastapi_sls_failure_does_not_change_business_response():
+    class FakeMySQL:
+        async def fetch_all(self, sql, params):
+            return []
+
+    class BrokenAliyunLogger:
+        def logging_batch(self, events):
+            raise RuntimeError('SLS unavailable')
+
+    app = build_test_app(FakeMySQL())
+    app.state.aliyun_logger = BrokenAliyunLogger()
+    with TestClient(app, raise_server_exceptions=False) as client:
+        response = client.post(API_PATH, json={})
+
+    assert response.status_code == 200
+    assert response.json()['code'] == 0
+    assert any('阿里云API日志上报失败' in message for _, message in app.state.logger.records)

test/test_video_query_api.py

@@ -1,24 +1,112 @@
-import asyncio
 from datetime import datetime, timedelta, timezone
 
 import pytest
-from aiohttp.test_utils import TestClient, TestServer
 
-from api.app import ALIYUN_LOGGER_KEY, API_PATH, LOGGER_KEY, create_app
 from api.chui_zhi.videos import (
     BusinessValidationError,
-    MYSQL_KEY,
-    QUERY_SEMAPHORE_KEY,
     FilterCondition,
-    VideoQueryRequest,
+    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 = VideoQueryRequest(
+    request = VideoQueryParams(
         platforms=['xiaoniangao', 'xiaoniangaotuijianliu'],
         start_time=datetime(2026, 8, 4),
         end_time=datetime(2026, 8, 5),
@@ -47,7 +135,7 @@ def test_build_query_contains_parameterized_filters():
 
 
 def test_deduplication_groups_ids_before_cursor_pagination():
-    request = VideoQueryRequest(
+    request = VideoQueryParams(
         platforms=['xiaoniangao'],
         start_time=datetime(2026, 8, 4),
         end_time=datetime(2026, 8, 5),
@@ -66,7 +154,7 @@ def test_deduplication_groups_ids_before_cursor_pagination():
 
 def test_query_limit_above_server_max_is_rejected():
     with pytest.raises(ValueError):
-        VideoQueryRequest(
+        VideoQueryParams(
             start_time=datetime(2026, 8, 4),
             end_time=datetime(2026, 8, 5),
             limit=settings.API_MAX_LIMIT + 1,
@@ -74,7 +162,7 @@ def test_query_limit_above_server_max_is_rejected():
 
 
 def test_id_cursor_builds_stable_group_pagination_and_fetches_one_extra_row():
-    request = VideoQueryRequest(
+    request = VideoQueryParams(
         start_time=datetime(2026, 8, 4),
         end_time=datetime(2026, 8, 5),
         limit=100,
@@ -92,7 +180,7 @@ 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 = VideoQueryRequest(
+    request = VideoQueryParams(
         start_time=int(start_time.timestamp() * 1000),
         end_time=int(end_time.timestamp() * 1000),
     )
@@ -103,7 +191,7 @@ def test_millisecond_timestamps_are_converted_to_china_time():
 
 def test_missing_time_defaults_to_latest_three_days():
     before = datetime.now(timezone(timedelta(hours=8))).replace(tzinfo=None)
-    request = VideoQueryRequest()
+    request = VideoQueryParams()
     after = datetime.now(timezone(timedelta(hours=8))).replace(tzinfo=None)
 
     assert before <= request.end_time <= after
@@ -123,7 +211,7 @@ def test_rejects_unsupported_filter_field():
 
 def test_rejects_unsupported_platform_and_unknown_filter_parameter():
     with pytest.raises(ValueError, match='不支持的平台'):
-        VideoQueryRequest(platforms=['douyin'])
+        VideoQueryParams(platforms=['douyin'])
     with pytest.raises(ValueError):
         FilterCondition(field='like_cnt', operator='>', value=1, unknown='value')
 
@@ -148,7 +236,7 @@ def test_supported_numeric_filter_mapping(field):
 
 
 def test_empty_filters_are_not_added_to_query():
-    request = VideoQueryRequest(filters=[])
+    request = VideoQueryParams(filters=[])
     sql, _ = build_query(request)
     where_sql = sql.split('WHERE', 1)[1]
     assert '`like_cnt` >' not in where_sql
@@ -193,286 +281,3 @@ 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)
-
-
-@pytest.mark.asyncio
-async def test_api_requires_token_and_returns_expected_contract(monkeypatch):
-    local_logs = []
-    cloud_logs = []
-
-    class FakeMySQL:
-        async def fetch_all(self, sql, params):
-            return [
-                {'id': 1, 'platform': params[0], 'create_time': '2026-08-04 12:00:00'},
-                {'id': 2, 'platform': params[0], 'create_time': '2026-08-04 11:00:00'},
-            ]
-
-    class FakeLogger:
-        def info(self, message):
-            local_logs.append(('info', message))
-
-        def error(self, message):
-            local_logs.append(('error', message))
-
-        def exception(self, message):
-            raise AssertionError(message)
-
-    class FakeAliyunLogger:
-        def logging_batch(self, events):
-            cloud_logs.extend(events)
-
-    monkeypatch.setattr(settings, 'CHUI_ZHI_API_TOKEN', 'test-token')
-    app = create_app()
-    app.cleanup_ctx.clear()
-    app[MYSQL_KEY] = FakeMySQL()
-    app[QUERY_SEMAPHORE_KEY] = asyncio.Semaphore(10)
-    app[LOGGER_KEY] = FakeLogger()
-    app[ALIYUN_LOGGER_KEY] = FakeAliyunLogger()
-    client = TestClient(TestServer(app))
-    await client.start_server()
-    payload = {
-        'platforms': ['xiaoniangao'],
-        'start_time': 1785772800000,
-        'end_time': 1785859200000,
-        'filters': [
-            {'field': 'like_cnt', 'operator': '>', 'value': 3},
-        ],
-        'limit': 1,
-    }
-    try:
-        unauthorized = await client.post(API_PATH, json=payload)
-        assert unauthorized.status == 401
-
-        authorized = await client.post(
-            API_PATH,
-            json=payload,
-            headers={'X-API-Key': 'test-token', 'X-Request-ID': 'scheduler-request-1'},
-        )
-        body = await authorized.json()
-        assert authorized.status == 200
-        assert body['code'] == 0
-        assert body['data']['count'] == 1
-        assert body['data']['has_more'] is True
-        assert body['data']['next_cursor'] == {'id': 1}
-        assert body['data']['start_time'] == payload['start_time']
-        assert body['data']['end_time'] == payload['end_time']
-        assert authorized.headers['X-Request-ID'] == 'scheduler-request-1'
-
-        invalid_payload = {
-            **payload,
-            'filters': [{'field': 'like_cnt', 'operator': '>', 'value': 'invalid'}],
-        }
-        failed = await client.post(
-            API_PATH,
-            json=invalid_payload,
-            headers={'X-API-Key': 'test-token'},
-        )
-        assert failed.status == 422
-
-        assert len(cloud_logs) == 3
-        assert all('event_type' not in log['data'] for log in cloud_logs)
-        assert all('request_count' not in log['data'] for log in cloud_logs)
-        assert all(log['data']['path'] == API_PATH for log in cloud_logs)
-        assert all(log['data']['request_duration_ms'] >= 0 for log in cloud_logs)
-        assert cloud_logs[0]['data']['status_code'] == 401
-        assert cloud_logs[0]['data']['request_params']['body'] is None
-        assert cloud_logs[0]['data']['message'] == 'unauthorized'
-        assert cloud_logs[0]['data']['failure_stage'] == 'authentication'
-        assert cloud_logs[0]['data']['error_type'] == 'AuthenticationError'
-        assert cloud_logs[1]['data']['status_code'] == 200
-        assert cloud_logs[1]['trace_id'] == 'scheduler-request-1'
-        assert cloud_logs[1]['data']['request_id'] == 'scheduler-request-1'
-        assert cloud_logs[1]['data']['request_params']['body'] == payload
-        assert 'response' not in cloud_logs[1]['data']
-        assert cloud_logs[1]['data']['success'] is True
-        assert cloud_logs[1]['data']['query_result_count'] == 2
-        assert cloud_logs[1]['data']['query_duration_ms'] >= 0
-        assert cloud_logs[1]['data']['query_success'] is True
-        assert cloud_logs[2]['data']['status_code'] == 422
-        assert cloud_logs[2]['data']['success'] is False
-        assert cloud_logs[2]['data']['query_result_count'] is None
-        assert cloud_logs[2]['data']['query_duration_ms'] is None
-        assert cloud_logs[2]['data']['failure_stage'] == 'business_validation'
-        assert cloud_logs[2]['data']['error_type'] == 'BusinessValidationError'
-        assert 'Traceback (most recent call last)' in cloud_logs[2]['data']['message']
-        assert '必须是数字' in cloud_logs[2]['data']['message']
-        assert cloud_logs[2]['message'] == cloud_logs[2]['data']['message']
-        assert any(level == 'error' and 'status=401' in message for level, message in local_logs)
-        assert any(level == 'info' and 'status=200' in message for level, message in local_logs)
-        assert all('duration_ms=' in message for _, message in local_logs)
-        assert any(level == 'error' and '必须是数字' in message for level, message in local_logs)
-        assert all('response=' not in message for _, message in local_logs)
-    finally:
-        await client.close()
-
-
-@pytest.mark.asyncio
-async def test_cloud_log_failure_does_not_change_api_response(monkeypatch):
-    exception_logs = []
-
-    class FakeMySQL:
-        async def fetch_all(self, sql, params):
-            return []
-
-    class FakeLogger:
-        def info(self, message):
-            pass
-
-        def error(self, message):
-            pass
-
-        def exception(self, message):
-            exception_logs.append(message)
-
-    class BrokenAliyunLogger:
-        def logging_batch(self, events):
-            raise RuntimeError('SLS unavailable')
-
-    monkeypatch.setattr(settings, 'CHUI_ZHI_API_TOKEN', 'test-token')
-    app = create_app()
-    app.cleanup_ctx.clear()
-    app[MYSQL_KEY] = FakeMySQL()
-    app[QUERY_SEMAPHORE_KEY] = asyncio.Semaphore(10)
-    app[LOGGER_KEY] = FakeLogger()
-    app[ALIYUN_LOGGER_KEY] = BrokenAliyunLogger()
-    client = TestClient(TestServer(app))
-    await client.start_server()
-    try:
-        response = await client.post(
-            API_PATH,
-            json={},
-            headers={'X-API-Key': 'test-token'},
-        )
-        assert response.status == 200
-        assert (await response.json())['code'] == 0
-        assert any('阿里云API日志上报失败' in message for message in exception_logs)
-    finally:
-        await client.close()
-
-
-@pytest.mark.asyncio
-async def test_unsupported_parameters_return_specific_error_message(monkeypatch):
-    class FakeMySQL:
-        async def fetch_all(self, sql, params):
-            raise AssertionError('参数校验失败时不应查询数据库')
-
-    class FakeLogger:
-        def info(self, message):
-            pass
-
-        def error(self, message):
-            pass
-
-        def exception(self, message):
-            pass
-
-    class FakeAliyunLogger:
-        def logging_batch(self, events):
-            pass
-
-    monkeypatch.setattr(settings, 'CHUI_ZHI_API_TOKEN', 'test-token')
-    app = create_app()
-    app.cleanup_ctx.clear()
-    app[MYSQL_KEY] = FakeMySQL()
-    app[QUERY_SEMAPHORE_KEY] = asyncio.Semaphore(10)
-    app[LOGGER_KEY] = FakeLogger()
-    app[ALIYUN_LOGGER_KEY] = FakeAliyunLogger()
-    client = TestClient(TestServer(app))
-    await client.start_server()
-    try:
-        unknown_parameter = await client.post(
-            API_PATH,
-            json={'unknown_parameter': 1},
-            headers={'X-API-Key': 'test-token'},
-        )
-        unknown_body = await unknown_parameter.json()
-        assert unknown_parameter.status == 400
-        assert '不支持的参数: unknown_parameter' in unknown_body['msg']
-
-        unsupported_filter = await client.post(
-            API_PATH,
-            json={'filters': [{'field': 'unknown_field', 'operator': '>', 'value': 1}]},
-            headers={'X-API-Key': 'test-token'},
-        )
-        filter_body = await unsupported_filter.json()
-        assert unsupported_filter.status == 400
-        assert 'filters.0.field不支持值 unknown_field' in filter_body['msg']
-
-        unknown_filter_key = await client.post(
-            API_PATH,
-            json={
-                'filters': [
-                    {
-                        'field': 'like_cnt',
-                        'operator': '>',
-                        'value': 1,
-                        'column': 'like_cnt',
-                    }
-                ]
-            },
-            headers={'X-API-Key': 'test-token'},
-        )
-        unknown_key_body = await unknown_filter_key.json()
-        assert unknown_filter_key.status == 400
-        assert '不支持的参数: filters.0.column' in unknown_key_body['msg']
-
-        missing_filter_key = await client.post(
-            API_PATH,
-            json={'filters': [{'field': 'like_cnt', 'value': 1}]},
-            headers={'X-API-Key': 'test-token'},
-        )
-        missing_key_body = await missing_filter_key.json()
-        assert missing_filter_key.status == 400
-        assert 'filters.0.operator不能为空' in missing_key_body['msg']
-    finally:
-        await client.close()
-
-
-def test_api_routes_are_registered():
-    app = create_app()
-    route_names = {route.name for route in app.router.routes()}
-    assert 'chui_zhi_videos' in route_names
-    assert {'health', 'ready'} <= route_names
-
-
-@pytest.mark.asyncio
-async def test_health_and_readiness_do_not_require_business_token():
-    cloud_logs = []
-
-    class FakeMySQL:
-        async def fetch_one(self, sql):
-            assert sql == 'SELECT 1 AS ok'
-            return {'ok': 1}
-
-    class FakeLogger:
-        def info(self, message):
-            pass
-
-        def error(self, message):
-            pass
-
-        def exception(self, message):
-            raise AssertionError(message)
-
-    class FakeAliyunLogger:
-        def logging_batch(self, events):
-            cloud_logs.extend(events)
-
-    app = create_app()
-    app.cleanup_ctx.clear()
-    app[MYSQL_KEY] = FakeMySQL()
-    app[LOGGER_KEY] = FakeLogger()
-    app[ALIYUN_LOGGER_KEY] = FakeAliyunLogger()
-    client = TestClient(TestServer(app))
-    await client.start_server()
-    try:
-        health_response = await client.get('/health')
-        ready_response = await client.get('/ready')
-
-        assert health_response.status == 200
-        assert ready_response.status == 200
-        assert (await health_response.json())['data']['status'] == 'ok'
-        assert (await ready_response.json())['data']['status'] == 'ready'
-        assert len(cloud_logs) == 2
-    finally:
-        await client.close()