video-gen/video-gen-api/app/services/api_v3/logging_service.py

"""外部 API v3 日志服务。

按天分类存储在 log/api/ 目录下：
- log/api/requests/YYYY-MM-DD.log — 所有外部请求和响应
- log/api/models/YYYY-MM-DD.log — 模型调用（Volcano Ark SDK）
- log/api/upscale/YYYY-MM-DD.log — 超分轮询
- log/api/errors/YYYY-MM-DD.log — 错误日志
"""

import json
import logging
import os
from datetime import datetime, timezone

# === 日志目录 ===
# video-gen-api/log/api/
_THIS_DIR = os.path.dirname(os.path.abspath(__file__))
# 上溯3级: services/api_v3 -> services -> app -> video-gen-api (即项目根目录)
_BASE_DIR = os.path.dirname(os.path.dirname(os.path.dirname(_THIS_DIR)))
BASE_LOG_DIR = os.path.join(_BASE_DIR, "log", "api")
os.makedirs(BASE_LOG_DIR, exist_ok=True)

# 子目录
REQUESTS_LOG_DIR = os.path.join(BASE_LOG_DIR, "requests")
MODELS_LOG_DIR = os.path.join(BASE_LOG_DIR, "models")
UPSCALE_LOG_DIR = os.path.join(BASE_LOG_DIR, "upscale")
ERRORS_LOG_DIR = os.path.join(BASE_LOG_DIR, "errors")

for d in [REQUESTS_LOG_DIR, MODELS_LOG_DIR, UPSCALE_LOG_DIR, ERRORS_LOG_DIR]:
    os.makedirs(d, exist_ok=True)


def _get_date_str() -> str:
    """获取当前日期字符串。"""
    return datetime.now(timezone.utc).strftime("%Y-%m-%d")


class _DailyFileHandler(logging.Handler):
    """按天写入的日志处理器。"""

    def __init__(self, log_dir: str):
        super().__init__()
        self.log_dir = log_dir
        self._current_date = None
        self._file_handler = None
        self._open_file()

    def _open_file(self):
        """打开当天的日志文件。"""
        date_str = _get_date_str()
        if date_str == self._current_date and self._file_handler:
            return

        if self._file_handler:
            self._file_handler.close()

        self._current_date = date_str
        filepath = os.path.join(self.log_dir, f"{date_str}.log")
        self._file_handler = logging.FileHandler(filepath, encoding="utf-8")
        self._file_handler.setFormatter(
            logging.Formatter("%(asctime)s | %(levelname)s | %(message)s")
        )

    def emit(self, record):
        try:
            self._open_file()
            self._file_handler.emit(record)
        except Exception:
            self.handleError(record)

    def close(self):
        if self._file_handler:
            self._file_handler.close()
        super().close()


def _create_logger(name: str, log_dir: str) -> logging.Logger:
    """创建按天写入的 Logger。"""
    logger = logging.getLogger(name)
    logger.setLevel(logging.DEBUG)

    # 避免重复添加 handler
    if not logger.handlers:
        handler = _DailyFileHandler(log_dir)
        logger.addHandler(handler)

    return logger


# === Logger 实例 ===
requests_logger = _create_logger("api_v3.requests", REQUESTS_LOG_DIR)
models_logger = _create_logger("api_v3.models", MODELS_LOG_DIR)
upscale_logger = _create_logger("api_v3.upscale", UPSCALE_LOG_DIR)
errors_logger = _create_logger("api_v3.errors", ERRORS_LOG_DIR)


def _safe_json(obj) -> str:
    """安全地序列化为 JSON。"""
    try:
        return json.dumps(obj, ensure_ascii=False, default=str)
    except Exception:
        return str(obj)


# === 请求/响应日志 ===

def log_request(method: str, path: str, api_key_id: str, body: dict | None = None):
    """记录外部请求。"""
    requests_logger.info(
        f"REQUEST | {method} {path} | key={api_key_id} | body={_safe_json(body)}"
    )


def log_response(method: str, path: str, api_key_id: str, status_code: int, body=None, duration_ms: int = 0):
    """记录外部响应。"""
    requests_logger.info(
        f"RESPONSE | {method} {path} | key={api_key_id} | status={status_code} | duration={duration_ms}ms | body={_safe_json(body)}"
    )


def log_request_error(method: str, path: str, api_key_id: str, error: str, status_code: int = 500):
    """记录请求错误。"""
    errors_logger.error(
        f"REQUEST_ERROR | {method} {path} | key={api_key_id} | status={status_code} | error={error}"
    )


# === 模型调用日志 ===

def log_model_request(engine_id: str, model_name: str, task_id: str, params: dict):
    """记录模型调用请求。"""
    models_logger.info(
        f"MODEL_REQUEST | engine={engine_id} | model={model_name} | task={task_id} | params={_safe_json(params)}"
    )


def log_model_response(engine_id: str, model_name: str, task_id: str, success: bool, result: dict | None = None, error: str | None = None):
    """记录模型调用响应。"""
    if success:
        models_logger.info(
            f"MODEL_RESPONSE | engine={engine_id} | model={model_name} | task={task_id} | success | result={_safe_json(result)}"
        )
    else:
        models_logger.error(
            f"MODEL_RESPONSE | engine={engine_id} | model={model_name} | task={task_id} | failed | error={error}"
        )
        errors_logger.error(
            f"MODEL_ERROR | engine={engine_id} | model={model_name} | task={task_id} | error={error}"
        )


# === 超分轮询日志 ===

def log_upscale_poll_start(task_id: str, api_task_id: str):
    """记录超分轮询开始。"""
    upscale_logger.info(f"UPSCALE_POLL_START | task={task_id} | api_task={api_task_id}")


def log_upscale_poll(task_id: str, api_task_id: str, status: str, attempt: int, result: dict | None = None):
    """记录超分轮询状态。"""
    upscale_logger.info(
        f"UPSCALE_POLL | task={task_id} | api_task={api_task_id} | status={status} | attempt={attempt} | result={_safe_json(result)}"
    )


def log_upscale_poll_end(task_id: str, api_task_id: str, success: bool, final_status: str, total_attempts: int):
    """记录超分轮询结束。"""
    if success:
        upscale_logger.info(
            f"UPSCALE_POLL_END | task={task_id} | api_task={api_task_id} | success | status={final_status} | attempts={total_attempts}"
        )
    else:
        upscale_logger.error(
            f"UPSCALE_POLL_END | task={task_id} | api_task={api_task_id} | failed | status={final_status} | attempts={total_attempts}"
        )
        errors_logger.error(
            f"UPSCALE_ERROR | task={task_id} | api_task={api_task_id} | status={final_status} | attempts={total_attempts}"
        )


# === 通用错误日志 ===

def log_error(category: str, message: str, details: dict | None = None):
    """记录通用错误。"""
    errors_logger.error(
        f"{category} | {message} | details={_safe_json(details)}"
    )