日志规范 (Logging Standard)

定位：前后端统一日志体系，确保每条日志可追溯、可定位、可为 Bug 排查提供完整线索。

共享包: 后端日志已封装为独立共享包 logkit，详见 log_share.md。其它项目可通过 pip install -e D:/product/AIasMe/src/logkit 直接使用。

---

一、设计原则

原则	说明
完整性	关键路径每一步都留痕，失败必须记录原因和上下文
结构化	统一 [模块][操作] key=value 格式，便于 grep / jq 分析
分级明确	严格按级别使用，生产环境 INFO 以上，开发环境 DEBUG
可串联	同一请求链路用 trace_id 串联前后端所有日志
安全	禁止记录密码、Token 原文、Cookie 完整值等敏感信息

---

二、后端日志规范 (Python / FastAPI)

2.1 目录结构

backend/
├── log/                       # 日志目录（gitignore）
│   ├── app_2026-05-27.log     # 当天主日志
│   ├── app_2026-05-27.log.1   # 轮转备份（20MB/个，最多 10 个）
│   ├── error_2026-05-27.log   # 当天错误日志（仅 ERROR+）
│   └── access_2026-05-27.log  # 当天请求日志
└── utils/
    └── logger.py              # 统一日志配置模块

2.2 Logger 实现（参考 src/utils/logger.py）

# backend/utils/logger.py

import os
import logging
import logging.handlers
from datetime import datetime

LOG_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), "log")

# 日志格式
DETAIL_FORMAT = logging.Formatter(
    "%(asctime)s | %(levelname)-7s | %(name)s | %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S"
)


def setup_logging(app_name: str = "wechat", level: int = logging.DEBUG):
    """应用启动时调用一次，配置全局日志"""
    os.makedirs(LOG_DIR, exist_ok=True)
    today = datetime.now().strftime("%Y-%m-%d")

    root = logging.getLogger()
    root.setLevel(level)

    # 1. 主日志：全量，按大小轮转
    app_handler = logging.handlers.RotatingFileHandler(
        os.path.join(LOG_DIR, f"{app_name}_{today}.log"),
        maxBytes=20 * 1024 * 1024,  # 20MB
        backupCount=10,
        encoding="utf-8",
    )
    app_handler.setLevel(logging.DEBUG)
    app_handler.setFormatter(DETAIL_FORMAT)

    # 2. 错误日志：仅 ERROR 及以上，单独存储便于告警
    error_handler = logging.handlers.RotatingFileHandler(
        os.path.join(LOG_DIR, f"error_{today}.log"),
        maxBytes=10 * 1024 * 1024,
        backupCount=5,
        encoding="utf-8",
    )
    error_handler.setLevel(logging.ERROR)
    error_handler.setFormatter(DETAIL_FORMAT)

    # 3. 控制台输出（开发环境用，生产可关闭）
    console = logging.StreamHandler()
    console.setLevel(logging.INFO)
    console.setFormatter(DETAIL_FORMAT)

    root.addHandler(app_handler)
    root.addHandler(error_handler)
    root.addHandler(console)

    # 抑制第三方库的 DEBUG 噪音
    for lib in ("uvicorn", "httpx", "httpcore", "charset_normalizer", "urllib3"):
        logging.getLogger(lib).setLevel(logging.WARNING)

    logging.getLogger("apscheduler").setLevel(logging.INFO)

2.3 在 main.py 中初始化

# main.py
from utils.logger import setup_logging
setup_logging(app_name="wechat")

# 替代原来的 print()
logging.getLogger("wechat").info(
    "session restored from db | token=%s...", token[:10]
)

2.4 日志级别定义

级别	含义	使用场景
DEBUG	调试信息	变量值、中间态、详细的 API 请求/响应体
INFO	关键操作	任务开始/结束、数据量统计、状态变更
WARNING	异常但可恢复	重试、降级、频率限制触发、非预期但可处理
ERROR	操作失败	爬取失败、数据库写入失败、会话过期
CRITICAL	系统级故障	数据库连接断开、磁盘满、关键进程崩溃

2.5 关键日志埋点（必须记录）

爬取流程

# 任务启动
logger.info("crawl_start | account_id=%d | account_name=%s | limit=%s", aid, name, limit)

# 每页请求（DEBUG 级别，避免生产刷屏）
logger.debug("fetch_page | account_id=%d | page=%d | offset=%d", aid, page, offset)

# 频率限制
logger.warning("rate_limited | account_id=%d | wait=%.1fs | retry_count=%d", aid, wait, retry)

# 每页结果
logger.info("page_done | account_id=%d | page=%d | articles=%d | total=%d", aid, page, count, total)

# 任务完成
logger.info("crawl_done | account_id=%d | total=%d | status=%s | duration=%.1fs", aid, total, status, duration)

# 任务失败
logger.error("crawl_failed | account_id=%d | reason=%s | url=%s", aid, reason, url, exc_info=True)

文章内容获取

logger.info("content_fetch_start | article_id=%d | title=%s", aid, title[:50])
logger.warning("content_fetch_fail | article_id=%d | url=%s | reason=%s", aid, url, reason)
logger.info("content_fetch_done | article_id=%d | content_len=%d", aid, len(content))

数据库操作

logger.info("db_save | table=articles | count=%d | new=%d", total, new_count)
logger.error("db_error | table=%s | operation=%s | error=%s", table, op, e, exc_info=True)

认证/Session

logger.info("login_qr_generated | expire_at=%s", expire)
logger.info("login_success | token=%s...", token[:10])
logger.warning("session_expired | token=%s... | endpoint=%s", token[:10], api)

2.6 不记录的内容（安全规范）

• 完整 Cookie（只记录前 20 字符）
• 密码、Secret Key
• 用户手机号/身份证
• 文章正文完整内容（只记录长度）

---

三、前端日志规范 (Vue 3 / TypeScript)

3.1 设计定位

前端日志不落地本地文件（浏览器安全限制），而是： 1. console 输出：开发调试用，按级别分组 2. 上报后端：关键错误和用户操作通过 POST /api/log 发送到后端存储

3.2 基础实现

// frontend/src/utils/logger.ts

type LogLevel = 'debug' | 'info' | 'warn' | 'error'

interface LogEntry {
    level: LogLevel
    module: string
    message: string
    data?: Record<string, unknown>
    timestamp: string
    url: string
}

const LOG_LEVEL_ORDER: Record<LogLevel, number> = {
    debug: 0, info: 1, warn: 2, error: 3,
}

let currentLevel: LogLevel = import.meta.env.DEV ? 'debug' : 'info'

class Logger {
    private module: string

    constructor(module: string) {
        this.module = module
    }

    private log(level: LogLevel, message: string, data?: Record<string, unknown>) {
        if (LOG_LEVEL_ORDER[level] < LOG_LEVEL_ORDER[currentLevel]) return

        const entry: LogEntry = {
            level,
            module: this.module,
            message,
            data,
            timestamp: new Date().toISOString(),
            url: window.location.href,
        }

        // 控制台输出
        const consoleFn = level === 'error' ? console.error
            : level === 'warn' ? console.warn
            : console.log
        consoleFn(`[${entry.module}] ${message}`, data || '')

        // ERROR 级别自动上报后端
        if (level === 'error') {
            this.report(entry)
        }
    }

    debug(msg: string, data?: Record<string, unknown>) { this.log('debug', msg, data) }
    info(msg: string, data?: Record<string, unknown>) { this.log('info', msg, data) }
    warn(msg: string, data?: Record<string, unknown>) { this.log('warn', msg, data) }
    error(msg: string, data?: Record<string, unknown>) { this.log('error', msg, data) }

    private report(entry: LogEntry) {
        // 使用 sendBeacon 确保页面关闭时也能发送
        const blob = new Blob([JSON.stringify(entry)], { type: 'application/json' })
        if (navigator.sendBeacon) {
            navigator.sendBeacon('/api/log', blob)
        } else {
            fetch('/api/log', {
                method: 'POST',
                body: JSON.stringify(entry),
                headers: { 'Content-Type': 'application/json' },
                keepalive: true,
            }).catch(() => { /* 静默失败，避免循环 */ })
        }
    }
}

// 模块级工厂函数
export function createLogger(module: string) {
    return new Logger(module)
}

// 设置全局级别
export function setLogLevel(level: LogLevel) {
    currentLevel = level
}

3.3 使用方式

// 组件中使用
import { createLogger } from '@/utils/logger'
const log = createLogger('ArticleList')

async function startFetchContent() {
    log.info('fetch_content_start', { accountId: accountId.value })
    try {
        const res = await fetchBatchContent(accountId.value, 20)
        log.info('fetch_content_ok', { taskId: res.data.task_id, total: res.data.total })
    } catch (e: any) {
        log.error('fetch_content_fail', {
            error: e.message,
            status: e.response?.status,
            accountId: accountId.value,
        })
    }
}

3.4 前端关键日志埋点

场景	级别	示例
页面进入	DEBUG	nav_to | path=/articles?account_id=28
API 请求	DEBUG	api_call | method=GET | url=/api/accounts/28/crawl-status
API 成功	DEBUG	api_ok | method=POST | url=/api/accounts/28/crawl | status=200
API 失败	ERROR	api_fail | method=GET | url=... | status=500 | err=...
用户操作	INFO	user_action | action=crawl | account=刘润 | limit=10
渲染错误	ERROR	render_error | component=ArticleList | err=...

---

四、Trace ID 串联前后端

4.1 生成规则

# 后端中间件 — 每个请求生成唯一 trace_id
import uuid
from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware

class TraceMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        trace_id = request.headers.get("X-Trace-Id", uuid.uuid4().hex[:12])
        request.state.trace_id = trace_id
        response = await call_next(request)
        response.headers["X-Trace-Id"] = trace_id
        return response

app.add_middleware(TraceMiddleware)

4.2 前端注入

// frontend/src/api/index.ts — Axios 拦截器
import { v4 as uuid } from 'uuid'  // 或 crypto.randomUUID()

api.interceptors.request.use(config => {
    const traceId = uuid().slice(0, 12)
    config.headers['X-Trace-Id'] = traceId
    return config
})

4.3 后端记录 trace_id

import logging

# 绑定到当前上下文（可以用 ContextVar）
_trace_id: ContextVar[str] = ContextVar("trace_id", default="-")

def get_logger(name: str) -> logging.LoggerAdapter:
    logger = logging.getLogger(name)
    return logging.LoggerAdapter(logger, {"trace_id": _trace_id.get()})

# 使用
logger = get_logger("wechat.crawler")
logger.info("crawl_start | ...")  # 自动附加 trace_id

4.4 排查示例

# 通过 trace_id 串联整个请求链路的所有日志
grep "trace_id=a1b2c3d4" backend/log/app_2026-05-27.log | sort

# 输出示例：
# 2026-05-27 14:30:01 | INFO    | wechat.crawl | trace_id=a1b2c3d4 | crawl_start | account_id=28
# 2026-05-27 14:30:03 | INFO    | wechat.crawl | trace_id=a1b2c3d4 | page_done | page=1 | articles=5
# 2026-05-27 14:30:05 | WARNING | wechat.crawl | trace_id=a1b2c3d4 | rate_limited | wait=10.3s
# 2026-05-27 14:30:16 | INFO    | wechat.crawl | trace_id=a1b2c3d4 | page_done | page=2 | articles=5
# 2026-05-27 14:30:16 | INFO    | wechat.crawl | trace_id=a1b2c3d4 | crawl_done | total=10 | status=success

---

五、启动配置（主入口注册日志）

修改 backend/main.py，在创建 app 后立即初始化日志：

from fastapi import FastAPI
from utils.logger import setup_logging

setup_logging(app_name="wechat")

app = FastAPI(title="微言 WeChat Scanner", version="1.0.0")
# ... 其余代码

---

六、日志轮转与清理策略

策略	配置	说明
大小轮转	20MB/文件，保留 10 个	RotatingFileHandler
时间轮转	每天新文件（文件名含日期）	启动时检查日期
自动清理	保留最近 30 天日志	定时任务或启动时清理
错误日志	10MB/文件，保留 5 个	仅 ERROR+，长期保留

# 自动清理函数（可加在 on_startup 中）
import glob, time

def cleanup_old_logs(keep_days: int = 30):
    """删除 N 天前的日志文件"""
    cutoff = time.time() - keep_days * 86400
    for f in glob.glob(os.path.join(LOG_DIR, "*.log*")):
        if os.path.getmtime(f) < cutoff:
            os.remove(f)
            print(f"[log cleanup] removed {f}")

---

七、部署环境日志差异

环境	控制台输出	文件日志	级别	第三方库级别
开发	INFO 彩色	DEBUG 全量	DEBUG	WARNING
测试	INFO	INFO	INFO	WARNING
生产	关闭	INFO	INFO	ERROR

通过环境变量切换：

LOG_LEVEL = os.getenv("LOG_LEVEL", "DEBUG" if os.getenv("ENV") != "production" else "INFO")
setup_logging(level=getattr(logging, LOG_LEVEL))

---

八、Checklist（实施清单）

• [ ] backend/utils/logger.py — 统一日志模块
• [ ] backend/main.py — 调用 setup_logging()
• [ ] 各模块 print() 改为 logger.info/warning/error()
• [ ] 爬取流程埋点（start/done/fail/rate_limit）
• [ ] 异常捕获必须带 exc_info=True
• [ ] frontend/src/utils/logger.ts — 前端日志模块
• [ ] 前端 API 拦截器注入 X-Trace-Id
• [ ] 后端 TraceMiddleware 中间件
• [ ] log/ 目录加入 .gitignore
• [ ] 日志清理定时任务
• [ ] 确认不记录敏感信息

---

九、Bug 排查工作流

当用户报告 Bug 时，按以下步骤排查：

1. 获取发生时间点
   → 用户截图上的时间 / 前端日志时间戳

2. 搜时间窗口（前后 5 分钟）
   grep "2026-05-27 14:3[0-5]" backend/log/app_*.log

3. 搜关键词快速定位
   grep -i "error\|fail\|exception" backend/log/app_*.log

4. 通过 Trace ID 串联完整链路
   grep "trace_id=xxx" backend/log/app_*.log | sort

5. 前端日志（浏览器 Console 导出）
   → 按 timestamp 过滤

6. 锁定根因后记录结论
   → 在 Bug 单/ZhiYan 中添加日志片段作为证据