Metadata-Version: 2.4
Name: xstools2work
Version: 0.0.3
Summary: 一个示例库
Author-email: xs <w18320724197@163.com>
License-Expression: MIT
Project-URL: Homepage, https://github.com/xs/xstools2work
Project-URL: Repository, https://github.com/xs/xstools2work
Keywords: tools,work
Classifier: Programming Language :: Python :: 3
Classifier: Operating System :: OS Independent
Requires-Python: >=3.12
Description-Content-Type: text/markdown
License-File: LICENSE
Dynamic: license-file

# xstools2work

工作常用小工具集。当前提供模块：

| 模块 | 说明 |
|---|---|
| `xstools2work.log` | 生产级结构化日志：控制台彩色输出、可轮转文件、JSON 格式、异步非阻塞 |
| `xstools2work.decorator.time` | 生产级耗时统计：`@timer` 装饰器（同步/异步通用）、`Timer` 代码块计时、可聚合的注册表、超阈值告警 |

要求 Python ≥ 3.12，无第三方依赖。

---

## xstools2work.log

### 特性

- **兼容标准库** —— `get_logger()` 返回的对象完整暴露 `logging.Logger` 接口，`.info() / .debug() / .exception()` 直接可用
- **默认值合理** —— TTY 下自动彩色控制台输出；配置了日志目录时自动生成可轮转的 `<name>.log` 与 `<name>_error.log`
- **可选非阻塞** —— `use_queue=True` 时日志经 `QueueHandler`/`QueueListener` 后台线程落盘，不阻塞业务线程
- **可选机器可读** —— `fmt="json"` 时每行输出一个 JSON 对象，适配 ELK / Loki / CloudWatch
- **适合被库使用** —— 重复配置同名 logger 不会叠加 handler；`sys.excepthook` 捕获需显式开启

### 最简用法（控制台彩色输出）

```python
from xstools2work.log import get_logger

log = get_logger(__name__)

log.debug("debug message")
log.info("info message")
log.warning("something suspicious")
try:
    1 / 0
except ZeroDivisionError:
    log.exception("处理失败")          # except 块中自动附带异常栈
    # 等价写法：log.error("处理失败", exc_info=True)
```

### 同时写文件（自动轮转 + 错误分级）

```python
log = get_logger("worker", log_dir="logs")
log.info("processing")
```

- `logs/worker.log` 记录 **DEBUG 及以上**全部内容
- `logs/worker_error.log` 只记录 **ERROR 及以上**
- 单文件默认 50 MB 轮转，保留 10 个备份（`max_bytes` / `backup_count` 可调）
- 传了 `log_dir`（或设置了 `XSTOOLS_LOG_DIR`）时，`get_logger` 默认 `target="both"`（控制台+文件），无需显式指定

### JSON 格式 + 结构化字段（接 ELK / Loki）

```python
log = get_logger("api", log_dir="logs", fmt="json")
log.info("用户登录", extra={"user_id": 7, "ip": "10.0.0.1"})
```

每行输出一个 JSON 对象，`extra` 里的字段会平铺进同一对象：

```json
{"ts": "2026-09-19T14:03:07.123+00:00", "level": "INFO", "logger": "api", "msg": "用户登录", "file": "app.py", "line": 42, "func": "main", "user_id": 7, "ip": "10.0.0.1"}
```

注意：`fmt` 只作用于**文件** handler，控制台始终是人类可读文本；`extra` 字段在纯文本模式下不会显示，只有 JSON 模式能检索到。

### 异步非阻塞（服务型程序）

```python
log = get_logger("api", log_dir="logs", use_queue=True)
```

日志经 `QueueHandler`/`QueueListener` 后台线程落盘，磁盘 I/O 不阻塞业务线程。队列默认容量 10000，写满时丢弃新记录而不阻塞。

### 完整配置（`XLog` 直接构造）

`get_logger` 是**按名称的单例** —— 同一个 name 第二次调用会原样返回缓存实例，新参数被忽略。需要多套独立配置时直接用 `XLog`：

```python
from xstools2work.log import XLog

log = XLog(
    "job.runner",
    level="DEBUG",                # Level 枚举 / int / 字符串均可
    target="both",                # "console" | "file" | "both" | "none"
    log_dir="logs",               # 普通日志目录
    err_log_dir="logs/errors",    # 错误日志单独目录（可选）
    fmt="json",                   # 文件格式："plain"（默认）| "json"
    max_bytes=100 * 1024 * 1024,  # 轮转阈值
    backup_count=5,
    use_queue=True,
    queue_size=50_000,
    capture_unhandled=True,       # 未捕获异常（含子线程）以 CRITICAL 记录
    color=None,                   # None=自动探测，True/False=强制
)
```

### 上下文管理器

```python
with XLog("worker", target="both", log_dir="logs", use_queue=True) as log:
    log.info("processing")
# 退出 with 时自动 close()：停 listener、刷盘、关文件
```

### 环境变量

| 变量 | 作用 |
|---|---|
| `XSTOOLS_LOG_DIR` | 日志文件默认目录（设置后 `target` 默认变 `"both"`） |
| `XSTOOLS_ERR_LOG_DIR` | 错误日志目录（未设置时回退到 `XSTOOLS_LOG_DIR`） |
| `XSTOOLS_LOG_LEVEL` | 默认级别（`DEBUG`/`INFO`/`WARNING`/`ERROR`/`CRITICAL`），优先级低于入参 |
| `NO_COLOR` / `FORCE_COLOR` | 禁用 / 强制控制台彩色（参见 https://no-color.org） |

```bash
XSTOOLS_LOG_DIR=./logs XSTOOLS_LOG_LEVEL=DEBUG python app.py
```

### 运行期调整与工具方法

```python
log.set_level("DEBUG")            # 运行期调级别
if log.is_enabled_for("DEBUG"):   # 规避昂贵的参数构造
    payload = build_expensive_payload()
    log.debug("detail: %s", payload)

child = log.get_child("http")     # 标准库语义的子 logger
```

### 关闭

`shutdown()` 已通过 `atexit` 自动注册，进程正常退出时会排空异步队列并关闭所有 `get_logger` 创建的 logger，**一般无需手动调用**。只有 `os._exit()` 等跳过 atexit 的场景才需要自己收尾。

### 注意事项

- **单例语义**：`get_logger("a")` 第二次传不同配置不生效，首次调用就要把参数传全
- **`propagate = False`**：XLog 创建的 logger 不会向 root 冒泡；同一进程混用标准库配置时注意别重复挂 handler（重复配置同名 logger 时 XLog 会自动摘掉旧 handler，不会叠加）
- 请求了文件输出但目录不可用时，会打印一条 stderr 提示而不是静默丢日志

---

## xstools2work.decorator.time

给函数或代码块加耗时统计，只用标准库。`async def` 直接装饰，返回值 / 异常 / 协程语义原样保留。

### 特性

- **同步 / 异步通吃** —— 同一个 `@timer` 装饰 `def` 和 `async def` 都对，内部自动选择包装器
- **异常照常传播** —— 且异常路径**依然**记录耗时与失败次数，不会因为报错就丢样本
- **可选聚合** —— 把每次耗时记进 `TimerRegistry`，随时拿到 调用次数 / 总量 / 均值 / 最值 / 失败数
- **可告警** —— 单次耗时超过 `warn_threshold`（秒）自动打一条 WARNING
- **可回调** —— `on_complete` 每次拿到一份 `TimingStats`，用于上报监控系统；回调抛错不影响业务
- **低开销、线程安全** —— 计时用单调钟 `time.perf_counter()`，注册表内部加锁；日志级别关闭时不构造字符串

### 最简用法

```python
from xstools2work.decorator.time import timer

@timer
def load_config():
    ...
```

每次调用打一行 **DEBUG** 耗时，不写入任何注册表：

```
DEBUG xstools2work.decorator.time | __main__.load_config 耗时 12.300 ms
```

> 默认 logger 名为 `xstools2work.decorator.time`，库本身不挂 handler。想让 DEBUG / INFO 可见，需要自己在应用侧配置该 logger（见下面[配合日志模块](#配合-xstools2worklog-使用)）。

带参数的写法（注意是 `@timer(...)`，多一层括号）：

```python
@timer(name="user.fetch", log_level="info", warn_threshold=0.5)
def fetch_user(uid):
    ...
```

### 聚合统计

```python
from xstools2work.decorator.time import timer, TimerRegistry

reg = TimerRegistry()

@timer(registry=reg, name="order.create", label="下单")
def create_order(uid):
    ...

create_order(1)
create_order(2)
print(reg.report())
```

```
=== 耗时统计 ===
下单: n=2 total=20.676 ms mean=10.338 ms min=10.268 ms max=10.408 ms fail=0
```

`reg` 常用方法：

| 方法 | 返回 | 用途 |
|---|---|---|
| `report(key="total", reverse=True)` | `str` | 打印用的多行统计表 |
| `snapshot()` | `dict[str, dict]` | 纯字典副本，可安全 JSON 序列化 / 跨线程用 |
| `top(n=5, key="total")` | `list[StatAggregate]` | 最"重"的前 N 个；`key` 可用 `total` / `mean` / `max` / `count` / `last` / `failures` |
| `get(name)` | `StatAggregate \| None` | 单个名字的聚合对象 |
| `reset(name=None)` | — | 清零（不传名字则全清） |
| `names()` / `len(reg)` / `"x" in reg` | — | 查看都有哪些名字被统计了 |

模块级还提供一个进程共享的 `default_registry`，开箱即用：

```python
from xstools2work.decorator.time import timer, default_registry

@timer(registry=default_registry)          # 各模块独立装饰，最后在出口统一看
def process():
    ...

print(default_registry.report())           # 程序退出前 / 定时任务里调用
```

### 手动测量任意代码块

不方便加装饰器时（循环体里的一段、第三方调用、临时排查）用 `Timer`：

```python
from xstools2work.decorator.time import Timer, default_registry

with Timer("db.query", registry=default_registry) as t:
    rows = db.execute(sql)

print(t.elapsed, t.succeeded, t.exception)   # 秒 / 是否成功 / 异常对象
```

- `with` 内抛异常时 `elapsed` 仍然被赋值，异常**照常向外传播**（`Timer` 绝不吞异常）
- 同一实例可复用，每次进入 `with` 会重置状态

### 超阈值告警

```python
@timer(warn_threshold=0.5, name="http.call")   # 单位：秒
def call_upstream():
    ...
```

单次超过 0.5s 时打 WARNING，**同时**仍然打常规计时日志、仍然计入注册表：

```
WARNING | __main__.http.call 耗时 801.234 ms，超过阈值 500.000 ms
```

### 回调上报（对接监控系统）

```python
from xstools2work.decorator.time import timer

def push(stats):                    # stats 是 TimingStats
    if stats.elapsed > 0.5:
        monitor.incr("slow", tags={"op": stats.name})

@timer(on_complete=push, name="checkout")
def checkout(order):
    ...
```

`TimingStats` 字段：`name` / `label` / `elapsed`（秒）/ `started_at`（墙钟 epoch 秒）/ `succeeded` / `exception`，`as_dict()` 可直接序列化。回调里抛异常只会被记一条 WARNING，不会影响被装饰函数的返回值。

### 异步函数

```python
@timer(registry=default_registry, name="fetch.all")
async def fetch_all(urls):
    async with session.get(urls[0]) as resp:
        return await resp.json()
```

装饰后仍是协程函数，`asyncio.iscoroutinefunction(fetch_all)` 为 `True`，`await` / `asyncio.run()` / `gather()` 都照常工作。

### 配合 xstools2work.log 使用

`logger` 参数接受任何带 `.log()` 的对象，直接传 `XLog` 即可让计时日志走统一格式与文件：

```python
from xstools2work.log import get_logger
from xstools2work.decorator.time import timer, default_registry

log = get_logger("perf", level="INFO", log_dir="logs")

@timer(logger=log, log_level="info", warn_threshold=0.5, registry=default_registry)
def heavy_job():
    ...
```

### 全部参数

| 参数 | 默认 | 说明 |
|---|---|---|
| `name` | 函数 `__qualname__` | 注册表键（聚合按它分组） |
| `label` | `模块.qualname` | 日志 / 报表里的展示名 |
| `logger` | `logging.getLogger("xstools2work.decorator.time")` | 目标 logger |
| `log_level` | `logging.DEBUG` | 常规耗时日志级别，整数或名字（`"info"` 等） |
| `warn_threshold` | `None` | 超过该**秒**数打一条 WARNING |
| `registry` | `None` | 聚合器；传 `default_registry` 或自建 `TimerRegistry()` |
| `on_complete` | `None` | 每次调用结束后回调，参数为 `TimingStats` |
| `log_args` / `log_result` | `False` | 把入参 / 返回值的 `repr`（截断到 200 字符）放进 `TimingStats` |
| `enabled` | `True` | `False` 时零开销，原样返回原函数 |

### 时长格式化

`humanize_duration(seconds, *, unit=None)` 可单独使用，自动选单位：

```python
>>> humanize_duration(0.000123)
'123.00 μs'
>>> humanize_duration(0.45)
'450.000 ms'
>>> humanize_duration(2.5)
'2.500 s'
>>> humanize_duration(92, unit="s")      # 强制单位：us / ms / s / min / h
'92.000 s'
```

### 注意事项

- **默认级别是 DEBUG 且库不挂 handler**：应用侧没配置 `xstools2work.decorator.time` 这个 logger 时，常规耗时日志会被静默丢弃（只有 `warn_threshold` / 异常路径的 WARNING 及以上会经 root 落到 stderr）。生产上想看到耗时，要么 `log_level="info"` 并配好 logger，要么直接 `registry=` 看报表
- **`log_args` / `log_result` 只在有 `on_complete` 时生效**：它们只往 `TimingStats` 里塞字段，不影响日志内容，也不配 `on_complete` 时等于没传
- **传了 `name` 之后 `label` 会变成 `模块.name`**（例如 `name="user.fetch"` → 日志里显示 `__main__.user.fetch`）。想让报表干净就 `name` 和 `label` 一起传
- **生成器函数只计"创建生成器"的时间**，不含迭代过程（装饰后 `inspect.isgeneratorfunction()` 还会变成 `False`）；要测迭代请在 `for` 循环外再用 `Timer` 包一层
- **`name` 别用动态值**：注册表按名字累积条目且不限容量，用 `f"op.{user_id}"` 这类高基数键会让 `default_registry` 一直涨
- **`registry` 有锁，`report()` 可以随便调**：但热路径上每次调用多一次加锁，极高频函数（每秒十万次级）建议只在关心的层挂 registry
