把 Java 翻译成 Python

# 装 uv（一次性）
curl -LsSf https://astral.sh/uv/install.sh | sh

# 建项目
uv init myproj && cd myproj     # 生成 pyproject.toml，≈ mvn archetype:generate
uv python install 3.12          # 装解释器，≈ sdk install java
uv add numpy fastapi            # 加依赖，自动建 .venv 并写 uv.lock
uv add --dev pytest ruff mypy   # 开发依赖，≈ <scope>test</scope>
uv run python script.py         # 在项目环境里跑，不用手动 activate
uv sync --frozen                # 按 lock 精确复现环境，≈ mvn -o

"""阶段一 1.2 —— Java 习惯写法的 Python 对照，全部可运行。"""

# ── 1. null / Optional ────────────────────────────────────────────────
# Java: if (name != null && !name.isEmpty()) {...}
name: str | None = None
print("1.", "空" if not name else name)       # None、""、[]、0 都是假值
# Java: Optional.ofNullable(name).orElse("default")
print("1.", name or "default")                # or 返回第一个真值
# ⚠️ 只有判断"是不是 None"时才用 is
print("1.", name is None, 0 or "fallback")    # 注意 0 也是假值，这是头号坑

# ── 2. 字符串格式化：f-string 一招吃遍 ────────────────────────────────
who, score = "alice", 93.456
print("2.", f"{who} scored {score:.2f}")                    # %.2f
print("2.", f"|{who:>10}|{score:<8.1f}|{1234567:,}|{0.873:.1%}|")
print("2.", f"{score=}")                                    # 调试神器，打印表达式本身

# ── 3. switch → match/case（3.10+，比 switch 强）────────────────────
def classify(code: int) -> str:
    match code:
        case 200 | 201 | 204:
            return "ok"
        case c if 400 <= c < 500:
            return "client-error"
        case c if c >= 500:
            return "server-error"
        case _:
            return "other"

print("3.", classify(200), classify(429), classify(503), classify(302))

# ── 4. 遍历：几乎永远不用下标 ─────────────────────────────────────────
hosts = ["gpu-01", "gpu-02", "gpu-03"]
loads = [0.91, 0.42, 0.77]
for i, h in enumerate(hosts, start=1):        # Java: for (int i=0; ...)
    print("4.", i, h)
for h, l in zip(hosts, loads):                # Java: 手写下标并行遍历
    print("4.", f"{h} load={l}")

# ── 5. 切片：Java 没有的负索引 ────────────────────────────────────────
xs = [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
print("5.", xs[2:5], xs[-3:], xs[::2], xs[::-1])
print("5.", "gpu-01.cluster.local".split(".")[0])

# ── 6. 多返回值：元组解包 ─────────────────────────────────────────────
def min_max(vs: list[float]) -> tuple[float, float]:
    return min(vs), max(vs)                   # Java 要造个 Pair 类

lo, hi = min_max(loads)
print("6.", lo, hi)
first, *rest = hosts                          # 星号解包，Java 没有
print("6.", first, rest)

# ── 7. 异常：全是 unchecked，但有 Java 没有的 else ───────────────────
def parse_port(raw: str) -> int:
    try:
        port = int(raw)
    except ValueError as e:                   # Java: catch (E e)
        raise ValueError(f"bad port: {raw!r}") from e   # from = initCause
    else:                                     # 没抛异常才走这里（Java 没有）
        return port
    finally:
        pass                                  # 和 Java 一样

print("7.", parse_port("8080"))
try:
    parse_port("八千")
except ValueError as e:
    print("7.", type(e).__name__, e, "| cause:", type(e.__cause__).__name__)

# ── 8. with ≈ try-with-resources，但更泛化 ──────────────────────────
import contextlib, io, time

@contextlib.contextmanager                    # 三行就能造一个"资源"
def timed(label: str):
    t0 = time.perf_counter()
    yield
    print("8.", f"{label} took {time.perf_counter()-t0:.4f}s")

with timed("sum"):
    total = sum(range(1_000_000))
print("8.", total)

buf = io.StringIO()
with contextlib.redirect_stdout(buf):         # with 可以做的事远超关文件
    print("这句被重定向了")
print("8.", "captured:", buf.getvalue().strip())

# ── 9. 真值陷阱：Java 人踩得最狠的一个 ───────────────────────────────
def page_size(n: int | None) -> int:
    return n or 100          # ❌ n=0 时会返回 100
def page_size_ok(n: int | None) -> int:
    return 100 if n is None else n            # ✅

print("9.", page_size(0), page_size_ok(0))

# ── 10. main：脚本与库两用 ───────────────────────────────────────────
if __name__ == "__main__":
    print("10.", "作为脚本直接跑时才执行")

"""阶段一 1.3 —— Java Stream API 的逐条 Python 对照，可运行。"""
from collections import Counter, defaultdict, deque
from dataclasses import dataclass
import functools, itertools, sys

@dataclass
class Req:
    endpoint: str
    status: int
    ms: float

reqs = [
    Req("/v1/chat", 200, 812.0), Req("/v1/chat", 200, 1340.5),
    Req("/v1/chat", 429, 3.1),   Req("/healthz", 200, 1.2),
    Req("/v1/embed", 200, 44.8), Req("/v1/embed", 500, 120.0),
    Req("/healthz", 200, 0.9),   Req("/v1/chat", 200, 655.0),
]

# .map(f)                     → 列表推导
print("map      ", [r.ms for r in reqs][:4])
# .filter(p)                  → 带 if 的推导
print("filter   ", [r.endpoint for r in reqs if r.status >= 400])
# .map().filter()             → 一行搞定，不用链式
print("map+filter", [round(r.ms) for r in reqs if r.ms > 100])
# .collect(toSet())
print("toSet    ", sorted({r.endpoint for r in reqs}))
# .collect(toMap(...))        → 字典推导
print("toMap    ", {r.endpoint: r.status for r in reqs})
# .anyMatch / .allMatch / .noneMatch
print("match    ", any(r.status >= 500 for r in reqs), all(r.ms > 0 for r in reqs))
# .count()
print("count    ", sum(1 for r in reqs if r.status == 200))
# .sorted(Comparator.comparing(...).reversed())
print("sorted   ", [r.endpoint for r in sorted(reqs, key=lambda r: r.ms, reverse=True)][:3])
# .min / .max
print("max      ", max(reqs, key=lambda r: r.ms).ms)
# .reduce
print("reduce   ", functools.reduce(lambda a, r: a + r.ms, reqs, 0.0))
#   ↑ 但 Python 里更地道的是直接 sum(...)
print("sum      ", sum(r.ms for r in reqs))
# .flatMap                    → 双重推导（先外后内，顺序和 Java 反直觉）
tags = [["gpu", "a100"], ["cpu"], ["gpu", "h100"]]
print("flatMap  ", [t for group in tags for t in group])
# .distinct()
print("distinct ", list(dict.fromkeys(t for g in tags for t in g)))  # 保序去重
# .limit(n) —— 对惰性流用 islice
print("limit    ", list(itertools.islice((r.ms for r in reqs), 3)))

# .collect(groupingBy(...))  → 两种写法
by_ep: dict[str, list[Req]] = defaultdict(list)
for r in reqs:
    by_ep[r.endpoint].append(r)
print("groupBy  ", {k: len(v) for k, v in by_ep.items()})

# itertools.groupby 是"相邻分组"，必须先排序 —— Java 人最容易踩的一个坑
key = lambda r: r.endpoint
print("groupby  ", {k: len(list(g)) for k, g in itertools.groupby(sorted(reqs, key=key), key)})
# ❌ 不排序的结果是错的：
print("groupby❌", {k: len(list(g)) for k, g in itertools.groupby(reqs, key)})

# .collect(counting()) → Counter
print("counting ", Counter(r.endpoint for r in reqs).most_common(2))

# ── 惰性：方括号 vs 圆括号，就是 OOM 与不 OOM 的区别 ──────────────
eager = [r.ms ** 2 for r in reqs]          # list，立刻算完、全在内存
lazy  = (r.ms ** 2 for r in reqs)          # generator，一次一个
print("eager    ", type(eager).__name__, sys.getsizeof(eager), "bytes")
print("lazy     ", type(lazy).__name__, sys.getsizeof(lazy), "bytes")
print("lazy sum ", sum(lazy))              # 生成器只能消费一次
print("再消费一次", sum(lazy))              # → 0，已耗尽！Java Stream 同理

# ArrayDeque → collections.deque
window: deque[float] = deque(maxlen=3)     # 定长滑动窗口，自动挤掉旧的
for r in reqs:
    window.append(r.ms)
print("deque    ", list(window))

"""阶段一 1.4 —— 函数、参数、作用域：六个 Java 人必踩的坑，全部可复现。"""

# ── 坑 1：可变默认参数（Python 头号陷阱）────────────────────────────
def add_bad(item, bucket=[]):          # ❌ 默认值在"定义时"求值一次
    bucket.append(item)
    return bucket

print("坑1 ❌", add_bad("a"), add_bad("b"), add_bad("c"))   # 三次调用共享同一个 list！
print("坑1 证据", add_bad.__defaults__)                      # 默认值是函数对象的属性

def add_ok(item, bucket=None):         # ✅ None 哨兵
    bucket = [] if bucket is None else bucket
    bucket.append(item)
    return bucket

print("坑1 ✅", add_ok("a"), add_ok("b"), add_ok("c"))

# ── 坑 2：默认值只求值一次，连时间戳也是 ────────────────────────────
import datetime as dt
def stamp_bad(t=dt.datetime.now()):    # ❌ 永远是 import 那一刻的时间
    return t
print("坑2 ❌", stamp_bad() is stamp_bad())   # True → 同一个对象

# ── 关键字参数：AI 库里到处都是 ─────────────────────────────────────
def generate(prompt, *, max_tokens=256, temperature=0.7, stream=False):
    #                 ↑ 星号之后必须用关键字传，防止调用方写 generate(p, 512, 0.7)
    return f"{prompt!r} max={max_tokens} temp={temperature} stream={stream}"

print("kwonly  ", generate("hi", max_tokens=512))
try:
    generate("hi", 512)                        # 位置传参会直接报错
except TypeError as e:
    print("kwonly  ", "TypeError:", e)

# ── *args / **kwargs：装饰器和配置透传的基础 ───────────────────────
def trace(fn, *args, **kwargs):
    print("trace   ", f"call {fn.__name__}(*{args}, **{kwargs})")
    return fn(*args, **kwargs)

print("trace   ", trace(generate, "hello", max_tokens=8, stream=True))

defaults = {"max_tokens": 128, "temperature": 0.2}
override = {"temperature": 0.9}
print("merge   ", generate("q", **{**defaults, **override}))   # 后者覆盖前者

# ── 坑 3：LEGB —— 赋值会静默新建局部变量 ───────────────────────────
counter = 0
def bump_bad():
    try:
        counter = counter + 1          # ❌ 一旦有赋值，counter 整个函数内都是局部的
    except UnboundLocalError as e:
        print("坑3 ❌", "UnboundLocalError:", e)

bump_bad()

def make_counter():                    # ✅ 闭包 + nonlocal
    n = 0
    def bump():
        nonlocal n
        n += 1
        return n
    return bump

c = make_counter()
print("坑3 ✅", c(), c(), c())

# ── 坑 4：循环里建闭包，抓的是变量不是值（Java lambda 要求 final，Python 不管）──
bad = [lambda: i for i in range(3)]
print("坑4 ❌", [f() for f in bad])                     # [2,2,2]
good = [lambda i=i: i for i in range(3)]                # 默认参数把值钉住
print("坑4 ✅", [f() for f in good])

# ── 浅拷贝 vs 深拷贝 ───────────────────────────────────────────────
import copy
cfg = {"model": "qwen", "gpus": [0, 1]}
shallow = dict(cfg)                    # ❌ 只拷一层
deep = copy.deepcopy(cfg)              # ✅
shallow["gpus"].append(2)
print("拷贝    ", "原 dict 被改了:", cfg["gpus"], "| deep 不受影响:", deep["gpus"])

#!/usr/bin/env python3
"""svclog —— 推理服务访问日志聚合器（阶段一综合练习）。

把一份 CSV 访问日志按 endpoint 聚合，算出 QPS、错误率、时延分位数，
按 P95 倒序打印成一张对齐的表格。这是你在 Java 里写过无数次的那类小工具。

用法：
    python svclog.py access.csv
    python svclog.py access.csv --top 3 --min-count 5
    python svclog.py access.csv --format csv > report.csv
"""

from __future__ import annotations

import argparse
import csv
import math
import sys
from collections import defaultdict
from collections.abc import Iterable, Iterator
from dataclasses import dataclass, field

# ── 领域模型 ────────────────────────────────────────────────────────────
# Java 里这是一个 record + 一个带 Builder 的统计类；Python 里两个 dataclass。


@dataclass(frozen=True, slots=True)
class Entry:
    """一行日志。frozen=True ≈ Java record，slots=True 省内存。"""

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

    @classmethod
    def from_row(cls, row: dict[str, str]) -> Entry:
        """工厂方法。Python 只有一个 __init__，重载靠 @classmethod。"""
        return cls(
            ts=float(row["ts"]),
            endpoint=row["endpoint"].strip(),
            status=int(row["status"]),
            ms=float(row["ms"]),
        )


@dataclass(slots=True)
class Stat:
    """一个 endpoint 的聚合结果。"""

    endpoint: str
    latencies: list[float] = field(default_factory=list)  # 可变默认值必须用 factory
    errors: int = 0
    first_ts: float = math.inf
    last_ts: float = -math.inf

    def add(self, e: Entry) -> None:
        self.latencies.append(e.ms)
        self.errors += e.status >= 400  # bool 是 int 的子类，可以直接加
        self.first_ts = min(self.first_ts, e.ts)
        self.last_ts = max(self.last_ts, e.ts)

    @property  # ≈ Java 的 getter，但调用方写 s.count 不写 s.getCount()
    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)

    def pct(self, q: float) -> float:
        """最近邻插值分位数（够用，不引 numpy）。"""
        if not self.latencies:
            return 0.0
        ordered = sorted(self.latencies)
        idx = min(len(ordered) - 1, max(0, math.ceil(q * len(ordered)) - 1))
        return ordered[idx]


# ── 读取：生成器，100 GB 日志也只占 O(1) 内存 ───────────────────────────
def read_entries(path: str) -> Iterator[Entry]:
    """惰性逐行产出 Entry —— Java 的 Files.lines() + map()。"""
    with open(path, newline="", encoding="utf-8") as f:  # with ≈ try-with-resources
        for lineno, row in enumerate(csv.DictReader(f), start=2):
            try:
                yield Entry.from_row(row)
            except (KeyError, ValueError) as e:
                print(f"skip line {lineno}: {e}", file=sys.stderr)


def aggregate(entries: Iterable[Entry]) -> dict[str, Stat]:
    """≈ stream().collect(groupingBy(Entry::endpoint, ...))。"""
    stats: dict[str, Stat] = defaultdict(lambda: Stat(""))
    for e in entries:
        s = stats[e.endpoint]
        if not s.endpoint:
            s.endpoint = e.endpoint
        s.add(e)
    return dict(stats)


# ── 输出 ────────────────────────────────────────────────────────────────
COLUMNS = ("endpoint", "count", "qps", "err%", "p50", "p95", "p99", "max")


def rows_of(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"{max(s.latencies):.0f}",
        )


def print_table(stats: list[Stat]) -> None:
    rows = [COLUMNS, *rows_of(stats)]
    widths = [max(len(r[i]) for r in rows) for i in range(len(COLUMNS))]
    for n, row in enumerate(rows):
        line = "  ".join(
            cell.ljust(w) if i == 0 else cell.rjust(w)
            for i, (cell, w) in enumerate(zip(row, widths))
        )
        print(line)
        if n == 0:
            print("  ".join("-" * w for w in widths))


def print_csv(stats: list[Stat]) -> None:
    w = csv.writer(sys.stdout)
    w.writerow(COLUMNS)
    w.writerows(rows_of(stats))


# ── 入口 ────────────────────────────────────────────────────────────────
def main(argv: list[str] | None = None) -> int:
    p = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    p.add_argument("path", help="CSV，表头需含 ts,endpoint,status,ms")
    p.add_argument("--top", type=int, default=0, help="只看 P95 最高的 N 个")
    p.add_argument("--min-count", type=int, default=1, help="样本数低于此值的丢弃")
    p.add_argument("--format", choices=("table", "csv"), default="table")
    args = p.parse_args(argv)

    stats = aggregate(read_entries(args.path))
    picked = [s for s in stats.values() if s.count >= args.min_count]
    picked.sort(key=lambda s: s.pct(0.95), reverse=True)  # ≈ Comparator.reversed()
    if args.top:
        picked = picked[: args.top]

    if not picked:
        print("没有满足条件的数据", file=sys.stderr)
        return 1
    (print_csv if args.format == "csv" else print_table)(picked)
    return 0


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

"""gen_log.py —— 造一份 2000 行的样例日志，便于试跑 svclog.py。"""
import random, csv, sys
random.seed(42)
SPEC = {   # endpoint: (基准延迟 ms, 抖动, 错误率, 权重)
    "/v1/chat/completions": (700, 400, 0.04, 50),
    "/v1/embeddings":       (35,  20,  0.01, 30),
    "/v1/models":           (4,   2,   0.00, 12),
    "/healthz":             (1,   1,   0.00, 8),
}
eps = list(SPEC); weights = [SPEC[e][3] for e in eps]
w = csv.writer(sys.stdout); w.writerow(["ts", "endpoint", "status", "ms"])
t = 1_726_600_000.0
for _ in range(2000):
    t += random.expovariate(20)                       # 平均 20 QPS
    ep = random.choices(eps, weights)[0]
    base, jitter, err, _ = SPEC[ep]
    bad = random.random() < err
    ms = round(random.gauss(base, jitter) if not bad else random.uniform(1, 20), 1)
    if random.random() < 0.02: ms = round(ms * random.uniform(4, 9), 1)   # 长尾
    w.writerow([f"{t:.3f}", ep, 500 if bad else 200, max(0.1, ms)])