AutoScraperX/api/fastapi_app.py

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
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
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:
    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} -> {route_code_location(route)}',
                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()