AutoScraperX/api/app.py

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