import asyncio import hmac import json import re import time import traceback import uuid from typing import Any 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, ) if __name__ == '__main__': main()