skill-template/scripts/jiangchang_skill_core/unified_logging.py

"""
统一文件日志：{JIANGCHANG_DATA_ROOT}/{JIANGCHANG_USER_ID}/logs/jiangchang.log
按日轮转；行内带 trace_id 与 skill_slug，便于跨技能排查。
"""

from __future__ import annotations

import logging
import os
import sys
import uuid
from logging.handlers import TimedRotatingFileHandler
from typing import Optional

from .runtime_env import get_data_root, get_user_id

_skill_slug: str = ""
_logger_name: str = ""


def get_unified_logs_dir() -> str:
    path = os.path.join(get_data_root(), get_user_id(), "logs")
    try:
        os.makedirs(path, exist_ok=True)
    except OSError:
        pass
    return path


def get_skill_log_file_path() -> str:
    """与 setup 写入的主日志文件一致（供终端提示等）。"""
    override = (os.getenv("JIANGCHANG_LOG_FILE") or "").strip()
    if override:
        parent = os.path.dirname(os.path.abspath(override))
        if parent:
            try:
                os.makedirs(parent, exist_ok=True)
            except OSError:
                pass
        return os.path.abspath(override)
    return os.path.join(get_unified_logs_dir(), "jiangchang.log")


def ensure_trace_for_process() -> str:
    """本进程调用链 ID：沿用 JIANGCHANG_TRACE_ID，否则生成并写入环境（子进程可继承）。"""
    existing = (os.getenv("JIANGCHANG_TRACE_ID") or "").strip()
    if existing:
        return existing
    tid = uuid.uuid4().hex[:12]
    os.environ["JIANGCHANG_TRACE_ID"] = tid
    return tid


def subprocess_env_with_trace(environ: Optional[dict] = None) -> dict:
    """subprocess.run(..., env=...) 时使用，保证带上当前 trace。"""
    ensure_trace_for_process()
    tid = (os.getenv("JIANGCHANG_TRACE_ID") or "").strip()
    if not tid:
        tid = ensure_trace_for_process()
    base = os.environ if environ is None else environ
    return {**base, "JIANGCHANG_TRACE_ID": tid}


class _SkillContextFilter(logging.Filter):
    def filter(self, record: logging.LogRecord) -> bool:
        record.trace_id = (os.getenv("JIANGCHANG_TRACE_ID") or "").strip() or "-"
        record.skill_slug = _skill_slug or "-"
        return True


_FORMAT = (
    "%(asctime)s | %(levelname)-8s | %(trace_id)s | %(skill_slug)s | %(name)s | %(message)s"
)
_DATEFMT = "%Y-%m-%dT%H:%M:%S"


def _log_level_from_env() -> int:
    v = (os.getenv("JIANGCHANG_LOG_LEVEL") or "INFO").strip().upper()
    return getattr(logging, v, None) or logging.INFO


def _backup_count() -> int:
    try:
        n = int((os.getenv("JIANGCHANG_LOG_BACKUP_COUNT") or "30").strip())
        return max(1, min(n, 365))
    except ValueError:
        return 30


def _attach_fallback_handler(log: logging.Logger, fmt: logging.Formatter) -> None:
    """文件日志不可用时仅挂内存 handler，不向 stderr 打印 logging 异常。"""
    log.addHandler(logging.NullHandler())


def setup_skill_logging(skill_slug: str, logger_name: str) -> None:
    """
    幂等：为指定 logger 挂载统一文件日志与可选 stderr。
    须在进程早期调用（如 CLI main 在业务日志之前）。
    文件日志目录无权限或不可写时降级为 no-op，不影响 CLI 返回码。
    """
    global _skill_slug, _logger_name
    ensure_trace_for_process()
    _skill_slug = skill_slug
    _logger_name = logger_name

    log = logging.getLogger(logger_name)
    if log.handlers:
        return
    log.setLevel(_log_level_from_env())
    fmt = logging.Formatter(_FORMAT, datefmt=_DATEFMT)
    filt = _SkillContextFilter()

    try:
        path = get_skill_log_file_path()
        fh = TimedRotatingFileHandler(
            path,
            when="midnight",
            interval=1,
            backupCount=_backup_count(),
            encoding="utf-8",
            delay=True,
        )
        fh.setFormatter(fmt)
        fh.addFilter(filt)
        log.addHandler(fh)
    except (OSError, PermissionError, ValueError):
        _attach_fallback_handler(log, fmt)
    except Exception:
        _attach_fallback_handler(log, fmt)

    if (os.getenv("JIANGCHANG_LOG_TO_STDERR") or "").strip().lower() in (
        "1",
        "true",
        "yes",
        "on",
    ):
        try:
            sh = logging.StreamHandler(sys.stderr)
            sh.setLevel(logging.WARNING)
            sh.setFormatter(fmt)
            sh.addFilter(filt)
            log.addHandler(sh)
        except Exception:
            pass
    log.propagate = False


def get_skill_logger() -> logging.Logger:
    if not _logger_name:
        raise RuntimeError("get_skill_logger: call setup_skill_logging first")
    return logging.getLogger(_logger_name)


def attach_unified_file_handler(
    log_path: str,
    *,
    skill_slug: str,
    logger_name: str,
    level: int = logging.DEBUG,
) -> logging.Logger:
    """
    独立子进程内追加同一日志文件（如 login 检测子进程），格式与主进程一致。
    """
    global _skill_slug
    ensure_trace_for_process()
    _skill_slug = skill_slug
    lg = logging.getLogger(logger_name)
    lg.handlers.clear()
    lg.setLevel(level)
    fmt = logging.Formatter(_FORMAT, datefmt=_DATEFMT)
    try:
        parent = os.path.dirname(os.path.abspath(log_path))
        if parent:
            os.makedirs(parent, exist_ok=True)
        fh = logging.FileHandler(log_path, encoding="utf-8")
        fh.setFormatter(fmt)
        fh.addFilter(_SkillContextFilter())
        lg.addHandler(fh)
    except (OSError, PermissionError, ValueError):
        lg.addHandler(logging.NullHandler())
    except Exception:
        lg.addHandler(logging.NullHandler())
    lg.propagate = False
    return lg