日志 loguru

def publish_article(article_id: int) -> None:
    print("开始发布")
    print(article_id)
    ...
    print("发布完成")  # 到底执行到哪一句了？几点执行的？哪个请求触发的？

import logging
import logging.handlers

logger = logging.getLogger("inkhub")

formatter = logging.Formatter(
    "%(asctime)s | %(levelname)-7s | %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)

console = logging.StreamHandler()
console.setLevel(logging.INFO)
console.setFormatter(formatter)

file_handler = logging.handlers.RotatingFileHandler(
    "inkhub.log", maxBytes=10 * 1024 * 1024, backupCount=5, encoding="utf-8",
)
file_handler.setLevel(logging.DEBUG)
file_handler.setFormatter(formatter)

logger.setLevel(logging.DEBUG)
logger.addHandler(console)
logger.addHandler(file_handler)

uv add loguru

logger.add(
    sink,                    # 出口：终端 / 文件路径 / 文件对象 / 自定义函数 / 协程……
    level="DEBUG",           # 这个 sink 只接收该级别及以上的消息
    format="{time} | {level} | {message}",   # 格式模板
    filter=None,             # 过滤器：更细粒度地决定哪些消息进来（第 4 节）
    serialize=False,         # True 则每条日志序列化成一行 JSON（第 6 节）
    backtrace=True,          # 异常链完整展开（第 8 节）
    diagnose=True,           # 堆栈里标注变量值（第 8 节，生产要关！）
    enqueue=False,           # True 则走队列写入，多进程安全（第 11 节）
    # sink 是文件路径时，还能透传这些文件专属参数（第 5 节）：
    # rotation=..., retention=..., compression=..., encoding=...
) -> int                     # 返回 sink 的 id，remove 时要用

import sys
from loguru import logger

logger.remove()  # 撤掉默认的终端 sink（不可定制，留着它日志会双份）
logger.configure(extra={"request_id": "-"})  # 给 extra 设默认值，见第 4 节的坑

logger.add(sys.stderr, level="INFO",
           format="<green>{time:HH:mm:ss}</green> | <level>{level: <7}</level> | {message}")

logger.add("logs/inkhub_{time:YYYYMMDD}.log", level="DEBUG",
           rotation="100 MB", retention="30 days", encoding="utf-8")

logger.info("墨集日志系统装配完成")
logger.success("文章发布完成：article_id=101")

logger.add(sys.stderr, level="INFO")            # 终端：INFO 起步，DEBUG 刷屏免了
logger.add("logs/inkhub.log", level="DEBUG")    # 文件：DEBUG 起步，排障要的细节都在

logger.debug("生成 slug：loguru-十分钟上手")   # 只进文件
logger.info("文章发布完成")                     # 终端 + 文件都进

# ① 函数：接收 record 字典，返回 True 才放行
logger.add("logs/billing.log",
           filter=lambda record: record["extra"].get("billing"))

# ② 字符串：只收某个模块（按 logger 名）的日志
logger.add("logs/payment.log", filter="inkhub.payment")

# ③ 字典：不同模块给不同级别（模块名 → 门槛）
logger.add(sys.stderr, filter={"inkhub": "INFO", "sqlalchemy.engine": "WARNING"})

logger.add(
    "logs/inkhub_{time:YYYYMMDD}.log",
    rotation="100 MB",       # ① 什么时候换新文件
    retention="30 days",     # ② 旧文件留多久
    compression="zip",       # ③ 退下来的旧文件压成压缩包
)

rotation="100 MB"     # 按大小：写满 100 MB 就换新文件（最常用）
rotation="00:00"      # 按时刻：每天零点换（"midnight" 等价）
rotation="1 week"     # 按间隔：每 7 天换（"Monday" 表示每周一换）

retention="30 days"   # 只留最近 30 天的旧档
retention=10          # 只留最近 10 个旧档

compression="zip"     # 旧档自动压成 .zip（"gz" 也常用）

logger.add(
    "logs/inkhub_{time:YYYYMMDD_HHmmss}.log",
    level="DEBUG",
    rotation="100 MB",       # 单文件最多 100 MB
    retention="30 days",     # 旧档留一个月
    compression="zip",       # 旧档压缩，磁盘占用再砍一刀
    encoding="utf-8",
)

logger.add("logs/inkhub.json", serialize=True, level="INFO")

{"text": "2026-08-27 10:32:11.402 | INFO    | req-a1b2c3 | 文章发布完成：article_id=101\n", "record": {"elapsed": {"repr": "...", "seconds": 3.2}, "exception": null, "extra": {"request_id": "req-a1b2c3"}, "file": {"name": "publish.py", "path": "..."}, "function": "publish_article", "level": {"icon": "ℹ️", "name": "INFO", "no": 20}, "line": 42, "message": "文章发布完成：article_id=101", "module": "publish", "name": "inkhub.publish", "process": {"id": 90367, "name": "MainProcess"}, "thread": {"id": 1, "name": "MainThread"}, "time": {"repr": "...", "timestamp": 1770000000.0}}}

log = logger.bind(request_id="req-a1b2c3")
log.info("收到发布请求")        # 这两条日志的 extra 里有 request_id
log.success("文章发布完成")     # 这条也是
logger.info("另一条普通日志")    # 这条没有——bind 只作用于返回的那个 log 对象

with logger.contextualize(request_id="req-a1b2c3"):
    logger.info("请求开始")            # 自动带 request_id
    publish_article(101, "...")        # 函数内部所有 logger 调用都带（不用传参！）
    logger.info("请求结束")            # 还是带
logger.info("出了作用域")              # 不带了

import contextvars

request_id_var: contextvars.ContextVar[str] = contextvars.ContextVar(
    "request_id", default="-"
)

logger.configure(patcher=lambda record: record["extra"].update(
    request_id=request_id_var.get()
))

def publish_article(article_id: int) -> None:
    try:
        sync_to_search(article_id)
    except TypeError:
        # 显式带上堆栈——不需要在 except 块里，任何地方都能用
        logger.opt(exception=True).error("同步搜索引擎失败：article_id={}", article_id)

10:32:11 | ERROR | req-g7h8i9 | 同步搜索引擎失败：article_id=103
Traceback (most recent call last):

  File "publish.py", line 93, in publish_article
    sync_to_search(slug, article_id)
    │              │     └ 103
    │              └ 'loguru-十分钟上手'
    └ <function sync_to_search at 0x1096a2790>

  File "publish.py", line 83, in sync_to_search
    url = "/search/index/" + article_id
                             └ 103

TypeError: can only concatenate str (not "int") to str

logger.add("logs/inkhub.log", diagnose=False)   # 生产：堆栈照打，变量值不打

def build_tracking_payload(article_id: int) -> str:
    ...  # 读缓存、拼 JSON，一套昂贵操作

# 生产 sink 是 INFO 起步：这条 DEBUG 没人收，lambda 根本不执行，零开销
logger.opt(lazy=True).debug("埋点：{}", lambda: build_tracking_payload(101))

# 万一临时把 sink 调到 DEBUG 排障，lambda 自动开始执行，一个字不用改

import inspect
import logging
from loguru import logger


class InterceptHandler(logging.Handler):
    """标准 logging → loguru 的摆渡人。"""

    def emit(self, record: logging.LogRecord) -> None:
        # ① 级别映射：标准 logging 的级别名转成 loguru 的名字；
        #    对不上号（loguru 特有的 TRACE/SUCCESS）就退回数字级别
        try:
            level: str | int = logger.level(record.levelname).name
        except ValueError:
            level = record.levelno

        # ② 栈深度探测：往上翻调用栈，跳过 logging 模块自己的帧，
        #    让 loguru 显示的「文件:行号」是真正发起调用的地方
        frame, depth = inspect.currentframe(), 0
        while frame and (depth == 0 or frame.f_code.co_filename == logging.__file__):
            frame = frame.f_back
            depth += 1

        # ③ 转发：带上原始堆栈（record.exc_info），按探测到的深度记录
        logger.opt(depth=depth, exception=record.exc_info).log(
            level, record.getMessage()
        )

logging.basicConfig(handlers=[InterceptHandler()], level=0, force=True)

# 收编前：uvicorn 的原生格式，跟 loguru 的格式打架
INFO:     127.0.0.1:5173 - "POST /api/articles HTTP/1.1" 201 Created

# 收编后：uvicorn 的日志也走 loguru 格式，进 loguru 的所有 sink
10:32:11 | INFO    | - | 127.0.0.1:5173 - "POST /api/articles HTTP/1.1" 201

import logging
import sys

from loguru import logger


def setup_logging() -> None:
    """墨集日志总装：终端看 INFO，文件收 DEBUG，标准 logging 全部收编。"""
    logger.remove()
    logger.configure(extra={"request_id": "-"})

    # 终端：短格式，开发时盯屏幕
    logger.add(
        sys.stderr,
        level="INFO",
        format="<green>{time:HH:mm:ss}</green> | <level>{level: <7}</level> "
               "| {extra[request_id]} | {message}",
    )

    # 文件：长格式，带轮转（参数含义见第 5 节）
    logger.add(
        "logs/inkhub_{time:YYYYMMDD_HHmmss}.log",
        level="DEBUG",
        format="{time:YYYY-MM-DD HH:mm:ss.SSS} | {level: <7} "
               "| {extra[request_id]} | {name}:{function}:{line} | {message}",
        rotation="100 MB",
        retention="30 days",
        compression="zip",
        encoding="utf-8",
        diagnose=False,          # 生产环境：堆栈要，变量值不要（第 8.2 节的坑）
    )

    # 收编标准 logging（第 9 节）
    logging.basicConfig(handlers=[InterceptHandler()], level=0, force=True)

import uuid
from collections.abc import Awaitable, Callable

from fastapi import Request, Response


@app.middleware("http")
async def request_id_middleware(
    request: Request, call_next: Callable[[Request], Awaitable[Response]]
) -> Response:
    rid = uuid.uuid4().hex[:8]          # 每次请求一枚新牌子
    with logger.contextualize(request_id=rid):
        logger.info("请求进来：{} {}", request.method, request.url.path)
        response = await call_next(request)   # 下游所有日志自动带 rid
        logger.info("请求结束：{} {}", request.url.path, response.status_code)
    response.headers["X-Request-Id"] = rid    # 回写响应头，用户报障时报这个号
    return response

10:32:11 | INFO    | cc3fbd41 | 请求进来：GET /api/articles/1
10:32:11 | INFO    | cc3fbd41 | 查询文章：article_id=1
10:32:11 | INFO    | cc3fbd41 | 请求结束：/api/articles/1 200
10:32:11 | INFO    | -        | 127.0.0.1:62149 - "GET /api/articles/1 HTTP/1.1" 200

import contextvars

request_id_var: contextvars.ContextVar[str] = contextvars.ContextVar(
    "request_id", default="-"
)

@app.middleware("http")
async def request_id_middleware(request: Request, call_next):
    request_id_var.set(uuid.uuid4().hex[:8])   # 入口 set，下游全可见
    response = await call_next(request)
    response.headers["X-Request-Id"] = request_id_var.get()
    return response

@app.exception_handler(Exception)
async def unknown_exception_handler(request: Request, exc: Exception) -> JSONResponse:
    logger.opt(exception=exc).error("未处理异常：{}", request.url.path)
    return JSONResponse(status_code=500,
                        content={"code": "500000", "message": "服务器开小差了，请稍后重试"})

@app.middleware("http")
async def request_id_middleware(
    request: Request, call_next: Callable[[Request], Awaitable[Response]]
) -> Response:
    rid = uuid.uuid4().hex[:8]
    with logger.contextualize(request_id=rid):
        try:
            response = await call_next(request)
        except Exception:
            # 还在 with 作用域内：堆栈带 request_id，记完放行给兜底 handler 变 500
            logger.opt(exception=True).error("未处理异常：{} {}", request.method, request.url.path)
            raise
        response.headers["X-Request-Id"] = rid
        return response

10:32:12 | INFO    | 63732cef | 请求进来：GET /api/boom
10:32:12 | INFO    | 63732cef | [端点] 毒丸路由
10:32:12 | ERROR   | 63732cef | 未处理异常：GET /api/boom      ← 带 request_id 的完整堆栈

uvicorn.run(app, log_config=None)          # uvicorn 不自己配日志
        │
        ▼
setup_logging()                            # remove 默认 → 终端 sink + 文件 sink
        │                                   # + basicConfig 装 InterceptHandler
        ▼
request_id 中间件（contextualize / contextvar）
        │  进门发牌 → 端点/业务异常 handler 自动带牌
        │  未捕获异常穿出时：中间件 catch 记堆栈（带牌）→ raise
        ▼
Exception 兜底 handler（最外层）           # 变 500 响应；contextvar 方案下也带牌
        │
        ▼
文件 sink（rotation/retention/compression）# 带牌日志落盘，按 request_id 检索

import os
import sys

IS_PROD = os.getenv("INKHUB_ENV") == "prod"


def setup_logging() -> None:
    logger.remove()
    logger.configure(extra={"request_id": "-"})

    if IS_PROD:
        # 生产：JSON 文件（喂采集系统），INFO 起，关 diagnose
        logger.add("logs/inkhub.json", level="INFO", serialize=True,
                   rotation="100 MB", retention="30 days", compression="zip",
                   encoding="utf-8", diagnose=False, enqueue=True)
    else:
        # 本地：彩色文本终端，DEBUG 起，开着 diagnose 排障
        logger.add(sys.stderr, level="DEBUG", diagnose=True,
                   format="<green>{time:HH:mm:ss}</green> | <level>{level: <7}</level> "
                          "| {extra[request_id]} | {message}")

    logging.basicConfig(handlers=[InterceptHandler()], level=0, force=True)

def mask_secret(value: str, keep: int = 4) -> str:
    """token/密码类：只露尾巴几位。"""
    return value[:0] + "*" * max(len(value) - keep, 0) + value[-keep:]


def mask_phone(phone: str) -> str:
    """手机号：139****1234。"""
    return f"{phone[:3]}****{phone[-4:]}" if len(phone) == 11 else "***"


# ① 手里的敏感值，落笔前脱敏
logger.info("登录成功：token={}", mask_secret(access_token))
logger.info("作者手机号：{}", mask_phone(author.phone))

# ② 请求体别整包记——password 字段迟早混进去；
#    要记就挑白名单字段记
logger.info("创建文章：title={} author_id={}", body.title, body.author_id)

# ③ diagnose=True 的变量值标注是隐形泄密通道（第 8.2 节），生产必关

# 办法一：enqueue=True——写入丢给后台线程队列，多线程/多进程安全，
# 顺带把「写盘慢拖慢请求」也解了（磁盘卡顿时日志先排队）
logger.add("logs/inkhub.json", enqueue=True, serialize=True, ...)

# 办法二（更彻底）：多 worker 直接把日志发到采集系统（Loki/ELK 的 agent），
# 本地不落文件，谁也不跟谁抢

uv add loguru
uv run demo/demo_loguru.py

00:01:48 | INFO    | -         | 墨集日志系统装配完成，文件 sink 目录：.../demo/logs
00:01:48 | INFO    | req-a1b2c3 | 请求开始
00:01:48 | INFO    | req-a1b2c3 | 收到发布请求：author=陆墨 title=FastAPI 异步 SQLAlchemy 入门
00:01:48 | ERROR   | req-a1b2c3 | 同步搜索引擎失败：article_id=101
Traceback (most recent call last):
  ...
> File "demo_loguru.py", line 93, in publish_article
    sync_to_search(slug, article_id)
    │              │     └ 101
    │              └ 'fastapi-异步-sqlalchemy-入门'
TypeError: can only concatenate str (not "int") to str
00:01:48 | SUCCESS | req-a1b2c3 | 文章发布完成：article_id=101 slug=fastapi-异步-sqlalchemy-入门
00:01:48 | INFO    | req-d4e5f6 | 请求开始          ← 换了一枚牌子，行李牌各跟各的
...
00:01:48 | INFO    | - | 上一条 TRACE 被丢弃，lambda 没执行（终端没有『真的被调用了』）
00:01:48 | INFO    | - | >>> build_tracking_payload 真的被调用了（article_id=2）
00:01:48 | INFO    | - | 127.0.0.1:5173 - "POST /api/articles HTTP/1.1" 201   ← 标准 logging 被收编
00:01:48 | WARNING | - | SELECT articles WHERE id = 1 ... (0.8 ms)
00:01:48 | ERROR   | - | 计费回调验签失败：order=NO-1024
00:01:55 | INFO    | - | 演示结束，logs/ 目录产物：
00:01:55 | INFO    | - |   inkhub_20260828_000155.2026-08-28_00-01-55_562411.log.zip (940 bytes)
00:01:55 | INFO    | - |   inkhub_20260828_000155.2026-08-28_00-01-55_564721.log.zip (942 bytes)
00:01:55 | INFO    | - |   inkhub_20260828_000155.2026-08-28_00-01-55_566155.log.zip (938 bytes)
00:01:55 | INFO    | - |   inkhub_20260828_000155.log (739 bytes)