把工程素养接回来

def test_shows_what_pytest_gives_you_for_free():
    stats = aggregate([Entry(1.0, "/chat", 200, 800.0), Entry(2.0, "/chat", 500, 15.0)])
    # 裸 assert —— 失败时 pytest 会自动把两边展开给你看
    assert {k: v.count for k, v in stats.items()} == {"/chat": 3}

"""共享夹具。conftest.py 里的 fixture 自动对同目录及子目录的测试可见 —— 不用 import。"""

import pytest

from svclog.models import Entry


@pytest.fixture
def entries() -> list[Entry]:
    """一份小而确定的数据集。故意让 /chat 有一条 500。"""
    return [
        Entry(ts=100.0, endpoint="/chat", status=200, ms=800.0),
        Entry(ts=100.5, endpoint="/chat", status=200, ms=1200.0),
        Entry(ts=101.0, endpoint="/chat", status=500, ms=15.0),
        Entry(ts=101.2, endpoint="/healthz", status=200, ms=1.0),
        Entry(ts=102.0, endpoint="/healthz", status=200, ms=2.0),
    ]


@pytest.fixture
def csv_file(tmp_path, entries):
    """tmp_path 是 pytest 内置夹具：每个测试一个干净的临时目录，自动清理。

    注意 csv_file 依赖 entries —— fixture 可以依赖 fixture，pytest 自动按图求值。
    JUnit 的 @BeforeEach 做不到这件事。
    """
    p = tmp_path / "access.csv"
    lines = ["ts,endpoint,status,ms"]
    lines += [f"{e.ts},{e.endpoint},{e.status},{e.ms}" for e in entries]
    p.write_text("\n".join(lines) + "\n", encoding="utf-8")
    return p

"""领域对象的测试。注意：函数名以 test_ 开头即可，不需要 @Test。"""

import math

import pytest

from svclog.models import Entry, Stat, percentile


def test_entry_is_immutable():
    e = Entry(ts=1.0, endpoint="/a", status=200, ms=5.0)
    with pytest.raises(AttributeError):        # ≈ assertThrows
        e.status = 500                          # type: ignore[misc]


def test_from_row_strips_and_converts():
    e = Entry.from_row({"ts": "1.5", "endpoint": "  /a  ", "status": "429", "ms": "3"})
    assert (e.ts, e.endpoint, e.status, e.ms) == (1.5, "/a", 429, 3.0)
    assert e.failed is True


@pytest.mark.parametrize(                       # ≈ @ParameterizedTest + @CsvSource
    ("raw", "exc"),
    [
        ({"ts": "x", "endpoint": "/a", "status": "200", "ms": "1"}, ValueError),
        ({"ts": "1", "endpoint": "/a", "status": "200"}, KeyError),
    ],
)
def test_from_row_rejects_bad_input(raw, exc):
    with pytest.raises(exc):
        Entry.from_row(raw)


@pytest.mark.parametrize(
    ("q", "expected"),
    [(0.0, 1.0), (0.5, 5.5), (0.95, 9.55), (1.0, 10.0)],
)
def test_percentile_matches_numpy_convention(q, expected):
    """线性插值口径应与 numpy.percentile 一致。"""
    assert percentile([float(i) for i in range(1, 11)], q) == pytest.approx(expected)


def test_percentile_edge_cases():
    assert percentile([], 0.5) == 0.0
    assert percentile([42.0], 0.99) == 42.0


def test_stat_accumulates(entries):
    s = Stat("/chat")
    for e in entries[:3]:
        s.add(e)
    assert s.count == 3
    assert s.errors == 1
    assert s.error_rate == pytest.approx(1 / 3)
    assert s.worst == 1200.0
    assert s.qps == pytest.approx(3 / 1.0)      # 3 条落在 1 秒的跨度里


def test_empty_stat_does_not_divide_by_zero():
    s = Stat("/x")
    assert s.error_rate == 0.0 and s.qps == 0.0 and s.worst == 0.0
    assert math.isinf(s.first_ts)

"""读取 → 聚合 → 渲染 的集成测试。"""

import pytest

from svclog.aggregate import aggregate, bucketize, rank, status_filter
from svclog.cli import main
from svclog.reader import open_log


def test_read_csv_roundtrip(csv_file, entries):
    assert list(open_log(csv_file)) == entries      # frozen dataclass 的 __eq__ 是值相等


def test_reader_skips_bad_lines(tmp_path, capsys):
    p = tmp_path / "dirty.csv"
    p.write_text("ts,endpoint,status,ms\n1,/a,200,5\nBAD,/a,200,5\n3,/b,500,9\n")
    got = list(open_log(p))
    assert [e.endpoint for e in got] == ["/a", "/b"]     # 坏行被跳过，不中断
    assert "跳过坏行" in capsys.readouterr().err          # 但要有告警


def test_reader_is_lazy(csv_file):
    """★ 测性质：生成器必须是惰性的 —— 拿到对象时不该读任何数据。"""
    import inspect

    gen = open_log(csv_file)
    assert inspect.getgeneratorstate(gen) == "GEN_CREATED"
    next(gen)
    assert inspect.getgeneratorstate(gen) == "GEN_SUSPENDED"


def test_aggregate_groups_by_endpoint(entries):
    stats = aggregate(entries)
    assert set(stats) == {"/chat", "/healthz"}
    assert stats["/chat"].count == 3
    assert stats["/healthz"].errors == 0


@pytest.mark.parametrize(
    ("spec", "expected"), [("all", 5), ("2xx", 4), ("5xx", 1), ("err", 1), ("4xx", 0)]
)
def test_status_filter(entries, spec, expected):
    keep = status_filter(spec)
    assert sum(1 for e in entries if keep(e)) == expected


def test_status_filter_rejects_unknown():
    with pytest.raises(ValueError, match="未知的 --status"):
        status_filter("7xx")


def test_bucketize_splits_by_time(entries):
    buckets = list(bucketize(entries, width=1.0))
    assert [(k, len(v)) for k, v in buckets] == [(100.0, 2), (101.0, 2), (102.0, 1)]


def test_rank_sorts_by_p95_desc(entries):
    ordered = rank(aggregate(entries).values())
    assert [s.endpoint for s in ordered] == ["/chat", "/healthz"]


def test_cli_table_output(csv_file, capsys):
    assert main([str(csv_file)]) == 0
    out = capsys.readouterr().out
    assert "endpoint" in out and "/chat" in out


def test_cli_json_output(csv_file, capsys):
    import json

    assert main([str(csv_file), "--format", "json"]) == 0
    payload = json.loads(capsys.readouterr().out)
    assert payload[0]["endpoint"] == "/chat"


def test_cli_returns_1_when_nothing_matches(csv_file, capsys):
    assert main([str(csv_file), "--min-count", "999"]) == 1
    assert "没有满足条件的数据" in capsys.readouterr().out


def test_cli_does_not_load_whole_file(csv_file, monkeypatch):
    """★ monkeypatch ≈ Mockito：证明我们走的是惰性路径，没有 read() 整个文件。"""
    import pathlib

    original = pathlib.Path.read_text
    calls = []

    def spy(self, *a, **k):
        calls.append(self)
        return original(self, *a, **k)

    monkeypatch.setattr(pathlib.Path, "read_text", spy)
    main([str(csv_file)])
    assert calls == []

[project]
name = "svclog"
version = "0.2.0"
requires-python = ">=3.12"
dependencies = ["httpx>=0.27"]        # 运行时依赖

[dependency-groups]                    # 开发依赖，不会被下游装到
dev = ["pytest>=8", "ruff>=0.7", "mypy>=1.13"]

[project.scripts]
svclog = "svclog.cli:main"             # 装完包直接敲 `svclog`

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

# ── 为什么从 CUDA 镜像起，而不是 python:slim ──────────────────────────
# 接真 vLLM 时需要 CUDA 运行时和驱动兼容的用户态库。自己往 python:slim 里装
# 是一条不归路。只跑 CPU 的话可以换成 python:3.12-slim，其余照抄。
FROM nvidia/cuda:12.4.1-runtime-ubuntu22.04 AS base

ENV PYTHONUNBUFFERED=1 \
    PYTHONDONTWRITEBYTECODE=1 \
    UV_COMPILE_BYTECODE=1 \
    UV_LINK_MODE=copy \
    HF_HOME=/models          # ← 权重放挂载卷，**绝不**打进镜像

COPY --from=ghcr.io/astral-sh/uv:0.5 /uv /usr/local/bin/uv
RUN apt-get update && apt-get install -y --no-install-recommends python3.12 curl \
    && rm -rf /var/lib/apt/lists/*

WORKDIR /app

# ★ 依赖层和代码层分开 COPY：改一行代码不会让依赖重新装一遍。
#   这和你在 Java 里先 COPY pom.xml 再 mvn dependency:go-offline 是同一个技巧。
COPY pyproject.toml uv.lock ./
RUN uv sync --frozen --no-dev --no-install-project

COPY src/ src/
RUN uv sync --frozen --no-dev

EXPOSE 8000

# liveness 用 /healthz（只证明进程活着）。
# ⚠️ 不要在这里用 /readyz：模型加载要几分钟，会被判定失败然后无限重启。
HEALTHCHECK --interval=10s --timeout=2s --start-period=5s --retries=3 \
    CMD curl -fsS http://localhost:8000/healthz || exit 1

# 单进程 + 多副本，不要用 --workers：
# 每个 worker 是独立进程，会各自去抢显存，也不共享准入控制的计数器。
# GPU 服务的正确扩展方式是"一卡一进程，前面放负载均衡"。
CMD ["uv", "run", "uvicorn", "llmgw.app:app", "--host", "0.0.0.0", "--port", "8000"]

"""阶段三 3.3 —— 用一个可复现的实验，把 GIL 给你的直觉校准过来。

结论先行（这三条推翻了你全部的 Java 并发直觉）：
    1. CPU 密集的**纯 Python** 代码：多线程不会加速（GIL），多进程才会。
    2. IO 密集的代码：多线程/asyncio 大幅加速（等待时 GIL 是释放的）。
    3. 数值计算：交给 NumPy/PyTorch —— 它们在算的时候**释放 GIL**，
       这才是 AI Infra 的正解，而不是纠结开几个线程。

脚本会先探测这台机器**实际可用**的并行度再跑，所以在容器里跑出来的数字也是诚实的。
"""
from __future__ import annotations

import os
import sys
import time
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor

N = 10_000_000
WORKERS = 4


def cpu_task(n: int = N) -> int:
    """纯 Python 的 CPU 密集活儿：一个热循环。"""
    total = 0
    for i in range(n):
        total += i * i % 7
    return total


def io_task(seconds: float = 0.25) -> float:
    """IO 密集的活儿：等待。sleep 期间 GIL 被释放。"""
    time.sleep(seconds)
    return seconds


def timeit(fn) -> float:
    t0 = time.perf_counter()
    fn()
    return time.perf_counter() - t0


def row(label: str, dt: float, baseline: float | None = None) -> None:
    tail = "" if baseline is None else f"   加速比 {baseline / dt:5.2f}×"
    print(f"  {label:<36} {dt:6.2f}s{tail}")


def probe_parallelism() -> float:
    """先量一把：这台机器真的能同时跑几个进程？

    os.cpu_count() 在容器里会撒谎 —— 它报的是宿主机核数，
    而 cgroup 的 cpu.max 配额才是你真正拿到的。所以直接实测。
    """
    one = timeit(lambda: list(ProcessPoolExecutor(1).map(cpu_task, [N // 4])))
    two = timeit(lambda: list(ProcessPoolExecutor(2).map(cpu_task, [N // 4] * 2)))
    return 2 * one / two          # 理想 2.0（真两核），1.0 表示只有一核可用


def main() -> None:
    print(f"机器：os.cpu_count()={os.cpu_count()} · Python {sys.version.split()[0]}")
    eff = probe_parallelism()
    print(f"实测有效并行度：{eff:.2f}（2.0=真有两核可并行，1.0=被限流到单核）\n")

    print(f"① CPU 密集：{WORKERS} 份 {N:,} 次的纯 Python 循环")
    serial = timeit(lambda: [cpu_task() for _ in range(WORKERS)])
    row("串行 for 循环", serial)
    row(f"ThreadPool({WORKERS}) ← 你的 Java 直觉",
        timeit(lambda: list(ThreadPoolExecutor(WORKERS).map(cpu_task, [N] * WORKERS))), serial)
    row(f"ProcessPool({WORKERS}) ← 正确答案",
        timeit(lambda: list(ProcessPoolExecutor(WORKERS).map(cpu_task, [N] * WORKERS))), serial)
    print("  → 线程池的加速比 ≈ 1.00（甚至 <1）：GIL 让它们排队，还多付了切换成本。")
    print(f"  → 进程池的加速比上限 = 实测并行度 {eff:.2f}，再减去进程启动与序列化开销。\n")

    print(f"② IO 密集：{WORKERS} 次 0.25s 的等待")
    serial_io = timeit(lambda: [io_task() for _ in range(WORKERS)])
    row("串行", serial_io)
    row(f"ThreadPool({WORKERS})",
        timeit(lambda: list(ThreadPoolExecutor(WORKERS).map(io_task, [0.25] * WORKERS))), serial_io)
    print("  → 同样是线程，这里就有效：等待时 GIL 被释放了。\n")

    print("③ 同一笔计算交给 NumPy（底层 C，算的时候释放 GIL）")
    try:
        import numpy as np
    except ImportError:
        print("  （未装 numpy，跳过）")
        return
    dt = timeit(lambda: int(((np.arange(N, dtype=np.int64) ** 2) % 7).sum()))
    checksum = int(((np.arange(N, dtype=np.int64) ** 2) % 7).sum())
    row("NumPy 向量化（一份，不是四份）", dt)
    print(f"  → 比一份纯 Python 循环快 {serial / WORKERS / dt:.1f}×，"
          f"结果一致（{checksum} == {cpu_task(N)}）。")
    print("  → 记住这个比例：优化 Python 循环最多快几倍，换成向量化是一个数量级。")


if __name__ == "__main__":
    main()

"""阶段三 3.3b —— asyncio：心智模型 ≈ Reactor / CompletableFuture，但**单线程**。

重点演示第 2 节：在 async 函数里做阻塞调用，会把整个事件循环卡死。
这是 Java 人转 Python 写服务时最常见、也最贵的一次事故。
"""
from __future__ import annotations

import asyncio
import time


async def fetch(name: str, delay: float) -> str:
    """模拟一次网络调用。await 会把控制权交还事件循环。"""
    await asyncio.sleep(delay)
    return f"{name}({delay}s)"


# ── 1. gather：并发发起，不是并行计算 ────────────────────────────────
async def demo_gather() -> None:
    t0 = time.perf_counter()
    results = await asyncio.gather(fetch("a", 0.3), fetch("b", 0.3), fetch("c", 0.3))
    print(f"1. gather  {results}  耗时 {time.perf_counter()-t0:.2f}s（不是 0.9s）")


# ── 2. ★ 阻塞调用毒死事件循环 ────────────────────────────────────────
def blocking_call(seconds: float) -> str:
    """任何同步库都长这样：requests、psycopg2(同步)、time.sleep、model.generate()。"""
    time.sleep(seconds)                     # ← 注意：不是 await asyncio.sleep
    return "blocked"


async def demo_blocking() -> None:
    async def heartbeat(stop: asyncio.Event) -> int:
        """每 50ms 跳一次。事件循环健康的话，1 秒该跳 ~20 次。"""
        beats = 0
        while not stop.is_set():
            await asyncio.sleep(0.05)
            beats += 1
        return beats

    for label, work in (
        ("❌ 直接调用阻塞函数", lambda: blocking_call(1.0)),
        ("✅ asyncio.to_thread 包一层", None),
    ):
        stop = asyncio.Event()
        hb = asyncio.create_task(heartbeat(stop))
        await asyncio.sleep(0.05)
        t0 = time.perf_counter()
        if work is None:
            await asyncio.to_thread(blocking_call, 1.0)   # 丢到线程池，循环不受影响
        else:
            work()                                        # 循环在这 1 秒里彻底停摆
        dt = time.perf_counter() - t0
        stop.set()
        beats = await hb
        print(f"2. {label:<24} 耗时{dt:.2f}s  这期间心跳只跳了 {beats:>2} 次"
              f"（健康值 ~20）")


# ── 3. 超时与取消：客户端断开时必须真的停下游 ────────────────────────
async def demo_cancel() -> None:
    cleaned = []

    async def long_generation() -> str:
        try:
            await asyncio.sleep(10)          # 假装在 GPU 上生成 2000 个 token
            return "done"
        except asyncio.CancelledError:
            cleaned.append("释放了 KV cache")   # ← 推理网关里这行值很多钱
            raise

    try:
        async with asyncio.timeout(0.2):     # 3.11+，比 wait_for 好用
            await long_generation()
    except TimeoutError:
        print(f"3. 超时被取消，清理动作执行了：{cleaned}")


# ── 4. 限流：Semaphore ≈ Java 的舱壁 / 连接池上限 ────────────────────
async def demo_semaphore() -> None:
    sem = asyncio.Semaphore(2)               # 同时最多 2 个在跑
    running, peak = 0, 0

    async def guarded(i: int) -> None:
        nonlocal running, peak
        async with sem:
            running += 1
            peak = max(peak, running)
            await asyncio.sleep(0.1)
            running -= 1

    t0 = time.perf_counter()
    await asyncio.gather(*(guarded(i) for i in range(6)))
    print(f"4. 6 个任务 / 并发上限 2 → 实际峰值并发 {peak}，"
          f"耗时 {time.perf_counter()-t0:.2f}s（=3 批 × 0.1s）")


# ── 5. 常见错误：忘了 await，什么也不会发生 ─────────────────────────
async def demo_forgot_await() -> None:
    coro = fetch("x", 0.01)
    print(f"5. 忘了 await 的话拿到的是：{type(coro).__name__}，不是结果")
    print(f"5. await 之后才是：{await coro}")


async def main() -> None:
    await demo_gather()
    await demo_blocking()
    await demo_cancel()
    await demo_semaphore()
    await demo_forgot_await()


if __name__ == "__main__":
    asyncio.run(main())

"""阶段三 3.4 —— 先剖析，再优化。四个工具，各解决一个问题。"""
from __future__ import annotations

import cProfile
import io
import pstats
import random
import timeit

random.seed(0)
DATA = [random.gauss(500, 200) for _ in range(200_000)]


# ── 一段有意写慢的代码：你能一眼看出问题吗？ ──────────────────────────
def p95_slow(values: list[float]) -> float:
    ordered = sorted(values)                      # 每次调用都重排一次
    idx = int(len(ordered) * 0.95)
    return ordered[idx]


def summarize_slow(values: list[float]) -> dict[str, float]:
    return {
        "mean": sum(values) / len(values),
        "p50": p95_slow(values) * 0,              # 装傻：又排一次
        "p95": p95_slow(values),                  # 再排一次
        "p99": sorted(values)[int(len(values) * 0.99)],   # 第三次
    }


def summarize_fast(values: list[float]) -> dict[str, float]:
    ordered = sorted(values)                      # 只排一次，复用
    n = len(ordered)
    return {
        "mean": sum(ordered) / n,
        "p50": ordered[n // 2],
        "p95": ordered[int(n * 0.95)],
        "p99": ordered[int(n * 0.99)],
    }


# ── 1. cProfile：函数级，先定位"时间花在谁身上" ──────────────────────
print("① cProfile —— 谁吃掉了时间")
buf = io.StringIO()
cProfile.run("summarize_slow(DATA)", "/tmp/prof.out")
stats = pstats.Stats("/tmp/prof.out", stream=buf).sort_stats("cumulative")
stats.print_stats(6)
print("\n".join(buf.getvalue().splitlines()[3:13]))

# ── 2. timeit：微基准，别用 time.time() 手测 ────────────────────────
print("\n② timeit —— 改完到底快了多少")
slow = timeit.timeit(lambda: summarize_slow(DATA), number=5) / 5
fast = timeit.timeit(lambda: summarize_fast(DATA), number=5) / 5
print(f"  summarize_slow  {slow*1000:7.1f} ms")
print(f"  summarize_fast  {fast*1000:7.1f} ms   快 {slow/fast:.1f}×（只是别重复排序）")

# ── 3. 再往下一层：同一件事交给 NumPy ───────────────────────────────
import numpy as np

arr = np.asarray(DATA)
npy = timeit.timeit(lambda: np.percentile(arr, [50, 95, 99]), number=5) / 5
print(f"  numpy.percentile{npy*1000:7.1f} ms   比 fast 再快 {fast/npy:.1f}×")
print("  → 这就是那条界线：算法层面能省几倍，换成向量化能省一个数量级。")

# ── 4. tracemalloc：内存去哪了 ──────────────────────────────────────
print("\n③ tracemalloc —— 内存去哪了")
import tracemalloc

tracemalloc.start()
snap_before = tracemalloc.take_snapshot()
waste = [list(DATA) for _ in range(5)]            # 故意复制 5 份
snap_after = tracemalloc.take_snapshot()
for stat in snap_after.compare_to(snap_before, "lineno")[:2]:
    print(f"  {stat}")
tracemalloc.stop()
del waste

py-spy dump   --pid 1234              # 立刻打出所有线程的栈：卡在哪一行
py-spy top    --pid 1234              # 像 top 一样实时看函数耗时
py-spy record --pid 1234 -o f.svg -d 30   # 30 秒火焰图
py-spy dump   --pid 1234 --locals     # 连局部变量一起打