对象模型 + Java 没有的那些机制

"""阶段二 2.1 —— dataclass 就是 Java 的 record + Lombok，但选项更多。"""
from dataclasses import dataclass, field, asdict, replace, fields
import sys

# ── 最常见的四种配方 ─────────────────────────────────────────────────

@dataclass(frozen=True, slots=True)          # ≈ Java record（不可变 + 省内存）
class ModelSpec:
    name: str
    dtype: str = "bfloat16"
    params_b: float = 7.0
    tags: list[str] = field(default_factory=list)   # 可变默认值必须 factory

    def __post_init__(self) -> None:         # ≈ record 的 compact constructor
        if self.params_b <= 0:
            raise ValueError("params_b must be > 0")

    @property                                # ≈ 派生 getter
    def weight_gib(self) -> float:
        bytes_per = {"float32": 4, "bfloat16": 2, "float16": 2, "fp8": 1, "int4": 0.5}
        return self.params_b * 1e9 * bytes_per[self.dtype] / 2**30

    @classmethod                             # ≈ 静态工厂，Python 的"构造器重载"
    def from_hf_id(cls, hf_id: str, **kw) -> "ModelSpec":
        size = hf_id.rsplit("-", 1)[-1].rstrip("Bb")
        return cls(name=hf_id, params_b=float(size), **kw)


m = ModelSpec.from_hf_id("Qwen3-8B", tags=["chat"])
print("repr    ", m)                          # __repr__ 自动生成
print("weight  ", f"{m.weight_gib:.1f} GiB")
print("eq      ", m == ModelSpec.from_hf_id("Qwen3-8B", tags=["chat"]))   # 值相等

# ⚠️ frozen 不等于"深度不可变"：字段里有 list 就既不可 hash、内容也仍可改
try:
    hash(m)
except TypeError as e:
    print("hash ❌  ", e, "← tags 是 list")
m.tags.append("偷偷改了")                     # frozen 拦不住这个！
print("frozen 漏 ", m.tags)

@dataclass(frozen=True, slots=True)
class Spec2:                                  # ✅ 不可变字段用 tuple
    name: str
    tags: tuple[str, ...] = ()
print("hash ✅  ", hash(Spec2("a", ("x",))) == hash(Spec2("a", ("x",))))

# frozen ⇒ 想"改"就复制一份（≈ record 的 withers）
m2 = replace(m, dtype="fp8")
print("replace ", m2.dtype, f"{m2.weight_gib:.1f} GiB")
try:
    m.dtype = "fp8"
except Exception as e:
    print("frozen  ", type(e).__name__, e)

# 内省：字段列表、转 dict（≈ Jackson 的序列化）
print("fields  ", [f.name for f in fields(m)])
print("asdict  ", asdict(m))

# __post_init__ 做校验
try:
    ModelSpec("bad", params_b=0)
except ValueError as e:
    print("validate", e)

# ── slots 到底省多少内存？实测 ────────────────────────────────────────
@dataclass
class Fat:
    a: int; b: int; c: int
@dataclass(slots=True)
class Slim:
    a: int; b: int; c: int

def deep_size(o) -> int:
    n = sys.getsizeof(o)
    d = getattr(o, "__dict__", None)
    return n + (sys.getsizeof(d) if d is not None else 0)

print("slots   ", f"无 slots={deep_size(Fat(1,2,3))}B  有 slots={deep_size(Slim(1,2,3))}B")

# ── 排序：order=True 自动生成 __lt__ 等（≈ Comparable）────────────────
@dataclass(order=True, frozen=True)
class Job:
    priority: int
    name: str = field(compare=False)          # 不参与比较，只参与 repr/eq

jobs = [Job(3, "eval"), Job(1, "serve"), Job(2, "train")]
print("order   ", [j.name for j in sorted(jobs)])

# ── 三种"值对象"怎么选 ───────────────────────────────────────────────
from typing import NamedTuple, TypedDict
class Point(NamedTuple):      # 不可变、可解包、可当 dict key，比 dataclass 更轻
    x: float
    y: float
class Payload(TypedDict):     # 只是给 dict 加类型，运行时仍是普通 dict
    prompt: str
    max_tokens: int

p = Point(1.0, 2.0); x, y = p
print("namedtup", p, x, y, p._asdict())
d: Payload = {"prompt": "hi", "max_tokens": 8}
print("typeddic", type(d).__name__, d)

"""阶段二 2.1b —— Java 的 interface 在 Python 里有三种对应物，用哪个？"""
from abc import ABC, abstractmethod
from typing import Protocol, runtime_checkable

# ── 方式 A：Protocol —— 结构化类型（"长得像就算"）。infra 代码首选 ──────
class Tokenizer(Protocol):
    """注意：实现方 **不需要** import 这个类，更不需要 implements。"""
    def encode(self, text: str) -> list[int]: ...
    def decode(self, ids: list[int]) -> str: ...


class CharTokenizer:                       # 完全不知道 Tokenizer 存在
    def encode(self, text: str) -> list[int]:
        return [ord(c) for c in text]
    def decode(self, ids: list[int]) -> str:
        return "".join(chr(i) for i in ids)


class WordTokenizer:
    def __init__(self) -> None:
        self.vocab: dict[str, int] = {}
    def encode(self, text: str) -> list[int]:
        return [self.vocab.setdefault(w, len(self.vocab)) for w in text.split()]
    def decode(self, ids: list[int]) -> str:
        rev = {v: k for k, v in self.vocab.items()}
        return " ".join(rev[i] for i in ids)


def count_tokens(texts: list[str], tok: Tokenizer) -> dict[str, int]:
    """参数标 Protocol，mypy 会在调用点检查结构是否匹配。"""
    return {t: len(tok.encode(t)) for t in texts}


texts = ["hello world", "ai infra"]
print("char    ", count_tokens(texts, CharTokenizer()))
print("word    ", count_tokens(texts, WordTokenizer()))
print("roundtrip", CharTokenizer().decode(CharTokenizer().encode("gpu")))

# @runtime_checkable 让 isinstance 也能用（只查方法名，不查签名！）
@runtime_checkable
class Closeable(Protocol):
    def close(self) -> None: ...

class Conn:
    def close(self) -> None: ...
print("isinst  ", isinstance(Conn(), Closeable), isinstance(42, Closeable))

# ── 方式 B：ABC —— 名义类型，强制继承。需要"共享实现"时才用 ──────────
class Backend(ABC):
    @abstractmethod
    def generate(self, prompt: str) -> str: ...

    def generate_batch(self, prompts: list[str]) -> list[str]:
        """模板方法：子类只补 generate，批量版免费得到（≈ 抽象类的具体方法）。"""
        return [self.generate(p) for p in prompts]


class EchoBackend(Backend):
    def generate(self, prompt: str) -> str:
        return prompt.upper()

print("abc     ", EchoBackend().generate_batch(["a", "b"]))
try:
    Backend()                               # 抽象类不能实例化，和 Java 一样
except TypeError as e:
    print("abc     ", type(e).__name__, str(e)[:60])

# ── 方式 C：纯鸭子类型 —— 什么都不声明。脚本里可以，库里别这么干 ────
def close_all(things: list) -> None:
    for t in things:
        t.close()                           # 直到运行时才知道有没有 close

"""故意写错：签名不匹配的"实现"。mypy 会抓住，运行时不会。"""
from b_protocol import Tokenizer, count_tokens

class BrokenTokenizer:
    def encode(self, text: str) -> str:      # ← 返回 str，应该是 list[int]
        return text
    # 还漏了 decode

print(count_tokens(["hi"], BrokenTokenizer()))   # 运行时居然能跑出结果！

"""阶段二 2.3 —— 生成器：Python 相对 Java 的真实增益，也是数据管道的地基。"""
import itertools, sys, tracemalloc
from collections.abc import Iterator, Iterable

# ── 1. yield 让函数变成惰性流 ────────────────────────────────────────
def read_batches(lines: Iterable[str], size: int = 3) -> Iterator[list[str]]:
    batch: list[str] = []
    for line in lines:
        batch.append(line.strip())
        if len(batch) == size:
            yield batch                      # 交出一批，函数在这里"暂停"
            batch = []
    if batch:                                # 别忘了最后不足一批的尾巴
        yield batch

print("1.", list(read_batches([f"l{i}" for i in range(8)])))

# ── 2. 生成器是"暂停的函数"，可以亲眼看到它暂停 ──────────────────────
def noisy() -> Iterator[int]:
    print("   [gen] 开始")
    for i in range(3):
        print(f"   [gen] 准备产出 {i}")
        yield i
    print("   [gen] 结束")

g = noisy()
print("2. 创建了生成器，注意上面什么都没打印:", type(g).__name__)
print("2. next →", next(g))
print("2. next →", next(g))
print("2. 剩下的:", list(g))

# ── 3. 管道组合：每一级都是生成器，全流程内存 O(1) ───────────────────
def parse(lines: Iterable[str]) -> Iterator[dict]:
    for ln in lines:
        ep, ms = ln.split()
        yield {"endpoint": ep, "ms": float(ms)}

def slow_only(rows: Iterable[dict], threshold: float) -> Iterator[dict]:
    yield from (r for r in rows if r["ms"] > threshold)     # yield from = 转发

raw = ["/chat 812", "/healthz 1.2", "/chat 1340", "/embed 45", "/chat 3"]
pipeline = slow_only(parse(raw), 100)         # 到这里一行都还没读
print("3.", list(pipeline))

# ── 4. 内存实测：list 推导 vs 生成器 ─────────────────────────────────
N = 2_000_000
tracemalloc.start()
eager = [i * i for i in range(N)]
peak_eager = tracemalloc.get_traced_memory()[1]
del eager; tracemalloc.reset_peak()
lazy = (i * i for i in range(N))
total = sum(lazy)
peak_lazy = tracemalloc.get_traced_memory()[1]
tracemalloc.stop()
print(f"4. N={N:,}   list 推导峰值 = {peak_eager/2**20:7.1f} MiB"
      f"   |   生成器峰值 = {peak_lazy/2**10:.2f} KiB"
      f"   |   相差 {peak_eager/max(peak_lazy,1):,.0f} 倍")

# ── 5. 生成器只能消费一次 —— Java Stream 同理，但更容易忘 ────────────
g2 = (i for i in range(3))
print("5. 第一次", list(g2), "第二次", list(g2), "← 空了")

# ── 6. itertools：AI 数据管道天天用的六个 ────────────────────────────
stream = (i for i in itertools.count())                  # 无限流！
print("6. islice ", list(itertools.islice(stream, 5)))    # ≈ .limit(5)
print("6. batched", [list(b) for b in itertools.batched(range(7), 3)])  # 3.12+
print("6. chain  ", list(itertools.chain([1, 2], [3], [4, 5])))         # ≈ Stream.concat
print("6. tee    ", [list(t) for t in itertools.tee([1, 2, 3], 2)])     # 一分为二
print("6. pairwise", list(itertools.pairwise([1, 2, 3, 4])))            # 相邻两两
print("6. cycle  ", list(itertools.islice(itertools.cycle("AB"), 5)))   # 轮询调度

# ── 7. 生成器可以接收数据（很少用，但 asyncio 的底层就是它）──────────
def accumulator() -> Iterator[float]:
    total = 0.0
    while True:
        x = yield total                      # yield 既产出也接收
        total += x

acc = accumulator(); next(acc)               # 先"预热"到第一个 yield
print("7.", acc.send(10), acc.send(5), acc.send(0.5))

# ── 8. 生成器要清理资源？用 try/finally，close() 会触发它 ────────────
def with_cleanup() -> Iterator[int]:
    try:
        yield 1
        yield 2
    finally:
        print("   [gen] 清理执行了")

g3 = with_cleanup(); print("8. 取了", next(g3)); g3.close()

"""阶段二 2.4 —— 装饰器 ≈ Java 注解 + 动态代理，但纯 Python 就能写完。"""
import functools, random, time
from collections.abc import Callable
from typing import TypeVar, ParamSpec

P = ParamSpec("P"); R = TypeVar("R")

# ── 1. 最简装饰器：计时 ──────────────────────────────────────────────
def timed(fn: Callable[P, R]) -> Callable[P, R]:
    @functools.wraps(fn)                      # ★ 保留原函数的名字/docstring/签名
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
        t0 = time.perf_counter()
        try:
            return fn(*args, **kwargs)
        finally:
            print(f"   [timed] {fn.__name__} 耗时 {time.perf_counter()-t0:.4f}s")
    return wrapper

@timed
def load_model(name: str) -> str:
    """假装在加载权重。"""
    time.sleep(0.05)
    return f"{name}@loaded"

print("1.", load_model("qwen3-8b"))

# ── 2. 忘了 functools.wraps 会发生什么 ──────────────────────────────
def bad_deco(fn):
    def wrapper(*a, **k): return fn(*a, **k)
    return wrapper

@bad_deco
def documented() -> None:
    """我是文档。"""

print("2. 有 wraps:", load_model.__name__, "|", load_model.__doc__)
print("2. 无 wraps:", documented.__name__, "|", documented.__doc__, "← 元信息全丢了")

# ── 3. 带参数的装饰器 = 多一层函数（"装饰器工厂"）────────────────────
def retry(times: int = 3, backoff: float = 0.01,
          on: tuple[type[BaseException], ...] = (Exception,)):
    def deco(fn: Callable[P, R]) -> Callable[P, R]:
        @functools.wraps(fn)
        def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
            last: BaseException | None = None
            for attempt in range(1, times + 1):
                try:
                    return fn(*args, **kwargs)
                except on as e:
                    last = e
                    print(f"   [retry] 第 {attempt}/{times} 次失败: {e}")
                    if attempt < times:
                        time.sleep(backoff * 2 ** (attempt - 1))   # 指数退避
            raise RuntimeError(f"{fn.__name__} 重试 {times} 次仍失败") from last
        return wrapper
    return deco

random.seed(7)
@retry(times=4, on=(ConnectionError,))
def flaky_download(url: str) -> str:
    if random.random() < 0.6:
        raise ConnectionError("连接被重置")
    return f"{url} → 200 OK"

print("3.", flaky_download("https://hf.co/model.safetensors"))

# ── 4. 标准库自带的两个，你会天天用 ─────────────────────────────────
@functools.lru_cache(maxsize=128)             # ≈ Caffeine / Guava Cache
def tokenize_cost(text: str) -> int:
    time.sleep(0.02)                          # 假装很贵
    return len(text.split())

t0 = time.perf_counter(); tokenize_cost("a b c"); t1 = time.perf_counter()
tokenize_cost("a b c"); t2 = time.perf_counter()
print(f"4. lru_cache 首次 {t1-t0:.4f}s，命中 {t2-t1:.6f}s，{ (t1-t0)/max(t2-t1,1e-9):,.0f}× 快")
print("4. 缓存统计:", tokenize_cost.cache_info())

class Model:
    def __init__(self, path: str) -> None: self.path = path
    @functools.cached_property                # 第一次访问才算，之后存成实例属性
    def size_gib(self) -> float:
        print("   [cached_property] 真的去算了一次")
        return 14.9
m = Model("/w"); print("4.", m.size_gib, m.size_gib)   # 只算一次

# ── 5. 堆叠顺序（自下而上包裹，自上而下执行）────────────────────────
def announce(tag: str):
    def deco(fn):
        @functools.wraps(fn)
        def w(*a, **k):
            print(f"   [{tag}] enter"); r = fn(*a, **k); print(f"   [{tag}] exit"); return r
        return w
    return deco

@announce("outer")
@announce("inner")
def work() -> str:
    return "done"

print("5. 堆叠顺序 →"); print("5.", work())

svclog-pkg/
├── pyproject.toml            # 元数据 + 依赖 + ruff/mypy/pytest 配置
├── src/                      # ★ src layout：避免"根目录能 import、装完包 import 不到"
│   └── svclog/
│       ├── __init__.py       # 包门面，声明 __all__
│       ├── models.py         # 领域对象      ≈ record / VO
│       ├── reader.py         # 数据入口      ≈ Repository，全是生成器
│       ├── aggregate.py      # 业务逻辑      ≈ Service
│       ├── report.py         # 输出适配      ≈ View，按 Protocol 多态
│       └── cli.py            # 程序入口      ≈ CommandLineRunner
└── tests/                    # 阶段三补上

[project]
name = "svclog"
version = "0.2.0"
description = "推理服务访问日志聚合器"
requires-python = ">=3.12"
dependencies = []

[project.scripts]
svclog = "svclog.cli:main"          # 装完包后可以直接敲 `svclog access.csv`

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

[tool.ruff]
line-length = 100

[tool.ruff.lint]
select = ["E", "F", "I", "UP", "B", "SIM", "RUF"]   # 一个工具顶十个
# 中文注释必关：RUF001/2/3 会把全角标点当成"可疑的相似字符"
ignore = ["RUF001", "RUF002", "RUF003"]

[tool.mypy]
python_version = "3.12"
strict = true                       # 直接开最严，新项目没有历史包袱

[tool.pytest.ini_options]
pythonpath = ["src"]                 # 不用装包也能 import，本地迭代最省事
testpaths = ["tests"]
addopts = "-q --strict-markers"

"""领域对象。frozen dataclass ≈ Java record。"""

from __future__ import annotations

import math
from dataclasses import dataclass, field
from typing import Self


@dataclass(frozen=True, slots=True)
class Entry:
    """一条访问记录。frozen 让它可以安全地跨线程/进程传递。"""

    ts: float
    endpoint: str
    status: int
    ms: float

    @property
    def failed(self) -> bool:
        return self.status >= 400

    @classmethod
    def from_row(cls, row: dict[str, str]) -> Self:
        """从 CSV 的一行构造。Self 让子类继承时返回类型也跟着变（3.11+）。"""
        return cls(
            ts=float(row["ts"]),
            endpoint=row["endpoint"].strip(),
            status=int(row["status"]),
            ms=float(row["ms"]),
        )


def percentile(values: list[float], q: float) -> float:
    """线性插值分位数，口径与 numpy.percentile 一致。"""
    if not values:
        return 0.0
    ordered = sorted(values)
    if len(ordered) == 1:
        return ordered[0]
    pos = q * (len(ordered) - 1)
    lo = math.floor(pos)
    hi = min(lo + 1, len(ordered) - 1)
    return ordered[lo] + (ordered[hi] - ordered[lo]) * (pos - lo)


@dataclass(slots=True)
class Stat:
    """一个 endpoint 的聚合结果。可变，因为它是累加器。"""

    endpoint: str
    latencies: list[float] = field(default_factory=list)
    errors: int = 0
    first_ts: float = math.inf
    last_ts: float = -math.inf

    def add(self, entry: Entry) -> None:
        self.latencies.append(entry.ms)
        self.errors += int(entry.failed)
        self.first_ts = min(self.first_ts, entry.ts)
        self.last_ts = max(self.last_ts, entry.ts)

    @property
    def count(self) -> int:
        return len(self.latencies)

    @property
    def error_rate(self) -> float:
        return self.errors / self.count if self.count else 0.0

    @property
    def qps(self) -> float:
        span = self.last_ts - self.first_ts
        return self.count / span if span > 0 else float(self.count)

    @property
    def worst(self) -> float:
        return max(self.latencies, default=0.0)

    def pct(self, q: float) -> float:
        return percentile(self.latencies, q)

"""数据入口。全部返回生成器 —— 100 GB 日志也只占 O(1) 内存。"""

from __future__ import annotations

import csv
import json
import sys
from collections.abc import Iterator
from pathlib import Path

from svclog.models import Entry


def read_csv(path: Path) -> Iterator[Entry]:
    """逐行惰性产出。坏行只告警不中断 —— 日志工具必须能容忍脏数据。"""
    with path.open(newline="", encoding="utf-8") as handle:
        for lineno, row in enumerate(csv.DictReader(handle), start=2):
            try:
                yield Entry.from_row(row)
            except (KeyError, ValueError) as exc:
                print(f"{path}:{lineno}: 跳过坏行 ({exc})", file=sys.stderr)


def read_jsonl(path: Path) -> Iterator[Entry]:
    """同一个契约的另一种实现 —— 调用方一行都不用改。"""
    with path.open(encoding="utf-8") as handle:
        for lineno, line in enumerate(handle, start=1):
            line = line.strip()
            if not line:
                continue
            try:
                yield Entry.from_row(json.loads(line))
            except (KeyError, ValueError, TypeError) as exc:
                print(f"{path}:{lineno}: 跳过坏行 ({exc})", file=sys.stderr)


def open_log(path: Path) -> Iterator[Entry]:
    """按后缀分派。≈ 工厂方法。"""
    return read_jsonl(path) if path.suffix in {".jsonl", ".ndjson"} else read_csv(path)

"""业务逻辑。输入是任意 Entry 可迭代对象，不关心它从哪来。"""

from __future__ import annotations

from collections.abc import Callable, Iterable, Iterator

from svclog.models import Entry, Stat

Predicate = Callable[[Entry], bool]

_STATUS_RULES: dict[str, Predicate] = {
    "all": lambda e: True,
    "2xx": lambda e: 200 <= e.status < 300,
    "4xx": lambda e: 400 <= e.status < 500,
    "5xx": lambda e: e.status >= 500,
    "err": lambda e: e.failed,
}


def status_filter(spec: str) -> Predicate:
    try:
        return _STATUS_RULES[spec]
    except KeyError:
        raise ValueError(f"未知的 --status {spec!r}，可选：{sorted(_STATUS_RULES)}") from None


def aggregate(entries: Iterable[Entry]) -> dict[str, Stat]:
    """≈ stream().collect(groupingBy(Entry::endpoint))。"""
    stats: dict[str, Stat] = {}
    for entry in entries:
        stats.setdefault(entry.endpoint, Stat(entry.endpoint)).add(entry)
    return stats


def bucketize(entries: Iterable[Entry], width: float) -> Iterator[tuple[float, list[Entry]]]:
    """按 width 秒切桶。生成器进、生成器出，内存恒定。"""
    bucket: list[Entry] = []
    key: float | None = None
    for entry in entries:
        current = entry.ts // width * width
        if key is None:
            key = current
        elif current != key:
            yield key, bucket
            key, bucket = current, []
        bucket.append(entry)
    if key is not None and bucket:
        yield key, bucket


def rank(stats: Iterable[Stat], *, by: float = 0.95, min_count: int = 1,
         top: int = 0) -> list[Stat]:
    """按某个分位数倒序，过滤样本过少的，可选只取前 N。"""
    picked = [s for s in stats if s.count >= min_count]
    picked.sort(key=lambda s: s.pct(by), reverse=True)
    return picked[:top] if top else picked

"""输出适配层。三种渲染器满足同一个 Protocol —— 这就是 Java 的面向接口编程。"""

from __future__ import annotations

import csv
import json
import sys
from collections.abc import Iterable, Iterator, Sequence
from typing import Protocol

from svclog.models import Stat

COLUMNS: tuple[str, ...] = ("endpoint", "count", "qps", "err%", "p50", "p95", "p99", "max")


class Renderer(Protocol):
    """实现方不需要继承它，甚至不需要 import 它 —— 结构对得上就行。"""

    def render(self, stats: Sequence[Stat]) -> None: ...


def _rows(stats: Iterable[Stat]) -> Iterator[tuple[str, ...]]:
    for s in stats:
        yield (
            s.endpoint,
            f"{s.count}",
            f"{s.qps:.1f}",
            f"{s.error_rate:.1%}",
            f"{s.pct(0.50):.0f}",
            f"{s.pct(0.95):.0f}",
            f"{s.pct(0.99):.0f}",
            f"{s.worst:.0f}",
        )


class TableRenderer:
    def render(self, stats: Sequence[Stat]) -> None:
        rows = [COLUMNS, *_rows(stats)]
        widths = [max(len(r[i]) for r in rows) for i in range(len(COLUMNS))]
        for n, row in enumerate(rows):
            print("  ".join(
                cell.ljust(w) if i == 0 else cell.rjust(w)
                for i, (cell, w) in enumerate(zip(row, widths, strict=True))
            ))
            if n == 0:
                print("  ".join("-" * w for w in widths))


class CsvRenderer:
    def render(self, stats: Sequence[Stat]) -> None:
        writer = csv.writer(sys.stdout)
        writer.writerow(COLUMNS)
        writer.writerows(_rows(stats))


class JsonRenderer:
    def render(self, stats: Sequence[Stat]) -> None:
        payload = [dict(zip(COLUMNS, row, strict=True)) for row in _rows(stats)]
        json.dump(payload, sys.stdout, ensure_ascii=False, indent=2)
        print()


RENDERERS: dict[str, Renderer] = {
    "table": TableRenderer(),
    "csv": CsvRenderer(),
    "json": JsonRenderer(),
}

"""程序入口。只做参数解析和装配，逻辑一行都不放在这里。"""

from __future__ import annotations

import argparse
from pathlib import Path

from svclog.aggregate import aggregate, rank, status_filter
from svclog.reader import open_log
from svclog.report import RENDERERS


def build_parser() -> argparse.ArgumentParser:
    p = argparse.ArgumentParser(prog="svclog", description="访问日志时延报表")
    p.add_argument("path", type=Path, help="CSV 或 JSONL，字段：ts,endpoint,status,ms")
    p.add_argument("--status", default="all", help="all|2xx|4xx|5xx|err")
    p.add_argument("--min-count", type=int, default=1, help="样本数低于此值的 endpoint 丢弃")
    p.add_argument("--top", type=int, default=0, help="只看最慢的 N 个")
    p.add_argument("--by", type=float, default=0.95, help="排序用的分位数")
    p.add_argument("--format", choices=sorted(RENDERERS), default="table")
    return p


def main(argv: list[str] | None = None) -> int:
    args = build_parser().parse_args(argv)
    keep = status_filter(args.status)
    entries = (e for e in open_log(args.path) if keep(e))     # 惰性过滤
    stats = rank(aggregate(entries).values(), by=args.by,
                 min_count=args.min_count, top=args.top)
    if not stats:
        print("没有满足条件的数据")
        return 1
    RENDERERS[args.format].render(stats)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

uv run ruff check src/          # All checks passed!
uv run mypy --strict src/svclog # Success: no issues found in 6 source files
uv run python -m svclog.cli access.csv --top 3