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()