本节目标：把前 17 章零散的知识点拧成一个能跑、能测、能打包的真实项目，走完「设计→实现→测试→质量门禁→端到端验证」的闭环。
适用版本：Python 3.12+（实测 3.14.6）
18.1 从零构建一个完整项目
知识点是散的，只有亲手串成一个项目才会连成网。本节做一个命令行日志分析器 loglens：读入应用日志，统计级别分布、错误率与高频来源。
18.1.1 选题与目录结构
这个题目刚好覆盖全书主干：正则（10 章）解析日志行，dataclass / StrEnum（6 章）建数据模型，Counter 与生成器（3、8 章）做统计，异常（7 章）处理坏行，类型注解（9 章）约束接口，argparse（11 章）做入口，pytest / ruff / mypy（14 章）保证质量，pyproject.toml（15 章）完成打包。采用 src 布局：
loglens/
├── pyproject.toml README.md .gitignore
├── src/loglens/ # __init__ / __main__ / models / parser / analyzer / cli
└── tests/ # test_parser.py / test_analyzer.py
src 布局让导入的是已安装的包而非当前目录，能提前暴露「漏打包」（第 15 章）。
18.1.2 pyproject.toml
项目元数据、依赖与工具配置都写在这里（略去 [tool.ruff.lint] 等细项）：
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[project]
name = "loglens"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = []
[project.optional-dependencies]
dev = ["pytest==9.1.1", "ruff==0.16.10", "mypy==2.4.0"]
[project.scripts]
loglens = "loglens.cli:main"
[tool.hatch.build.targets.wheel]
packages = ["src/loglens"]
[tool.mypy]
strict = true
files = ["src"]
[project.scripts] 声明控制台入口 loglens（第 15.2 节），安装后直接敲 loglens app.log。
18.1.3 数据模型 models.py
用 StrEnum 约束级别，dataclass 描述条目：
"""数据模型：用 dataclass 描述一条日志。"""
from __future__ import annotations
from dataclasses import dataclass, field
from datetime import datetime
from enum import StrEnum
class Level(StrEnum):
"""日志级别；继承 StrEnum 后可直接当字符串比较。"""
DEBUG = "DEBUG"
INFO = "INFO"
WARNING = "WARNING"
ERROR = "ERROR"
CRITICAL = "CRITICAL"
@dataclass(frozen=True, slots=True)
class LogEntry:
"""一条已解析的日志。frozen 让它可作 dict 键，slots 省内存。"""
timestamp: datetime
level: Level
source: str
message: str
@dataclass(slots=True)
class Report:
"""分析结果：各级别计数、各来源计数与错误明细。"""
total: int = 0
by_level: dict[Level, int] = field(default_factory=dict)
by_source: dict[str, int] = field(default_factory=dict)
errors: list[LogEntry] = field(default_factory=list)
@property
def error_rate(self) -> float:
"""ERROR 及以上级别占比；空报告返回 0.0。"""
if self.total == 0:
return 0.0
bad = self.by_level.get(Level.ERROR, 0) + self.by_level.get(Level.CRITICAL, 0)
return bad / self.total
frozen=True 让它不可变（第 3.3 节的引用语义），slots=True 省内存——收益在 18.2 节用 tracemalloc 实测。
18.1.4 解析器 parser.py
解析用第 10 章的正则，命名分组让代码自解释：
"""解析器：把一行文本变成 LogEntry。"""
from __future__ import annotations
import re
from collections.abc import Iterable, Iterator
from datetime import datetime
from loglens.models import Level, LogEntry
# 命名分组让字段提取更清晰；正则编译一次，反复复用。
_LINE_RE = re.compile(
r"^(?P<ts>\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})\s+"
r"(?P<level>[A-Z]+)\s+(?P<source>\S+)\s+(?P<msg>.*)$"
)
class ParseError(ValueError):
"""一行日志无法解析时抛出。"""
def parse_line(line: str) -> LogEntry:
"""解析单行；失败抛 ParseError。"""
stripped = line.strip()
match = _LINE_RE.match(stripped)
if match is None:
raise ParseError(f"无法解析的日志行: {stripped!r}")
parts = match.groupdict()
try:
level = Level(parts["level"])
except ValueError as exc:
raise ParseError(f"未知日志级别: {parts['level']!r}") from exc
return LogEntry(
timestamp=datetime.strptime(parts["ts"], "%Y-%m-%d %H:%M:%S"),
level=level,
source=parts["source"],
message=parts["msg"],
)
def parse_lines(lines: Iterable[str]) -> Iterator[LogEntry]:
"""惰性解析多行；跳过空行，解析失败直接抛出。"""
for raw in lines:
if raw.strip():
yield parse_line(raw)
parse_lines 是生成器（第 8 章），既能吃文件也能吃 sys.stdin，不会一次性把整个文件读进内存。
18.1.5 分析器 analyzer.py
单次遍历完成统计，用 Counter 做频次统计：
"""分析器：把一批 LogEntry 汇总成 Report。"""
from __future__ import annotations
from collections import Counter
from collections.abc import Iterable
from loglens.models import Level, LogEntry, Report
_BAD_LEVELS = {Level.ERROR, Level.CRITICAL}
def analyze(entries: Iterable[LogEntry]) -> Report:
"""单次遍历完成计数与错误收集。"""
by_level: Counter[Level] = Counter()
by_source: Counter[str] = Counter()
errors: list[LogEntry] = []
total = 0
for entry in entries:
total += 1
by_level[entry.level] += 1
by_source[entry.source] += 1
if entry.level in _BAD_LEVELS:
errors.append(entry)
return Report(total=total, by_level=dict(by_level),
by_source=dict(by_source), errors=errors)
def top_sources(report: Report, n: int = 3) -> list[tuple[str, int]]:
"""按次数取前 n 个来源；次数相同按名称排序保证输出稳定。"""
ranked = sorted(report.by_source.items(), key=lambda kv: (-kv[1], kv[0]))
return ranked[:n]
排序键 (-kv[1], kv[0]) 是「次数降序、名称升序」的稳定写法（第 4.3 节）。
18.1.6 CLI 入口 cli.py
入口用第 11 章的 argparse：build_parser() 声明位置参数 path（可选）与 --top。核心是 render 与 main：
def render(report: Report, top: int) -> str:
"""把 Report 渲染成纯文本报告。"""
lines = [f"总行数: {report.total}", f"错误率: {report.error_rate:.1%}", "级别分布:"]
for level in Level:
if count := report.by_level.get(level, 0):
lines.append(f" {level.value:<8} {count}")
lines.append(f"高频来源 (Top {top}):")
lines += [f" {src:<12} {cnt}" for src, cnt in top_sources(report, top)]
return "\n".join(lines)
def main(argv: list[str] | None = None) -> int:
args = build_parser().parse_args(argv)
stream = args.path.open(encoding="utf-8") if args.path is not None else sys.stdin
try:
with stream:
report = analyze(parse_lines(stream))
except ParseError as exc:
print(f"解析失败: {exc}", file=sys.stderr)
return 1
print(render(report, args.top))
return 0
render 与 main 分离，让渲染可单独测试；main 返回退出码，方便 shell 与 CI 调用。加上两行 __main__.py 即可 python -m loglens。
18.1.7 测试与质量门禁
tests/test_analyzer.py 用 _entry 工厂造数据，覆盖计数与错误率等核心行为：
def _entry(level: Level, source: str, second: int = 0) -> LogEntry:
return LogEntry(datetime(2026, 9, 18, 10, 0, second), level, source, "msg")
def test_analyze_counts_levels_and_sources() -> None:
entries = [_entry(Level.INFO, "app"), _entry(Level.ERROR, "db"),
_entry(Level.INFO, "app"), _entry(Level.CRITICAL, "db")]
report = analyze(entries)
assert report.total == 4
assert report.by_level[Level.INFO] == 2
assert report.by_source["app"] == 2
assert len(report.errors) == 2
tests/test_parser.py 另有 4 个用例，用 pytest.raises 覆盖「垃圾行」「未知级别」「空行跳过」。跑测试与门禁：
PYTHONPATH=src pytest -v && ruff check src/ tests/ && mypy
tests/test_analyzer.py::test_analyze_counts_levels_and_sources PASSED [ 12%]
tests/test_analyzer.py::test_error_rate PASSED [ 25%]
tests/test_analyzer.py::test_top_sources_is_stable_on_ties PASSED [ 37%]
tests/test_analyzer.py::test_empty_report_error_rate_is_zero PASSED [ 50%]
tests/test_parser.py::test_parse_line_extracts_all_fields PASSED [ 62%]
tests/test_parser.py::test_parse_line_rejects_garbage PASSED [ 75%]
tests/test_parser.py::test_parse_line_rejects_unknown_level PASSED [ 87%]
tests/test_parser.py::test_parse_lines_skips_blank_lines PASSED [100%]
============================== 8 passed in 0.07s ===============================
All checks passed!
Success: no issues found in 6 source files
第一版其实没过：ruff 报了 import 顺序（I001）与 cli.py 该用三元表达式（SIM108），mypy 报 Report 未被 analyzer 显式导出（attr-defined）。修到全绿才算完工——工具的价值就在于抓这些易被忽略的细节（第 14.3 节）。
18.1.8 端到端跑一次
造一份 10 行日志 sample.log，真实运行：
python -m loglens sample.log
总行数: 10
错误率: 30.0%
级别分布:
INFO 5
WARNING 2
ERROR 2
CRITICAL 1
高频来源 (Top 3):
app 5
db 3
cache 2
喂一行坏数据时，解析器会拒绝：
printf '2026-09-18 10:00:01 INFO app ok\nBAD LINE\n' | python -m loglens
解析失败: 无法解析的日志行: 'BAD LINE'
退出码为 1，错误走 stderr——管道里能被脚本捕获，又不污染正常输出。
18.1.9 README、.gitignore 与下一步
README 写清四件事：做什么、怎么装、怎么用、日志格式长什么样。.gitignore 至少忽略 __pycache__/、.venv/、build/、dist/、*.egg-info/ 与各工具缓存目录。项目能跑只是起点，可以继续加：
--level过滤：只统计某级别及以上的日志。--json输出：用dataclasses.asdict转 JSON，接入监控。- 读 gzip / 目录：
gzip.open与pathlib.Path.glob批量处理。
小结
- 一个完整项目 = 数据模型（
dataclass+StrEnum）+ 解析（正则生成器）+ 统计（Counter）+ 入口（argparse）+ 测试 + 质量门禁 + 打包。 - 采用
src布局能提前暴露「漏打包」；[project.scripts]声明控制台入口。 - 让「已校验的数据」在模块间传递：
parser只吐LogEntry，analyzer只吃LogEntry，接口靠类型注解约束。 - 真实开发里第一版几乎一定过不了 ruff/mypy，修到全绿本身就是工程能力的一部分。
到这里，你已经能把全书知识拧成一个可交付的项目。但「能跑」和「跑得快」是两回事——下一节我们学习性能剖析与优化，先学会测量，再谈优化。
阅读导航：上一节：数据分析入门 · 下一节：性能剖析与优化入门 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。