video-gen/video-gen-api/app/services/operation_log_service.py

from __future__ import annotations

import atexit
import json
import logging
import os
import queue
import re
import threading
import traceback
from datetime import datetime
from typing import Any
from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit

from app.services.log_config import LOG_DATE_FORMAT, LOG_DIR, is_enabled

logger = logging.getLogger("video_gen")

MAX_LOG_FIELD_LENGTH = 20000
MAX_TRACEBACK_LENGTH = 12000
LOG_BASE_DIR = os.path.dirname(LOG_DIR)
OPERATION_LOG_ROOT = os.path.join(LOG_BASE_DIR, "OperationLogs")
MODULE_GENERATION_LOG_ROOT = os.path.join(LOG_BASE_DIR, "ModuleGeneration")
AI_MODEL_LOG_ROOT = LOG_DIR
SENSITIVE_KEY_PATTERNS = (
    "secret",
    "authorization",
    "cookie",
    "credential",
    "signature",
    "accesskey",
    "access_key",
    "api_key",
    "apikey",
    "security_token",
)
SENSITIVE_TOKEN_KEYS = {
    "token",
    "access_token",
    "refresh_token",
    "bearer_token",
    "security_token",
    "x_tos_security_token",
}
FILE_BASE64_KEYS = {
    "b64_json",
    "file_data",
    "file_base64",
    "content_base64",
    "image_base64",
    "video_base64",
    "audio_base64",
}
FILE_DATA_URI_MIME_PREFIXES = ("image/", "video/", "audio/")
FILE_DATA_URI_MIME_TYPES = {"application/pdf", "application/octet-stream"}
FILE_BASE64_PREVIEW_CHARS = 30
LOG_WRITE_QUEUE_SIZE = 10000

_LOG_WRITE_QUEUE: queue.Queue[tuple[str, str] | None] = queue.Queue(maxsize=LOG_WRITE_QUEUE_SIZE)
_LOG_WRITER_THREAD: threading.Thread | None = None
_LOG_WRITER_PID = os.getpid()
_LOG_WRITER_START_LOCK = threading.Lock()
_LOG_DIRECTORY_LOCK = threading.Lock()
_CREATED_LOG_DIRECTORIES: set[str] = set()


def _reset_log_writer_after_fork() -> None:
    """Celery prefork 子进程不得复用父进程的线程、Queue 或锁。"""
    global _LOG_WRITE_QUEUE, _LOG_WRITER_THREAD, _LOG_WRITER_PID
    global _LOG_WRITER_START_LOCK, _LOG_DIRECTORY_LOCK, _CREATED_LOG_DIRECTORIES
    _LOG_WRITE_QUEUE = queue.Queue(maxsize=LOG_WRITE_QUEUE_SIZE)
    _LOG_WRITER_THREAD = None
    _LOG_WRITER_PID = os.getpid()
    _LOG_WRITER_START_LOCK = threading.Lock()
    _LOG_DIRECTORY_LOCK = threading.Lock()
    _CREATED_LOG_DIRECTORIES = set()


if hasattr(os, "register_at_fork"):
    os.register_at_fork(after_in_child=_reset_log_writer_after_fork)


def _safe_name(value: str | None, default: str = "unknown") -> str:
    text = str(value or default).strip() or default
    text = re.sub(r"[^a-zA-Z0-9_.-]+", "_", text)
    return text[:120] or default


def _mask_string(value: str) -> str:
    if len(value) <= 8:
        return "***"
    return f"{value[:4]}***{value[-4:]}"


def _is_sensitive_key(key: str) -> bool:
    lower = str(key).replace("-", "_").lower()
    if lower == "sign" or lower in SENSITIVE_TOKEN_KEYS:
        return True
    return any(pattern in lower for pattern in SENSITIVE_KEY_PATTERNS)


def _decoded_base64_size(value: str) -> int:
    compact = "".join(value.split())
    if not compact:
        return 0
    padding = 2 if compact.endswith("==") else (1 if compact.endswith("=") else 0)
    return max(0, (len(compact) * 3) // 4 - padding)


def _file_base64_preview(value: str, key_path: tuple[str, ...]) -> str | None:
    data_uri = re.match(r"^data:([^;,]+);base64,(.*)$", value, flags=re.IGNORECASE | re.DOTALL)
    prefix = ""
    payload = value
    is_file = False
    if data_uri:
        mime = str(data_uri.group(1) or "").lower()
        is_file = mime.startswith(FILE_DATA_URI_MIME_PREFIXES) or mime in FILE_DATA_URI_MIME_TYPES
        prefix = value[: value.find(",") + 1]
        payload = data_uri.group(2)
    elif key_path and key_path[-1].replace("-", "_").lower() in FILE_BASE64_KEYS:
        # Raw base64 is treated as file content only for an explicit file field.
        is_file = len(value) >= 64 and bool(re.fullmatch(r"[A-Za-z0-9+/=\s]+", value))
    if not is_file:
        return None

    compact = "".join(payload.split())
    preview = compact[:FILE_BASE64_PREVIEW_CHARS]
    total_bytes = _decoded_base64_size(compact)
    preview_bytes = min(total_bytes, (len(preview) * 3) // 4)
    remaining_bytes = max(0, total_bytes - preview_bytes)
    return f"{prefix}{preview}...<remaining_file_bytes:{remaining_bytes}>"


def _sanitize_url(value: str) -> str:
    try:
        parts = urlsplit(value)
        if not parts.scheme or not parts.netloc:
            return value
        query = []
        for k, v in parse_qsl(parts.query, keep_blank_values=True):
            query.append((k, _mask_string(v) if _is_sensitive_key(k) else v))
        return urlunsplit((parts.scheme, parts.netloc, parts.path, urlencode(query), parts.fragment))
    except Exception:
        return value


def sanitize_log_value(value: Any, *, key_path: tuple[str, ...] = ()) -> Any:
    if value is None:
        return None
    if isinstance(value, str):
        file_preview = _file_base64_preview(value, key_path)
        if file_preview is not None:
            return file_preview
        text = _sanitize_url(value) if value.startswith(("http://", "https://")) else value
        if len(text) > MAX_LOG_FIELD_LENGTH:
            return text[:MAX_LOG_FIELD_LENGTH] + f"...<truncated:{len(text) - MAX_LOG_FIELD_LENGTH}>"
        return text
    if isinstance(value, dict):
        output: dict[str, Any] = {}
        for k, v in value.items():
            key = str(k)
            output[key] = (
                "***"
                if _is_sensitive_key(key)
                else sanitize_log_value(v, key_path=(*key_path, key))
            )
        return output
    if isinstance(value, (list, tuple)):
        return [sanitize_log_value(v, key_path=(*key_path, str(index))) for index, v in enumerate(value)]
    return value


def build_exception_detail(exc: BaseException | None, extra: dict[str, Any] | None = None) -> dict[str, Any]:
    detail: dict[str, Any] = dict(extra or {})
    if exc is not None:
        tb = "".join(traceback.format_exception(type(exc), exc, exc.__traceback__))
        if len(tb) > MAX_TRACEBACK_LENGTH:
            tb = tb[:MAX_TRACEBACK_LENGTH] + f"...<traceback_truncated:{len(tb) - MAX_TRACEBACK_LENGTH}>"
        detail.update(
            {
                "exception_type": type(exc).__name__,
                "exception_message": str(exc),
                "traceback": tb,
            }
        )
    return detail


def _ensure_log_directory(path: str) -> None:
    if path in _CREATED_LOG_DIRECTORIES:
        return
    with _LOG_DIRECTORY_LOCK:
        if path not in _CREATED_LOG_DIRECTORIES:
            os.makedirs(path, exist_ok=True)
            _CREATED_LOG_DIRECTORIES.add(path)


def _write_log_line(path: str, line: str) -> None:
    target_dir = os.path.dirname(path)
    _ensure_log_directory(target_dir)
    with open(path, "a", encoding="utf-8") as file_obj:
        file_obj.write(line)


def _log_writer_loop() -> None:
    while True:
        item = _LOG_WRITE_QUEUE.get()
        try:
            if item is None:
                return
            path, line = item
            _write_log_line(path, line)
        except Exception as exc:
            logger.warning("operation log async write failed: error=%s", exc, exc_info=True)
        finally:
            _LOG_WRITE_QUEUE.task_done()


def _ensure_log_writer() -> None:
    global _LOG_WRITER_THREAD
    if _LOG_WRITER_PID != os.getpid():
        # 非 POSIX/spawn 或 register_at_fork 不可用时的兜底。
        _reset_log_writer_after_fork()
    thread = _LOG_WRITER_THREAD
    if thread is not None and thread.is_alive():
        return
    with _LOG_WRITER_START_LOCK:
        thread = _LOG_WRITER_THREAD
        if thread is None or not thread.is_alive():
            thread = threading.Thread(
                target=_log_writer_loop,
                name="operation-log-writer",
                daemon=True,
            )
            thread.start()
            _LOG_WRITER_THREAD = thread


def _flush_pending_logs() -> None:
    """进程正常退出时尽力同步落盘尚未消费的日志。"""
    while True:
        try:
            item = _LOG_WRITE_QUEUE.get_nowait()
        except queue.Empty:
            return
        try:
            if item is not None:
                _write_log_line(*item)
        except Exception as exc:
            logger.warning("operation log shutdown flush failed: error=%s", exc, exc_info=True)
        finally:
            _LOG_WRITE_QUEUE.task_done()


atexit.register(_flush_pending_logs)


def _append_json_log(root_dir: str, domain: str | None, entry: dict[str, Any]) -> None:
    if not is_enabled():
        return
    target_dir = root_dir if domain is None else os.path.join(root_dir, _safe_name(domain, "default"))
    today = datetime.now().strftime(LOG_DATE_FORMAT)
    path = os.path.join(target_dir, f"{today}.log")
    try:
        line = json.dumps(sanitize_log_value(entry), ensure_ascii=False, default=str) + "\n"
        _ensure_log_writer()
        _LOG_WRITE_QUEUE.put_nowait((path, line))
    except queue.Full:
        # 队列满时同步降级，账务/异常日志不能静默丢失。
        try:
            _write_log_line(path, line)
        except Exception as exc:
            logger.warning("operation log fallback write failed: path=%s error=%s", path, exc, exc_info=True)
    except Exception as exc:
        logger.warning("operation log enqueue failed: root_dir=%s domain=%s error=%s", root_dir, domain, exc, exc_info=True)


def _append_operation_log(domain: str, entry: dict[str, Any]) -> None:
    _append_json_log(OPERATION_LOG_ROOT, domain, entry)


def _base_entry(
    *,
    log_type: str,
    domain: str,
    event_type: str,
    module: str | None = None,
    event_status: str = "success",
    source: str | None = None,
    trace_id: str | None = None,
    request_id: str | None = None,
    user_id: str | None = None,
    project_id: str | None = None,
    session_id: str | None = None,
    group_id: str | None = None,
    asset_id: str | None = None,
    task_id: str | None = None,
    step_id: str | None = None,
    remote_action: str | None = None,
    remote_request_id: str | None = None,
    api_key_id: str | None = None,
    message: str | None = None,
    detail: dict[str, Any] | None = None,
    error: str | None = None,
) -> dict[str, Any]:
    return {
        "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
        "log_type": log_type,
        "domain": domain,
        "module": module or domain,
        "event_type": event_type,
        "event_status": event_status,
        "source": source,
        "trace_id": trace_id,
        "request_id": request_id,
        "user_id": user_id,
        "project_id": project_id,
        "session_id": session_id,
        "group_id": group_id,
        "asset_id": asset_id,
        "task_id": task_id,
        "step_id": step_id,
        "remote_action": remote_action,
        "remote_request_id": remote_request_id,
        "api_key_id": api_key_id,
        "message": message,
        "detail": detail or {},
        "error": error,
    }


def log_operation_event(
    *,
    domain: str,
    event_type: str,
    module: str | None = None,
    event_status: str = "success",
    source: str | None = None,
    trace_id: str | None = None,
    request_id: str | None = None,
    user_id: str | None = None,
    project_id: str | None = None,
    session_id: str | None = None,
    group_id: str | None = None,
    asset_id: str | None = None,
    task_id: str | None = None,
    step_id: str | None = None,
    remote_action: str | None = None,
    remote_request_id: str | None = None,
    api_key_id: str | None = None,
    message: str | None = None,
    detail: dict[str, Any] | None = None,
    error: str | None = None,
) -> None:
    _append_operation_log(
        domain,
        _base_entry(
            log_type="operation_event",
            domain=domain,
            module=module,
            event_type=event_type,
            event_status=event_status,
            source=source,
            trace_id=trace_id,
            request_id=request_id,
            user_id=user_id,
            project_id=project_id,
            session_id=session_id,
            group_id=group_id,
            asset_id=asset_id,
            task_id=task_id,
            step_id=step_id,
            remote_action=remote_action,
            remote_request_id=remote_request_id,
            api_key_id=api_key_id,
            message=message,
            detail=detail,
            error=error,
        ),
    )


def log_module_generation_event(
    *,
    module: str,
    event_type: str,
    event_status: str = "success",
    source: str | None = None,
    trace_id: str | None = None,
    request_id: str | None = None,
    user_id: str | None = None,
    project_id: str | None = None,
    task_id: str | None = None,
    step_id: str | None = None,
    remote_action: str | None = None,
    remote_request_id: str | None = None,
    api_key_id: str | None = None,
    message: str | None = None,
    detail: dict[str, Any] | None = None,
    error: str | None = None,
) -> None:
    """Write module business logs to log/ModuleGeneration/{module}/YYYY-MM-DD.log."""
    _append_json_log(
        MODULE_GENERATION_LOG_ROOT,
        module,
        _base_entry(
            log_type="module_generation_event",
            domain="module_generation",
            module=module,
            event_type=event_type,
            event_status=event_status,
            source=source,
            trace_id=trace_id,
            request_id=request_id,
            user_id=user_id,
            project_id=project_id,
            task_id=task_id,
            step_id=step_id,
            remote_action=remote_action,
            remote_request_id=remote_request_id,
            api_key_id=api_key_id,
            message=message,
            detail=detail,
            error=error,
        ),
    )


def log_ai_model_event(
    *,
    event_type: str,
    module: str | None = None,
    step_code: str | None = None,
    call_id: str | None = None,
    event_phase: str | None = None,
    owner_type: str | None = None,
    owner_id: str | None = None,
    generation_attempt_no: int | None = None,
    latency_ms: int | None = None,
    event_status: str = "success",
    source: str | None = None,
    trace_id: str | None = None,
    request_id: str | None = None,
    user_id: str | None = None,
    project_id: str | None = None,
    task_id: str | None = None,
    step_id: str | None = None,
    remote_action: str | None = None,
    remote_request_id: str | None = None,
    model_config_id: str | None = None,
    model_config_name: str | None = None,
    model_name: str | None = None,
    provider: str | None = None,
    api_base: str | None = None,
    http_status: int | None = None,
    request: dict[str, Any] | None = None,
    response: dict[str, Any] | list[Any] | str | None = None,
    token_usage: dict[str, Any] | None = None,
    message: str | None = None,
    detail: dict[str, Any] | None = None,
    error: str | None = None,
) -> None:
    """Write AI model call logs to existing log/AiModel/YYYY-MM-DD.log."""
    final_detail = dict(detail or {})
    if request is not None:
        final_detail["request"] = request
    if response is not None:
        final_detail["response"] = response
    if token_usage is not None:
        final_detail["token_usage"] = token_usage
    entry = _base_entry(
        log_type="ai_model_event",
        domain="ai_model",
        module=module or "ai_model",
        event_type=event_type,
        event_status=event_status,
        source=source,
        trace_id=trace_id,
        request_id=request_id,
        user_id=user_id,
        project_id=project_id,
        task_id=task_id,
        step_id=step_id,
        remote_action=remote_action,
        remote_request_id=remote_request_id,
        message=message,
        detail=final_detail,
        error=error,
    )
    entry.update(
        {
            "call_id": call_id,
            "event_phase": event_phase or event_type,
            "step_code": step_code,
            "owner_type": owner_type,
            "owner_id": owner_id,
            "generation_attempt_no": generation_attempt_no,
            "latency_ms": latency_ms,
            # Preserve the legacy fields for existing log readers while also
            # exposing unambiguous configuration/provider model names.
            "model_name": model_config_name,
            "model_id": model_name,
            "model_config_name": model_config_name,
            "provider_model_name": model_name,
            "model_config_id": model_config_id,
            "provider": provider,
            "api_base": api_base,
            "http_status": http_status,
        }
    )
    _append_json_log(AI_MODEL_LOG_ROOT, None, entry)


def log_operation_error(*, domain: str, event_type: str, exc: BaseException | None = None, detail: dict[str, Any] | None = None, **kwargs: Any) -> None:
    kwargs.setdefault("event_status", "failed")
    kwargs["detail"] = build_exception_detail(exc, detail)
    kwargs.setdefault("error", str(exc) if exc is not None else None)
    log_operation_event(domain=domain, event_type=event_type, **kwargs)


def log_remote_api_event(
    *,
    domain: str,
    remote_action: str,
    event_type: str,
    event_status: str,
    request: dict[str, Any] | None = None,
    response: dict[str, Any] | None = None,
    remote_request_id: str | None = None,
    remote_code: str | None = None,
    remote_message: str | None = None,
    **kwargs: Any,
) -> None:
    detail = dict(kwargs.pop("detail", {}) or {})
    if request is not None:
        detail["request"] = request
    if response is not None:
        detail["response"] = response
    if remote_code is not None:
        detail["remote_code"] = remote_code
    if remote_message is not None:
        detail["remote_message"] = remote_message
    log_operation_event(
        domain=domain,
        event_type=event_type,
        event_status=event_status,
        remote_action=remote_action,
        remote_request_id=remote_request_id,
        detail=detail,
        error=remote_message if event_status == "failed" else None,
        **kwargs,
    )