guanlan-src/app_common/app_common_guanlan/logfile.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
r"""日志目录/命名/格式的唯一口径 (2026-09-17 用户令 2「统一日志输出目录及日志文件命名、内容格式」)。

## 统一前是什么样 (实测, 不是推测)

  · 目录: 运行日志散在 `logs/`, 但 CMS 插件把 `analyze_stdout.log` 写进了**产物目录**
    (`outputs/<场>/windcms/...`), `_proc_reg.log` 这种自检残留也躺在 `logs/` 里;
  · 命名: 服务日志是 `<组件>.log` (gateway/detail/cms/sim/sim_sys/viewer), 但对不上服务键的还有
    `serve.log`、`start_hidden.log`; 运维动作是 `ops_<动作>_<YYYYmmdd_HHMMSS>.log` 直接堆在 `logs/` 顶层
    —— 实测 34 个文件里 22 个是历史动作日志, 没有保留策略, 只会越堆越多;
  · 内容: **没有时间戳、没有级别、编码与行尾都不统一** —— `detail.log` 首行是 `b'windscada serve :18033\r\n'`(CRLF!),
    `cms.log` 首行甚至是一条历史 `SyntaxWarning`; 机器审计反而叫 `.jsonl` 混在 `.log` 里;
  · 全仓 `import logging` 的文件数 = **0** —— 没有统一设施, 每个进程各 print 各的。

## 统一后的口径 (机器可查, 见 scripts/log_audit.py)

    logs/<组件>.log                长驻服务与启动器 (组件名 = configs/serve.json 的键 + gateway + serve/start_hidden)
    logs/ops/<动作>_<YYYYmmdd-HHMMSS>.log   运维动作日志 (保留最近 20 份 / 30 天, 超出的自动清理)
    logs/audit/<名字>.jsonl        机器审计流水 (JSON Lines: 一行一条 JSON)

    行格式 (每一行都要满足, 校验正则见 LINE_RE):
        YYYY-MM-DD HH:MM:SS LEVEL 组件 消息
    级别: DEBUG/INFO/WARN/ERROR;  UTF-8 无 BOM;  行尾 LF;  不含 ANSI 颜色码。

服务侧怎么落地: 各服务不用改自己的 print —— 入口处调一次 `prefix_stdout(组件名)`,
之后**每一行**都会自动带上时间戳/级别/组件 (实现见下面 _Prefixed 包装器)。这样"内容格式统一"
不是靠自觉, 而是由设施保证。
"""
from __future__ import annotations

try:
    from ._root import install_root as _install_root
except ImportError:                     # 直接当脚本跑（python <本文件>）时没有包上下文
    from _root import install_root as _install_root

import datetime as dt
import json
import os
import pathlib
import re
import sys

import pathlib as _p

ROOT = _install_root(__file__)
LOGS = _p.Path(os.environ.get('WINDSCADA_LOGS') or (ROOT / 'logs'))
OPS_DIR = LOGS / 'ops'
AUDIT_DIR = LOGS / 'audit'
BUILD_DIR = LOGS / 'build'          # 构建/摄入类脚本的日志 (按产物子路径归档; 不许写在产物目录里)
LEVELS = ('DEBUG', 'INFO', 'WARN', 'ERROR')

# 一行日志的规范形式: 时间戳 + 级别 + 组件 + 消息 (组件名允许中文/点/下划线/连字符)
LINE_RE = re.compile(r'^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2} (?:DEBUG|INFO|WARN|ERROR) [\w\u4e00-\u9fff.\-]+ ')
ANSI_RE = re.compile(r'\x1b\[[0-9;]*m')

ACTION_KEEP = 20          # 运维动作日志保留份数
ACTION_DAYS = 30          # 运维动作日志保留天数


def now() -> str:
    return dt.datetime.now().strftime('%Y-%m-%d %H:%M:%S')


def line(level: str, comp: str, msg: str, ts: str | None = None) -> str:
    """拼一行规范日志 (纯函数, 便于测试)。多行消息会被逐行加前缀。"""
    lv = (level or 'INFO').upper()
    if lv not in LEVELS:
        lv = 'INFO'
    out = []
    for i, part in enumerate(str(msg).replace('\r\n', '\n').replace('\r', '\n').split('\n')):
        out.append(f'{ts or now()} {lv} {comp} {part}')
    return '\n'.join(out)


def component_log(comp: str) -> pathlib.Path:
    """长驻服务/启动器的日志路径: logs/<组件>.log"""
    return LOGS / f'{comp}.log'


def action_log(tag: str, when: dt.datetime | None = None) -> pathlib.Path:
    """运维动作日志路径: logs/ops/<动作>_<YYYYmmdd-HHMMSS>.log (写入前会先按保留策略清理)"""
    w = when or dt.datetime.now()
    return OPS_DIR / f'{tag}_{w:%Y%m%d-%H%M%S}.log'


def audit_log(name: str) -> pathlib.Path:
    """机器审计流水路径: logs/audit/<名字>.jsonl"""
    return AUDIT_DIR / (name if name.endswith(('.jsonl', '.json')) else f'{name}.jsonl')


def ensure_dirs() -> None:
    for d in (LOGS, OPS_DIR, AUDIT_DIR, BUILD_DIR):
        d.mkdir(parents=True, exist_ok=True)


def build_log(rel: str, farm: str | None = None) -> pathlib.Path:
    """构建/摄入脚本的日志路径: `logs/build/<场>/<相对路径>.log`

    ★ 2026-09-17 用户令 2 的由来: 交付包里 39 个构建日志原本**躺在产物目录里**
    (`outputs/<场>/windscada/*.log` 等), 还被产物台账登记成 shipped 随包件 ——
    "日志在产物里"正是这次要统一掉的不一致。新脚本用本函数落 `logs/build/`, 旧的已迁移过去。
    """
    p = pathlib.Path(rel)
    parts = [farm] if farm else []
    return BUILD_DIR.joinpath(*parts, *p.parts)


def write(comp: str, msg: str, level: str = 'INFO', path: pathlib.Path | None = None) -> pathlib.Path:
    """追加一行规范日志 (UTF-8, LF)。"""
    ensure_dirs()
    p = pathlib.Path(path) if path else component_log(comp)
    p.parent.mkdir(parents=True, exist_ok=True)
    with open(p, 'a', encoding='utf-8', newline='\n') as f:
        f.write(line(level, comp, msg) + '\n')
    return p


def append_jsonl(name: str, rec: dict) -> pathlib.Path:
    """机器审计流水: 一行一条 JSON (ensure_ascii=False, 时间戳字段 ts)。"""
    ensure_dirs()
    p = audit_log(name)
    rec = dict(rec)
    rec.setdefault('ts', now())
    with open(p, 'a', encoding='utf-8', newline='\n') as f:
        f.write(json.dumps(rec, ensure_ascii=False, default=str) + '\n')
    return p


def prune_actions(keep: int = ACTION_KEEP, days: int = ACTION_DAYS, dry: bool = False) -> list[pathlib.Path]:
    """运维动作日志保留策略: 只留最近 keep 份、且不超过 days 天。→ 删掉的文件列表。"""
    if not OPS_DIR.is_dir():
        return []
    files = sorted((f for f in OPS_DIR.glob('*.log') if f.is_file()), key=lambda f: f.stat().st_mtime, reverse=True)
    cutoff = dt.datetime.now().timestamp() - days * 86400
    gone = []
    for i, f in enumerate(files):
        if i < keep and f.stat().st_mtime >= cutoff:
            continue
        gone.append(f)
        if not dry:
            try:
                f.unlink()
            except OSError:
                pass
    return gone


def normalize_file(path: pathlib.Path, comp: str, when: float | None = None) -> int:
    """把一份**裸输出**日志就地补成统一格式 → 补了几行。

    为什么需要: 运维动作的日志是"子进程直接写 fd"的产物 (`_ops_run.py` 把本进程的 stdout 句柄交给子命令,
    见那里的注释 —— 用管道收输出曾经导致日志空白/任务看起来卡住), 所以子命令的 print **绕过**了 Python
    层的流包装。这里在动作结束时补一次前缀: 已经是规范行的原样保留, 其余行补 `时间戳 级别 组件.raw`,
    并在开头插一行说明"以下为原样输出、前缀是事后补的" —— 不假装是原始时刻写的。
    """
    if not path or not pathlib.Path(path).is_file():
        return 0
    p = pathlib.Path(path)
    ts = dt.datetime.fromtimestamp(when if when is not None else p.stat().st_mtime).strftime('%Y-%m-%d %H:%M:%S')
    raw = p.read_text(encoding='utf-8', errors='replace').replace('\r\n', '\n').split('\n')
    fixed, n = [], 0
    for ln in raw:
        if not ln.strip():
            continue
        if LINE_RE.match(ln):
            fixed.append(ln)
            continue
        fixed.append(f'{ts} INFO {comp}.raw {ln}')
        n += 1
    if n:
        fixed.insert(0, f'{ts} INFO {comp} === 以下 {n} 行为子进程原样输出 (前缀由 logfile.normalize_file 事后补齐) ===')
        p.write_text('\n'.join(fixed) + '\n', encoding='utf-8', newline='\n')
    return n


# ── 服务侧: 让 print 出来的每一行都符合格式 ─────────────────────────────────────────────
class _Prefixed:
    """把写到 stdout/stderr 的每一行加上 `时间戳 级别 组件` 前缀。

    刻意做成**流包装**而不是要求 6 个服务各自改用 logging:
    服务里已有大量 print (启动横幅/请求日志/进度), 逐个改既费事又会漏; 包一层之后
    "内容格式统一"由设施保证。级别默认 INFO; 含 'error'/'traceback' 字样的行按 ERROR 记。
    """

    def __init__(self, stream, comp: str):
        self._s = stream
        self._comp = comp
        self._buf = ''

    def write(self, data):
        if not isinstance(data, str):
            data = str(data)
        self._buf += data
        while '\n' in self._buf:
            ln, self._buf = self._buf.split('\n', 1)
            self._emit(ln)
        return len(data)

    def _emit(self, ln: str):
        clean = ANSI_RE.sub('', ln.rstrip('\r'))
        lv = 'ERROR' if re.search(r'error|traceback|失败|异常', clean, re.I) else 'INFO'
        try:
            self._s.write(line(lv, self._comp, clean) + '\n')
            self._s.flush()
        except Exception:
            pass

    def flush(self):
        if self._buf:
            self._emit(self._buf)
            self._buf = ''
        try:
            self._s.flush()
        except Exception:
            pass

    def __getattr__(self, item):
        return getattr(self._s, item)


def prefix_stdout(comp: str) -> None:
    """服务入口调一次: 之后 stdout/stderr 的每一行都符合统一格式 (幂等)。"""
    if getattr(sys.stdout, '_guanlan_prefixed', False) or os.environ.get('GUANLAN_LOG_RAW'):
        return
    so, se = _Prefixed(sys.stdout, comp), _Prefixed(sys.stderr, comp)
    so._guanlan_prefixed = se._guanlan_prefixed = True
    sys.stdout, sys.stderr = so, se


if __name__ == '__main__':      # 直接跑 = 打印口径 + 清洗一次动作日志
    ensure_dirs()
    print(f'logs 目录: {LOGS}')
    print(f'  服务日志   logs/<组件>.log       (组件: configs/serve.json 的键 + gateway/serve/start_hidden)')
    print(f'  动作日志   logs/ops/<动作>_<YYYYmmdd-HHMMSS>.log   保留最近 {ACTION_KEEP} 份 / {ACTION_DAYS} 天')
    print(f'  审计流水   logs/audit/<名字>.jsonl')
    print(f'  行格式     {line("INFO", "示例", "一行长这样")}')
    gone = prune_actions(dry=True)
    print(f'  现有动作日志 {len(list(OPS_DIR.glob("*.log")))} 份, 按保留策略该清理 {len(gone)} 份')