实战篇 / 第 23 章

第 23 章 · 实战一：命令行工具开发

先看清楚：为什么要把脚本做成「工具」

很多人写 Python 的方式是「打开 IDLE，粘一段脚本，python xxx.py 跑一下」。这在一次性任务里没问题，但脚本一旦要交给同事用、要每天定时跑、要放进 CI（持续集成）流水线，就会立刻暴露出一堆问题。

1.1 脚本与工具的六道分水岭

1.2 本章要做的项目：pycount

我们做一个代码统计与质量体检工具，功能需求如下（这些需求会驱动后面每一个技术选型）：

1. 扫描一个或多个目录/文件，统计各语言的代码行数、注释行数、空行数；
2. 按文件体积找出「巨型文件」，列出最长的若干行（所谓「大代码味道」）；
3. 统计函数/类定义数量，粗略估算复杂度信号（分支数、循环数）；
4. 输出人看的表格，或输出机器读的 JSON / CSV；
5. 阈值不满足时以非零退出码结束，便于接进 CI。

1.3 技术选型表

argparse：参数解析与子命令体系

argparse 是标准库自带的命令行接口（Command-Line Interface，CLI）解析框架。很多教程只讲 add_argument("--name")，但真实工具需要处理子命令（subcommand）、互斥参数、位置参数、默认值与类型转换、自定义错误。这一节把它们一次讲透。

2.1 最小可用示例

demo_argparse.py —— 参数解析的五个组成部分

"""argparse 五要素：解析器、位置参数、选项参数、解析、错误处理。"""
import argparse


def build_parser() -> argparse.ArgumentParser:
    """把「参数定义」单独抽成一个函数，方便测试与复用。"""
    parser = argparse.ArgumentParser(
        prog="pycount",                       # 帮助信息里显示的程序名
        description="统计代码行数与基本质量指标",  # 一句话描述
        epilog="示例：pycount ./src --json out.json",  # 结尾补充说明
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )

    # (1) 位置参数：必须有，数量固定
    parser.add_argument("paths", nargs="+", help="要扫描的文件或目录")

    # (2) 可选短参数与长参数
    parser.add_argument("-v", "--verbose", action="store_true", help="输出调试日志")

    # (3) 带类型的选项，type= 负责把字符串转成目标类型
    parser.add_argument("-m", "--min-size", type=int, default=0,
                        help="只报告大于该 KB 数的文件")

    # (4) choices 限定取值范围
    parser.add_argument("-f", "--format",
                        choices=["table", "json", "csv"],
                        default="table", help="输出格式")

    # (5) 带默认值且可被命令行覆盖的选项
    parser.add_argument("--exclude", nargs="*", default=[],
                        help="要排除的目录名，例如 --exclude .git venv")

    return parser


def main() -> int:
    args = build_parser().parse_args()
    print("扫描路径:", args.paths)
    print("输出格式:", args.format)
    print("最小体积:", args.min_size, "KB")
    print("是否详细:", args.verbose)
    print("排除目录:", args.exclude)
    return 0


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

运行与预期输出：

$ python demo_argparse.py ./src ./tests -f json -m 20 --exclude .git venv
扫描路径: ['./src', './tests']
输出格式: json
最小体积: 20 KB
是否详细: False
排除目录: ['.git', 'venv']

$ python demo_argparse.py --help
usage: pycount [-h] [-v] [-m MIN_SIZE] [-f {table,json,csv}]
               [--exclude [EXCLUDE ...]]
               paths [paths ...]

统计代码行数与基本质量指标

positional arguments:
  paths        要扫描的文件或目录

options:
  -h, --help            show this help message and exit
  -v, --verbose         输出调试日志
  -m, --min-size MIN_SIZE
                        只报告大于该 KB 数的文件
  -f, --format {table,json,csv}
                        输出格式（默认: table）
  --exclude [EXCLUDE ...]
                        要排除的目录名，例如 --exclude .git venv

2.2 常用参数声明速查表

2.3 子命令：把一个工具拆成多个「动词」

当工具功能变多（pycount scan、pycount history、pycount clean），把所有选项堆在一个解析器里会互相打架。正解是子解析器（subparser）：

"""子命令模式：每个子命令拥有自己独立的解析器与参数集合。"""
import argparse
import sys


def cmd_scan(args: argparse.Namespace) -> int:
    print(f"[scan] 扫描 {args.paths}，格式={args.format}")
    return 0


def cmd_history(args: argparse.Namespace) -> int:
    print(f"[history] 读取最近 {args.days} 天的快照")
    return 0


def cmd_clean(args: argparse.Namespace) -> int:
    print(f"[clean] 清理 {args.target}，dry_run={args.dry_run}")
    if not args.yes:
        print("需要 --yes 才会真正删除，已跳过。")
        return 1
    return 0


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(prog="pycount", description="代码统计工具")
    # add_subparsers(dest="command", required=True)：必选子命令
    sub = parser.add_subparsers(dest="command", title="可用命令", required=True)

    p_scan = sub.add_parser("scan", help="扫描目录并输出统计")
    p_scan.add_argument("paths", nargs="+", help="目标路径")
    p_scan.add_argument("-f", "--format", choices=["table", "json", "csv"],
                        default="table", help="输出格式")
    p_scan.add_argument("--min-size", type=int, default=0, help="文件大小下限 KB")
    p_scan.set_defaults(func=cmd_scan)      # 把处理函数挂到命名空间上

    p_hist = sub.add_parser("history", help="查看历史趋势")
    p_hist.add_argument("-d", "--days", type=int, default=7, help="回看天数")
    p_hist.set_defaults(func=cmd_history)

    p_clean = sub.add_parser("clean", help="清理缓存与快照")
    # 互斥组：路径与天数只能给一个，避免「到底清哪个」的歧义
    g = p_clean.add_mutually_exclusive_group(required=True)
    g.add_argument("--cache", action="store_true", help="只清缓存")
    g.add_argument("--snapshots", action="store_true", help="只清快照")
    p_clean.add_argument("--yes", action="store_true", help="确认执行")
    p_clean.add_argument("--dry-run", action="store_true", help="只演示不真删")
    p_clean.set_defaults(func=cmd_clean)

    return parser


def main(argv: list[str] | None = None) -> int:
    parser = build_parser()
    args = parser.parse_args(argv)
    if not hasattr(args, "func"):
        parser.print_help()
        return 2
    try:
        return args.func(args)
    except KeyboardInterrupt:
        print("\n已被用户中断", file=sys.stderr)
        return 130


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

预期输出：

$ python tool.py scan ./src -f json
[scan] 扫描 ['./src']，格式=json

$ python tool.py clean --cache --snapshots
tool.py clean: error: argument --snapshots: not allowed with argument --cache

$ python tool.py clean --cache
[clean] 清理 cache，dry_run=False
需要 --yes 才会真正删除，已跳过。

2.4 让 --help 真正好用

• 每一行 help= 都用中文说清「输入什么、得到什么」，别只写「set output format」；
• 用 ArgumentDefaultsHelpFormatter 自动把默认值写进帮助；
• 长选项写成 --min-size（动宾/名词短语），短选项保留给最常用的 1–2 个；
• 危险操作（删除、覆盖）必须提供 --dry-run 或 --yes 二次确认。

项目骨架：从单文件到可安装包

工具的代码组织直接决定它能不能被别人读懂、能不能被长期维护。本节给出我们认可的「四层骨架」，以及每一层的职责边界。

3.1 目录结构

pycount/
├── pyproject.toml            # 项目元数据 + 依赖 + 打包配置（现代唯一标准）
├── README.md
├── LICENSE
├── .gitignore
├── src/
│   └── pycount/              # src 布局：避免「本地目录被误 import」
│       ├── __init__.py       # 导出 __version__
│       ├── __main__.py       # 支持 python -m pycount
│       ├── cli.py            # ① 接口层：只负责解析参数、调用 core、打印结果
│       ├── config.py         # ② 配置层：默认值 + 文件 + 环境变量 + 命令行
│       ├── core.py           # ③ 领域层：纯函数，不碰 IO，最容易测试
│       ├── scanners.py       # ④ 适配层：遍历文件、读源码、判断语言
│       ├── render.py         # 输出渲染：table / json / csv / 进度条
│       └── logsetup.py       # 日志配置
├── tests/
│   ├── conftest.py           # pytest 夹具
│   ├── test_core.py
│   ├── test_config.py
│   └── test_cli.py
└── examples/
    └── sample_project/        # 给测试用的迷你项目

3.2 为什么推荐 src 布局

传统布局（包直接放在仓库根目录）有个隐蔽的坑：在项目根目录运行 pytest 时，Python 会把当前目录放进 sys.path，于是 import pycount 可能命中本地源码目录而不是 site-packages 里安装的版本。结果就是：你明明 pip install pycount，测的却还是工作区那份半成品。src 布局强制你「必须安装才能 import」，测试才可信。

3.3 各层职责与反例

src/pycount/core.py —— 领域层写成纯函数

"""领域层：纯粹的数据处理，没有任何 IO 副作用，因此极易测试。"""
from __future__ import annotations

import io
import tokenize
from dataclasses import dataclass, field


@dataclass(frozen=True, slots=True)
class LineStat:
    """一个源文件的统计结果。frozen=True 表示不可变，更安全。"""
    total: int
    code: int
    comment: int
    blank: int

    @property
    def comment_ratio(self) -> float:
        return self.comment / self.total if self.total else 0.0


@dataclass(slots=True)
class FileStat:
    path: str
    language: str
    size_kb: float
    lines: LineStat
    long_lines: list[tuple[int, int]] = field(default_factory=list)  # (行号, 长度)
    defs_count: int = 0   # 函数 + 类 定义数量
    branches: int = 0     # 复杂度信号：分支与循环总数


def analyze_source(text: str, path: str = "?", language: str = "text") -> FileStat:
    """核心分析函数：输入源码文本，输出统计结构。纯函数，无 IO。"""
    total = code = comment = blank = 0
    long_lines: list[tuple[int, int]] = []
    in_docstring = False
    docstring_delim = ""

    for idx, raw in enumerate(text.splitlines(), start=1):
        line = raw.strip()
        if len(line) > 100:
            long_lines.append((idx, len(line)))

        if not line:
            blank += 1
            continue

        # 识别块注释状态（支持 C 风格 /* */ 与文档字符串的简单近似）
        if in_docstring:
            comment += 1
            if docstring_delim in line:
                in_docstring = False
            continue
        if line.startswith('"""') or line.startswith("'''"):
            delim = line[:3]
            # 单行闭合：同时以同引号结尾且长度 >= 6
            if len(line) >= 6 and line.endswith(delim):
                comment += 1
            else:
                in_docstring = True
                docstring_delim = delim
                comment += 1
            continue
        if line.startswith("#"):
            comment += 1
            continue

        code += 1

    return FileStat(
        path=path,
        language=language,
        size_kb=round(len(text.encode("utf-8")) / 1024, 2),
        lines=LineStat(total=total or (code + comment + blank),
                       code=code, comment=comment, blank=blank),
        long_lines=long_lines,
    )


def complexity_signals(text: str) -> tuple[int, int]:
    """用 tokenize 统计 def/class 与分支循环次数。"""
    import ast

    defs = branches = 0
    branch_nodes = (ast.If, ast.For, ast.While, ast.ExceptHandler,
                    ast.With, ast.Assert, ast.BoolOp)
    try:
        tree = ast.parse(text)
    except SyntaxError:
        return 0, 0
    for node in ast.walk(tree):
        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)):
            defs += 1
        elif isinstance(node, branch_nodes):
            branches += 1
    return defs, branches


def merge_stats(stats: list[FileStat]) -> dict[str, int]:
    """汇总：给 renderer 用的总览字典。"""
    return {
        "files": len(stats),
        "total": sum(s.lines.total for s in stats),
        "code": sum(s.lines.code for s in stats),
        "comment": sum(s.lines.comment for s in stats),
        "blank": sum(s.lines.blank for s in stats),
    }


if __name__ == "__main__":          # 手写的小自测，方便直接运行本章文件
    demo = '''
def add(a, b):
    # 相加
    if a > b:
        return a + b
    return b + a
'''
    st = analyze_source(demo, "demo.py", "python")
    print(st.path, st.lines, merge_stats([st]))
    # 输出：demo.py LineStat(total=7, code=5, comment=1, blank=1) {...}

3.4 __main__.py 的作用

"""src/pycount/__main__.py —— 让 `python -m pycount` 可用。"""
import sys

from .cli import main

if __name__ == "__main__":
    sys.exit(main())

有了它，python -m pycount ./src 与安装后的 pycount ./src 走的是同一份代码路径，测试与使用行为一致——这是排查「本地能跑、装完就坏」的重要习惯。

3.5 pyproject.toml：现代项目的唯一真相源

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

[project]
name = "pycount"
version = "1.2.0"
description = "代码行数统计与质量体检命令行工具"
readme = "README.md"
requires-python = ">=3.10"
license = { text = "MIT" }
authors = [{ name = "你的名字", email = "you@example.com" }]
keywords = ["cli", "code-metrics", "static-analysis"]
classifiers = [
  "Development Status :: 4 - Beta",
  "Environment :: Console",
  "Programming Language :: Python :: 3.12",
  "Topic :: Software Development :: Quality Assurance",
]

# 运行时依赖（面向用户的包才需要，这里统计功能纯标准库即可）
dependencies = []

[project.optional-dependencies]
dev = ["pytest>=8.0", "pytest-cov", "ruff>=0.6", "build", "twine"]

[project.scripts]
pycount = "pycount.cli:main"      # 安装后生成的可执行命令

[project.urls]
Homepage = "https://github.com/yourname/pycount"
Issues = "https://github.com/yourname/pycount/issues"

[tool.hatch.build.targets.wheel]
packages = ["src/pycount"]

[tool.pytest.ini_options]
testpaths = ["tests"]
addopts = "-q --cov=pycount --cov-report=term-missing"

[tool.ruff]
line-length = 100
target-version = "py310"

[tool.ruff.lint]
select = ["E", "F", "I", "UP", "B", "SIM"]

3.6 开发态安装

$ python -m venv .venv
$ source .venv/bin/activate        # Windows: .venv\Scripts\activate
$ pip install -e ".[dev]"          # 可编辑安装 + 开发依赖
$ pycount ./src -f table
$ pytest                           # 跑测试
$ ruff check .                     # 静态检查
$ ruff format .                    # 自动格式化

配置管理：四个来源，一条优先级

成熟 CLI 工具的配置通常来自四个地方，后者覆盖前者：内置默认值 < 配置文件 < 环境变量 < 命令行参数。这个顺序叫「优先级阶梯」，用户永远可以用最高优先级覆盖任何东西。

4.1 从 TOML 读配置（Python 3.11+ 内置 tomllib）

"""配置层：把四个来源合并成一个不可变的 Settings 对象。"""
from __future__ import annotations

import os
import tomllib
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any

ENV_PREFIX = "PYCOUNT_"
CONFIG_NAMES = ("pycount.toml", ".pycount.toml")


@dataclass(frozen=True, slots=True)
class Settings:
    """最终生效的配置。所有业务代码只认这个对象。"""
    format: str = "table"
    exclude: tuple[str, ...] = ("venv", ".git", "__pycache__", "node_modules")
    min_size_kb: int = 0
    long_line_limit: int = 100
    verbose: bool = False
    no_color: bool = False
    sources: tuple[str, ...] = field(default=(), compare=False)

    def merged_with(self, cli: dict[str, Any]) -> "Settings":
        """把命令行 Namespace 中「用户显式指定过」的项覆盖进来。

        关键技巧：用 argparse 的 dest 列表区分「用户给了」和「用了默认值」。
        """
        data = {
            "format": self.format,
            "exclude": self.exclude,
            "min_size_kb": self.min_size_kb,
            "long_line_limit": self.long_line_limit,
            "verbose": self.verbose,
            "no_color": self.no_color,
            "sources": self.sources,
        }
        for key, value in cli.items():
            if value is None:          # 用户没给，保持低优先级来源
                continue
            if key == "exclude" and isinstance(value, list):
                value = tuple(value)
            data[key] = value
        return Settings(**data)


def _find_config_file(explicit: str | None) -> Path | None:
    """按「显式路径 -> 当前目录 -> 用户配置目录 -> XDG」的顺序找配置。"""
    if explicit:
        p = Path(explicit).expanduser()
        if not p.is_file():
            raise FileNotFoundError(f"配置文件不存在: {p}")
        return p
    for name in CONFIG_NAMES:
        if (p := Path.cwd() / name).is_file():
            return p
    xdg = os.environ.get("XDG_CONFIG_HOME") or (Path.home() / ".config")
    if (p := Path(xdg) / "pycount" / "config.toml").is_file():
        return p
    return None


def _from_file(path: Path) -> dict[str, Any]:
    with path.open("rb") as fh:                 # tomllib 需要二进制模式
        raw = tomllib.load(fh)
    # 配置里允许写 [tool.pycount] 段来避免与其它工具冲突
    return raw.get("tool", {}).get("pycount", raw)


def _from_env() -> dict[str, Any]:
    """环境变量转配置：PYCOUNT_MIN_SIZE_KB=50 -> min_size_kb=50"""
    out: dict[str, Any] = {}
    for raw_key, raw_val in os.environ.items():
        if not raw_key.startswith(ENV_PREFIX):
            continue
        key = raw_key[len(ENV_PREFIX):].lower()
        if key == "exclude":
            out[key] = tuple(v for v in raw_val.split(",") if v)
        elif key == "verbose":
            out[key] = raw_val.strip().lower() in {"1", "true", "yes", "on"}
        elif key == "no_color":
            out[key] = raw_val.strip().lower() in {"1", "true", "yes", "on"}
        elif key == "format":
            out[key] = raw_val.strip().lower()
        elif key.endswith("_kb") or key == "long_line_limit":
            try:
                out[key] = int(raw_val)
            except ValueError:
                raise ValueError(f"环境变量 {raw_key} 需要整数，得到 {raw_val!r}") from None
    return out


def load_settings(config_path: str | None = None,
                  cli: dict[str, Any] | None = None) ->:
        """四层合并入口。返回最终 Settings。"""
    layers: list[dict[str, Any]] = [{}]        # 第 1 层：代码默认值（在 dataclass 里）
    sources: list[str] = ["default"]

    if (cfg_file := _find_config_file(config_path)) is not None:
        layers.append(_from_file(cfg_file))
        sources.append(str(cfg_file))

    env_layer = _from_env()
    if env_layer:
        layers.append(env_layer)
        sources.append(f"env:{ENV_PREFIX}*")

    merged = Settings()
    for layer in layers:
        merged = merged.merged_with(layer)

    if cli:
        merged = merged.merged_with(cli)
        sources.append("argv")

    object.__setattr__(merged, "sources", tuple(sources))  # frozen 数据的写法
    return merged


if __name__ == "__main__":
    s = load_settings()
    print("格式:", s.format, "| 来源:", s.sources, "| 排除:", s.exclude)
    # 输出：格式: table | 来源: ('default',) | 排除: ('venv', '.git', '__pycache__', 'node_modules')

日志系统：从 print 到 logging

CLI 工具的输出有两个互相冲突的诉求：人看的要友好、有颜色、有进度；机器看的要干净、可解析。所以必须分流：数据走 stdout，诊断信息走 stderr。

这一条决定了你的工具能不能进 Unix 管道：pycount ./src -f json | jq . 能工作，前提是 stdout 里只有 JSON。

5.1 一个够用的日志配置模块

"""日志配置：级别、格式化、颜色、文件输出，全部集中在一处。"""
from __future__ import annotations

import logging
import os
import sys
from logging.handlers import RotatingFileHandler

LOGGER_NAME = "pycount"

RESET = "\033[0m"
COLORS = {
    "DEBUG": "\033[36m",     # 青
    "INFO": "\033[32m",      # 绿
    "WARNING": "\033[33m",  # 黄
    "ERROR": "\033[31m",     # 红
    "CRITICAL": "\033[1;41m",  # 红底白字加粗
}


class ColorFormatter(logging.Formatter):
    """给级别名加色的 Formatter；非 TTY 时自动降级为无色。"""

    def __init__(self, fmt: str, use_color: bool = True) -> None:
        super().__init__(fmt=fmt, datefmt="%H:%M:%S")
        self.use_color = use_color

    def format(self, record: logging.LogRecord) -> str:
        original = record.levelname
        if self.use_color:
            color = COLORS.get(record.levelname, "")
            record.levelname = f"{color}{original}{RESET}"
        try:
            return super().format(record)
        finally:
            record.levelname = original      # 还原，避免其它 handler 看到彩色串


def setup_logging(verbose: bool = False,
                   log_file: str | None = None,
                   use_color: bool | None = None) -> logging.Logger:
    """配置并返回全局 logger。可重复调用，不会叠加 handler。"""
    logger = logging.getLogger(LOGGER_NAME)
    if logger.handlers:                 # 幂等：先清掉旧 handler
        for h in logger.handlers[:]:
            logger.removeHandler(h)
            h.close()

    level = logging.DEBUG if verbose else logging.INFO
    logger.setLevel(level)
    logger.propagate = False            # 不向 root logger 冒泡，避免重复输出

    if use_color is None:
        use_color = (sys.stderr.isatty()
                     and os.environ.get("NO_COLOR") is None
                     and os.environ.get("TERM") != "dumb")

    console = logging.StreamHandler(sys.stderr)   # 日志一律走 stderr
    console.setFormatter(ColorFormatter("%(asctime)s %(levelname)-8s %(message)s",
                                       use_color=use_color))
    logger.addHandler(console)

    if log_file:
        # RotatingFileHandler：超过 5MB 就轮转，保留 3 份历史，避免日志撑爆磁盘
        fh = RotatingFileHandler(log_file, maxBytes=5 * 1024 * 1024,
                                 backupCount=3, encoding="utf-8")
        fh.setFormatter(logging.Formatter(
            "%(asctime)s %(levelname)s %(name)s %(filename)s:%(lineno)d %(message)s"))
        fh.setLevel(logging.DEBUG)       # 文件里永远记 DEBUG
        logger.addHandler(fh)

    return logger


def get_logger(suffix: str | None = None) -> logging.Logger:
    return logging.getLogger(f"{LOGGER_NAME}.{suffix}" if suffix else LOGGER_NAME)


if __name__ == "__main__":
    log = setup_logging(verbose=True)
    log.debug("这是一条调试信息")
    log.info("扫描完成，共 128 个文件")
    log.warning("%s 超过大小阈值", "huge.py")
    try:
        1 / 0
    except ZeroDivisionError:
        log.exception("统计比例时发生除零")   # exception 自动带 traceback

预期输出：

16:20:31 DEBUG    这是一条调试信息
16:20:31 INFO     扫描完成，共 128 个文件
16:20:31 WARNING  huge.py 超过大小阈值
16:20:31 ERROR    统计比例时发生除零
Traceback (most recent call last):
  File "logsetup.py", line 61, in __main__
    log.exception("统计比例时发生除零")
ZeroDivisionError: division by zero

5.2 日志级别选择与实践规范

• 用 %s 惰性格式化：log.info("扫描 %s", path)，不要 f-string（DEBUG 不打印时也能省开销）；
• 捕获异常用 log.exception() 或 log.error(..., exc_info=True)，别只 print(str(e))，否则丢掉堆栈；
• 不要在日志里写密码、Token、身份证号（用 mask() 之类的脱敏函数）；
• 库的代码不要自己调 basicConfig，只暴露 logger，由入口程序统一配置。

终端呈现：颜色、进度条、表格与中文宽度

CLI 的「界面」就是终端。要让人愿意用，得处理好三件事：颜色不能乱、进度要可见、表格要对齐。

6.1 颜色与终端能力探测

"""终端渲染工具集：颜色、宽度、进度条、表格。全部只写 stdout。"""
from __future__ import annotations

import os
import shutil
import sys
import time
import unicodedata
from typing import Iterable, Sequence

# ---- ANSI 转义序列 ----
RESET = "\033[0m"
BOLD = "\033[1m"
DIM = "\033[2m"


def supports_color(stream=None) -> bool:
    """探测当前是否应该上色：尊重 NO_COLOR、FORCE_COLOR、非 TTY。"""
    stream = stream or sys.stdout
    if os.environ.get("NO_COLOR") is not None:
        return False
    if os.environ.get("FORCE_COLOR"):
        return True
    if os.environ.get("TERM") == "dumb":
        return False
    return hasattr(stream, "isatty") and stream.isatty()


class Palette:
    """把颜色包成可开关的对象，避免业务代码里到处 if。"""
    COLORS = {"red": "31", "green": "32", "yellow": "33", "blue": "34",
              "magenta": "35", "cyan": "36", "gray": "90"}

    def __init__(self, enabled: bool = True) -> None:
        self.enabled = enabled

    def __call__(self, text: str, color: str | None = None,
                 bold: bool = False, dim: bool = False) -> str:
        if not self.enabled or not color:
            return text
        prefix = "\033["
        if bold:
            prefix += "1;"
        if dim:
            prefix += "2;"
        return f"{prefix}{self.COLORS[color]}m{text}{RESET}"


def display_width(text: str) -> int:
    """计算字符串在等宽终端里占的列数：中文/全角算 2 列。

    这是中文 CLI 表格能对齐的关键。str.ljust 对中文是错的，
    因为 len("你好") == 2 但终端显示宽度是 4。
    """
    width = 0
    for ch in text:
        if unicodedata.combining(ch):
            continue
        width += 2 if unicodedata.east_asian_width(ch) in ("W", "F") else 1
    return width


def pad(text: str, width: int, align: str = "left") -> str:
    """按显示宽度补空格，而不是按字符个数。"""
    gap = max(0, width - display_width(text))
    if align == "right":
        return " " * gap + text
    if align == "center":
        left = gap // 2
        return " " * left + text + " " * (gap - left)
    return text + " " * gap


def term_width(default: int = 80) -& int:
    return shutil.get_terminal_size((default, 24)).columns


class ProgressBar:
    """极简进度条：用 \\r 原地刷新；非 TTY 时降级为每 10% 打一行。"""

    def __init__(self, total: int, desc: str = "", color: Palette | None = None,
                 unit: str = "文件") -> None:
        self.total = max(1, total)
        self.desc = desc
        self.palette = color or Palette()
        self.unit = unit
        self.n = 0
        self.start = time.perf_counter()
        self._last_printed = ""

    def update(self, step: int = 1) -> None:
        self.n += step
        self._render()

    def _render(self) -> None:
        ratio = min(1.0, self.n / self.total)
        bar_w = min(30, max(10, term_width() - 40))
        filled = int(bar_w * ratio)
        bar = "█" * filled + "░" * (bar_w - filled)
        elapsed = time.perf_counter() - self.start
        eta = (elapsed / ratio - elapsed) if ratio > 0 else 0.0
        line = (f"\r{self.palette(self.desc, 'cyan')} [{bar}] "
                f"{ratio * 100:5.1f}% {self.n}/{self.total} {self.unit} "
                f"ETA {eta:4.1f}s")

        if not sys.stdout.isatty():
            # 非交互环境只记里程碑，避免污染日志文件
            pct = int(ratio * 10) * 10
            if line != self._last_printed and int(ratio * 10) * 10 != getattr(self, "_last_pct", -1):
                self._last_pct = int(ratio * 10) * 10
                sys.stdout.write(f"{self.desc} {pct}% ({self.n}/{self.total})\n")
            self._last_printed = line
            return

        sys.stdout.write(line[:term_width() - 1])
        sys.stdout.flush()
        self._last_printed = line

    def close(self, message: str | None = None) -> None:
        """结束时必须换行，否则后续 shell 提示符会粘在进度条后面。"""
        elapsed = time.perf_counter() - self.start
        sys.stdout.write("\r" + " " * (term_width() - 1) + "\r")
        if message:
            sys.stdout.write(message + "\n")
        else:
            sys.stdout.write(self.palette(
                f"完成：{self.n} {self.unit}，耗时 {elapsed:.2f}s\n", "green"))
        sys.stdout.flush()


def render_table(headers: Sequence[str], rows: Iterable[Sequence[str]],
                 aligns: Sequence[str] | None = None,
                 color: Palette | None = None) -> str:
    """渲染一个带边框的表格，自动按中文显示宽度对齐。"""
    color = color or Palette()
    rows = [[str(c) for c in row] for row in rows]
    ncol = len(headers)
    aligns = list(aligns or ["left"] * ncol)
    widths = [display_width(h) for h in headers]
    for row in rows:
        for i, cell in enumerate(row[:ncol]):
            widths[i] = max(widths[i], display_width(cell))

    top = "┌" + "┬".join("─" * (w + 2) for w in widths) + "┐"
    sep = "├" + "┼".join("─" * (w + 2) for w in widths) + "┤"
    bottom = "└" + "┴".join("─" * (w + 2) for w in widths) + "┘"

    out = [color(top, "gray")]
    out.append(color("│", "gray")
               + color("│", "gray").join(f" {pad(h, widths[i])} "
                                         for i, h in enumerate(headers))
               + color("│", "gray"))
    out.append(color(sep, "gray"))
    for row in rows:
        cells = []
        for i in range(ncol):
            cell = row[i] if i < len(row) else ""
            body = pad(cell, widths[i], aligns[i])
            if color.enabled and i == 0:
                body = color(body, "cyan")
            cells.append(f" {body} ")
        out.append(color("│", "gray") + color("│", "gray").join(cells)
                   + color("│", "gray"))
    out.append(color(bottom, "gray"))
    return "\n".join(out)


if __name__ == "__main__":
    c = Palette(supports_color())
    print(render_table(["语言", "文件数", "代码行", "注释率"],
                       [["Python", "128", "24 310", "18.4%"],
                        ["JavaScript", "42", "9 820", "6.1%"]],
                       aligns=["left", "right", "right", "right"], color=c))
    bar = ProgressBar(5, "扫描")
    for _ in range(5):
        bar.update()
        time.sleep(0.05)
    bar.close()

    # 三种运行环境的输出：
    # 1) 真终端：带颜色、进度条原地刷新
    # 2) 重定向到文件：no_color=True，进度条每 10% 打一行
    # 3) CI 日志：同上，且 export NO_COLOR=1

6.2 JSON / CSV 渲染与「给人 vs 给机器」

"""输出渲染：同一种数据，三种面向。渲染层不认识业务。"""
from __future__ import annotations

import csv
import io
import json
import sys
from dataclasses import asdict
from typing import Any

from .core import FileStat, merge_stats


def to_dicts(stats: list[FileStat]) -> list[dict[str, Any]]:
    """领域对象 -> 可 JSON 序列化的普通字典。"""
    out = []
    for s in stats:
        d = asdict(s)
        d["lines"] = asdict(s.lines)
        d["comment_ratio"] = round(s.lines.comment_ratio, 4)
        out.append(d)
    return out


def render_json(stats: list[FileStat], indent: int = 2) -> str:
    payload = {"summary": merge_stats(stats), "files": to_dicts(stats)}
    # ensure_ascii=False 让中文原样输出；sort_keys 让结果可 diff
    return json.dumps(payload, ensure_ascii=False, indent=indent, sort_keys=True)


def render_csv(stats: list[FileStat]) -> str:
    buf = io.StringIO()
    w = csv.writer(buf, lineterminator="\n")
    w.writerow(["path", "language", "size_kb", "total", "code",
                "comment", "blank", "defs", "branches"])
    for s in stats:
        w.writerow([s.path, s.language, s.size_kb, s.lines.total,
                    s.lines.code, s.lines.comment, s.lines.blank,
                    s.defs_count, s.branches])
    return buf.getvalue()


def render_markdown(stats: list[FileStat], color=None) -&tr[str]:
    """顺带支持 Markdown，方便贴进 PR 评论或周报。"""
    out = ["## 代码统计", "",
           "| 文件 | 语言 | 代码 | 注释 | 空白 |",
           "| --- | --- | ---: | ---: | ---: |"]
    for s in stats:
        out.append(f"| {s.path} | {s.language} | {s.lines.code} "
                   f"| {s.lines.comment} | {s.lines.blank} |")
    return out


def emit(text: str, out_path: str | None = None) -> None:
    """统一出口：写文件或写 stdout，绝不打印到 stderr。"""
    if out_path:
        with open(out_path, "w", encoding="utf-8", newline="\n") as fh:
            fh.write(text)
        # 提示走 stderr 或 log，stdout 保持纯净
        print(f"已写入 {out_path}", file=sys.stderr)
    else:
        sys.stdout.write(text)
        if not text.endswith("\n"):
            sys.stdout.write("\n")

错误处理与退出码：工具的契约

CLI 工具最重要的对外接口有两个：stdout 的输出格式与退出码。用户会用 && / || / echo $?、CI 会用它判定成败，所以退出码必须严格遵守 Unix 惯例。

7.1 统一的主入口骨架

"""cli.py：整个工具的主入口，薄薄一层，只做编排。"""
from __future__ import annotations

import sys
import traceback
from pathlib import Path

from .config import Settings, load_settings
from .core import FileStat, analyze_source, complexity_signals, merge_stats
from .logsetup import get_logger, setup_logging
from .render import emit, render_csv, render_json, render_markdown
from .scanners import LANG_BY_SUFFIX, iter_source_files

EXIT_OK = 0
EXIT_FAIL = 1
EXIT_USAGE = 2
EXIT_CONFIG = 3
EXIT_INPUT = 4


class ToolError(Exception):
    """带退出码的业务异常：让错误处理变成一行 raise。"""

    def __init__(self, message: str, code: int = EXIT_FAIL) -> None:
        super().__init__(message)
        self.code = code


def build_parser():
    """参数定义，见第 2 节；此处省略重复部分。"""
    from .cli import build_parser as _bp  # 保持示例自洽
    return _bp()


def main(argv: list[str] | None = None) -> int:
    """CLI 入口。返回 int 作为进程退出码。"""
    from .argparser import build_parser   # 真实工程里单独放一个模块

    parser = build_parser()
    args = parser.parse_args(argv)

    # 1) 先配日志（需要尽早知道是否 verbose）
    color = not args.no_color
    log = setup_logging(verbose=args.verbose, log_file=args.log_file,
                        use_color=color)
    log.debug("解析到的参数: %s", args)

    try:
        # 2) 合并配置
        settings: Settings = load_settings(args.config, vars(args))
        log.info("配置来源: %s", " <- ".join(settings.sources)
                 if False else " <- ".join(settings.sources))

        # 3) 校验输入
        for p in args.paths:
            if not Path(p).exists():
                raise ToolError(f"路径不存在: {p}", EXIT_INPUT)

        # 4) 干活（核心逻辑全在纯函数里）
        stats: list[FileStat] = []
        for path in iter_source_files(args.paths, set(settings.exclude)):
            try:
                text = path.read_text(encoding="utf-8")
            except UnicodeDecodeError:
                log.warning("%s 不是 UTF-8 文本，已跳过", path)
                continue
            except OSError as exc:
                log.warning("无法读取 %s：%s", path, exc)
                continue
            st = analyze_source(text, str(path),
                                LANG_BY_SUFFIX.get(path.suffix, "text"))
            st.defs_count, st.branches = complexity_signals(text)
            if st.size_kb < settings.min_size_kb:
                continue
            stats.append(st)

        summary = merge_stats(stats)
        log.info("统计完成：%d 个文件 / %d 行代码", summary["files"], summary["code"])

        # 5) 渲染输出
        text = {"json": lambda: render_json(stats),
                "csv": lambda: render_csv(stats),
                "markdown": lambda: "\n".join(render_markdown(stats)),
                "table": lambda: _render_table_report(stats, summary)}[args.format]()
        emit(text, args.output)

        # 6) 用退出码表达「检查结果」（CI 关心的部分）
        return _judge(stats, settings, log)

    except ToolError as exc:
        log.error("%s", exc)
        return exc.code
    except KeyboardInterrupt:
        log.warning("已被用户中断")
        return 130
    except Exception as exc:                       # noqa: BLE001
        log.critical("未预期的错误: %s", exc)
        if args.verbose:
            traceback.print_exc()
        return EXIT_FAIL


def _judge(stats: list[FileStat], settings: Settings, log) -> int:
    """阈值判定：给 --max-lines / --max-long-lines 用的质量闸门。"""
    too_long = [s for s in stats if s.lines.total > 800]
    if too_long:
        log.warning("发现 %d 个超过 800 行的巨型文件", len(too_long))
        for s in too_long[:5]:
            log.warning("  %s (%d 行)", s.path, s.lines.total)
    if settings.min_size_kb and not stats:
        return EXIT_FAIL
    return EXIT_OK


def _render_table_report(stats: list[FileStat], summary: dict) -> str:
    from .render import Palette, render_table, supports_color
    c = Palette(supports_color())
    rows = [(s.path, s.language, str(s.lines.code), str(s.lines.comment),
             f"{s.lines.comment_ratio * 100:.1f}%") for s in stats[:20]]
    table = render_table(["文件", "语言", "代码", "注释", "注释率"],
                         rows, aligns=["left", "left", "right", "right", "right"],
                         color=c)
    foot = (f"\n合计：{summary['files']} 个文件，"
            f"{summary['code']} 行代码，{summary['comment']} 行注释，"
            f"{summary['blank']} 行空白")
    return table + c(foot, "bold")


if __name__ == "__main__":
    sys.exit(main())

7.2 文件遍历适配层

"""scanners.py：唯一做文件系统 IO 的地方。"""
from __future__ import annotations

import os
from collections.abc import Iterator
from pathlib import Path

# 常见扩展名 -> 语言名。新增语言只改这张表。
LANG_BY_SUFFIX: dict[str, str] = {
    ".py": "Python", ".pyi": "Python",
    ".js": "JavaScript", ".mjs": "JavaScript", ".ts": "TypeScript",
    ".java": "Java", ".kt": "Kotlin", ".go": "Go", ".rs": "Rust",
    ".c": "C", ".h": "C", ".cpp": "C++", ".cs": "C#", ".rb": "Ruby",
    ".php": "PHP", ".swift": "Swift", ".sh": "Shell", ".sql": "SQL",
    ".html": "HTML", ".css": "CSS", ".md": "Markdown", ".json": "JSON",
    ".yml": "YAML", ".yaml": "YAML", ".toml": "TOML",
}

DEFAULT_EXCLUDE = {".git", ".hg", ".svn", "venv", ".venv", "__pycache__",
                   "node_modules", "dist", "build", ".mypy_cache",
                   ".pytest_cache", ".ruff_cache", ".tox", "site-packages"}

# 二进制/超大文件快速排除：看扩展名 + 看前 8KB 有没有 NUL 字节
BINARY_HINT = b"\x00"
SNIFF_SIZE = 8192


def _looks_binary(path: Path) -> bool:
    try:
        with path.open("rb") as fh:
            return BINARY_HINT in fh.read(SNIFF_SIZE)
    except OSError:
        return True


def iter_source_files(roots: list[str] | Path,
                      exclude: set[str] | None = None,
                      follow_symlinks: bool = False) -> Iterator[Path]:
    """深度优先遍历，产出所有「有已知扩展名且非二进制」的文件。

    三个防坑点：
    1. 跳过排除目录，且要在 os.walk 的 dirnames 上原地修改才生效；
    2. 符号链接默认不跟随，否则可能绕成环；
    3. 单个文件权限错误不应中断整个遍历。
    """
    exclude = DEFAULT_EXCLUDE | set(exclude or ())
    for root in roots:
        p = Path(root)
        if p.is_file():
            if p.suffix in LANG_BY_SUFFIX and not _looks_binary(p):
                yield p
            continue
        for dirpath, dirnames, filenames in os.walk(p,
                                                    followlinks=follow_symlinks):
            # 原地修改 dirnames 才能真正剪枝（这是 os.walk 的约定）
            dirnames[:] = [d for d in dirnames if d not in exclude
                           and not d.startswith(".")]
            for name in sorted(filenames):
                fp = Path(dirpath) / name
                if fp.suffix not in LANG_BY_SUFFIX:
                    continue
                try:
                    if _looks_binary(fp):
                        continue
                except OSError:
                    continue
                yield fp


if __name__ == "__main__":
    import sys
    count = 0
    for f in iter_source_files(sys.argv[1:] or ["."]):
        count += 1
        if count <= 10:
            print(f)
    print(f"共找到 {count} 个源文件")
    # 预期：列出前 10 个路径，然后打印总数

单元测试：让重构不再是赌博

工具没有测试，用户就只能靠「不敢改」来维持它。测试的目标不是覆盖 100% 行，而是覆盖所有会咬人的分支：边界、异常、非 ASCII、异常输入。

8.1 pytest 基础与夹具

"""tests/conftest.py —— 夹具（fixture）：测试之间共享的准备与清理。"""
from __future__ import annotations

import os
from pathlib import Path

import pytest


@pytest.fixture
def sample_project(tmp_path: Path) -> Path:
    """动态生成一个迷你项目，避免把测试数据和源码混在一起。"""
    root = tmp_path / "sample"
    (root / "pkg").mkdir(parents=True)
    (root / "venv").mkdir()
    (root / "node_modules").mkdir()

    (root / "pkg" / "a.py").write_text(
        "def add(a, b):\n"
        "    # 相加\n"
        "    return a + b\n",
        encoding="utf-8",
    )
    (root / "pkg" / "b.py").write_text(
        '"""模块文档"""\n'
        "class Foo:\n"
        "    pass\n",
        encoding="utf-8",
    )
    (root / "pkg" / "big.py").write_text("x = 1\n" * 5, encoding="utf-8")
    (root / "venv" / "ignored.py").write_text("should = 'be'\nignored\n",
                                              encoding="utf-8")
    (root / "node_modules" / "dep.js").write_text("var x = 1;\n", encoding="utf-8")
    (root / "image.png").write_bytes(b"\x89PNG\r\n\x1a\n\x00\x00binary")
    (root / "notes.md").write_text("# 说明\n", encoding="utf-8")
    return root


@pytest.fixture
def clean_env(monkeypatch):
    """隔离环境变量，防止开发者本机的 PYCOUNT_* 影响测试。"""
    for key in list(os.environ):
        if key.startswith("PYCOUNT_"):
            monkeypatch.delenv(key, raising=False)
    monkeypatch.delenv("NO_COLOR", raising=False)
    return monkeypatch

"""tests/test_core.py —— 领域层纯函数测试。"""
import pytest

from pycount.core import analyze_source, complexity_signals, merge_stats


def test_analyze_basic_counts():
    src = "def f():\n    # 注释\n    return 1\n\n"
    st = analyze_source(src, "t.py", "Python")
    assert st.lines.code == 2
    assert st.lines.comment == 1
    assert st.lines.blank == 1
    assert st.lines.total == 4


def test_analyze_multiline_docstring_is_comment():
    src = '"""\n标题\n说明\n"""\ncode()\n'
    st = analyze_source(src, "t.py", "Python")
    assert st.lines.comment == 4        # 三引号三行 + 代码行前那行
    assert st.lines.code == 1


def test_analyze_empty_file_no_zero_division():
    st = analyze_source("", "e.py", "Python")
    assert st.lines.comment_ratio == 0.0
    assert st.lines.total == 0


def test_long_lines_detected():
    src = "y = '" + "a" * 200 + "'\n"
    st = analyze_source(src, "long.py", "Python")
    assert st.long_lines == [(1, 204)]


def test_complexity_signals_counts_defs_and_branches():
    src = (
        "def f(x):\n"
        "    if x:\n"
        "        for i in range(x):\n"
        "            while i:\n"
        "                i -= 1\n"
        "    return x\n"
        "class C:\n"
        "    pass\n"
    )
    defs, branches = complexity_signals(src)
    assert defs == 2            # f + C
    assert branches == 3        # if / for / while


def test_complexity_signals_on_broken_code_is_safe():
    defs, branches = complexity_signals("def (:\n")   # 语法错误
    assert (defs, branches) == (0, 0)


def test_merge_stats():
    a = analyze_source("x=1\n", "a.py", "Python")
    b = analyze_source("# c\n\n", "b.py", "Python")
    total = merge_stats([a, b])
    assert total["files"] == 2
    assert total["code"] == 1
    assert total["comment"] == 1
    assert total["blank"] == 1


@pytest.mark.parametrize("text,expect_code", [
    ("", 0),
    ("\n\n\n", 0),
    ("   \t  \n", 0),      # 只含空白的行算空行
    ("x=1", 1),            # 无末尾换行也要计入
])
def test_boundary_inputs(text, expect_code):
    assert analyze_source(text).lines.code == expect_code

"""tests/test_cli.py —— 端到端测试 CLI 行为与退出码。"""
import json

import pytest

from pycount.cli import main
from pycount.scanners import iter_source_files


def test_scan_json_output(sample_project, capsys, clean_env):
    rc = main(["scan", str(sample_project), "-f", "json"])
    assert rc == 0
    out = capsys.readouterr().out
    data = json.loads(out)               # stdout 必须是干净 JSON
    assert data["summary"]["files"] == 4  # a.py b.py big.py notes.md
    assert {"Python", "Markdown"} == {f["language"] for f in data["files"]}


def test_excluded_dirs_are_skipped(sample_project, capsys, clean_env):
    main(["scan", str(sample_project), "-f", "json"])
    out = capsys.readouterr().out
    assert "venv" not in out
    assert "node_modules" not in out


def test_missing_path_returns_exit_code_4(capsys, clean_env):
    rc = main(["scan", "/definitely/not/here"])
    assert rc == 4


def test_bad_subcommand_is_usage_error(clean_env):
    with pytest.raises(SystemExit) as exc:
        main(["nosuchcommand"])
    assert exc.value.code == 2


def test_output_to_file_keeps_stdout_clean(sample_project, tmp_path,
                                          capsys, clean_env):
    target = tmp_path / "r.json"
    rc = main(["scan", str(sample_project), "-f", "json", "-o", str(target)])
    assert rc == 0
    assert capsys.readouterr().out == ""      # stdout 必须为空
    assert json.loads(target.read_text(encoding="utf-8"))["summary"]["files"] == 4


def test_env_var_overridden_by_cli(sample_project, capsys, monkeypatch,
                                   clean_env):
    monkeypatch.setenv("PYCOUNT_FORMAT", "csv")
    main(["scan", str(sample_project)])                 # 用环境变量 -> csv
    first = capsys.readouterr().out
    assert first.splitlines()[0].startswith("path,language")
    main(["scan", str(sample_project), "-f", "json"])   # 命令行更高优先级
    assert capsys.readouterr().out.lstrip().startswith("{")


def test_iter_source_files_is_lazy_and_sorted(sample_project):
    files = list(iter_source_files([sample_project]))
    assert [p.name for p in files] == sorted(p.name for p in files)
    assert all(p.suffix != ".png" for p in files)

8.2 覆盖率与常见反模式

$ pytest --cov=pycount --cov-report=term-missing
=================== 87.4% coverage ====================
Name                 Stmts   Miss  Cover   Missing
--------------------------------------------------
pycount/__init__.py       4      0   100%
pycount/cli.py           96     11    89%   58-63, 91-94
pycount/config.py        73      4    95%   96-98
pycount/core.py          58      0   100%
pycount/render.py        63      6    90%   88, 102-106
pycount/scanners.py      41      2    95%   60-61
--------------------------------------------------
TOTAL                   335     23    93%

打包与发布：从源码到 PyPI

写完工具只是完成一半——另一半是让它可以被安装。这一步包含三条路径：源码分发（PyPI，pip install）、本地工具（pipx）、单文件可执行程序（PyInstaller）。

9.1 三种分发形态对比

9.2 构建与检查分发包

$ python -m build            # 产出 dist/*.whl 与 dist/*.tar.gz
$ twine check dist/*        # 校验 README 渲染与元数据
Successfully built pycount-1.2.0.tar.gz and pycount-1.2.0-py3-none-any.whl

$ twine upload --repository testpypi dist/*   # 先发到测试源验证流程
$ twine upload dist/*                          # 正式发布

发布前的自查清单：

1. pyproject.toml 里 version 是否与 git tag 一致；
2. README 是否含安装与用法（PyPI 页面就是你的落地页）；
3. LICENSE 是否存在、分类器是否声明兼容 Python 版本；
4. python -m build && pip install dist/*.whl && pycount --help 在干净环境跑通；
5. 版本号一旦上传不可覆盖，只能递增重发。

9.3 PyInstaller 打包单文件 exe

# install PyInstaller
$ pip install pyinstaller

# 最简命令：生成单文件、无控制台窗口（Windows GUI 程序）
$ pyinstaller --onefile --name pycount pycount_entry.py

# 更完整的命令
$ pyinstaller \
    --onefile \
    --name pycount \
    --distpath ./dist \
    --workpath ./build \
    --specpath . \
    --noconfirm \
    --clean \
    --hidden-import pycount.core \
    --exclude-module tkinter \
    --exclude-module matplotlib \
    pycount_entry.py

pycount_entry.py —— 打包入口文件

"""PyInstaller 需要一个顶层脚本作为入口。"""
import multiprocessing
import sys

from pycount.cli import main

if __name__ == "__main__":
    # Windows 下冻结的 exe 用 multiprocessing 必须走 freeze_support
    multiprocessing.freeze_support()
    sys.exit(main())

9.4 运行时路径：源码模式 vs 冻结模式

"""paths.py —— 解决「打包后 os.getcwd() 变了 / 模板文件读不到」的问题。"""
from __future__ import annotations

import os
import sys
from pathlib import Path


def is_frozen() -> bool:
    """判断当前是否运行在 PyInstaller 打包出的程序里。"""
    return getattr(sys, "frozen", False) and hasattr(sys, "_MEIPASS")


def app_dir() -> Path:
    """应用资源所在目录。

    PyInstaller 的 onefile 模式会把资源解压到临时目录 sys._MEIPASS，
    因此不能用 __file__ 的父目录，得走 _MEIPASS。
    """
    if is_frozen():
        return Path(sys._MEIPASS)          # type: ignore[attr-defined]
    return Path(__file__).resolve().parent.parent


def user_data_dir() -> Path:
    """用户可写目录：放配置、缓存、导出文件。

    绝不要往程序安装目录写东西——Windows 的 Program Files 是只读的。
    """
    if sys.platform == "win32":
        base = Path(os.environ.get("APPDATA", Path.home() / "AppData" / "Roaming"))
    elif sys.platform == "darwin":
        base = Path.home() / "Library" / "Application Support"
    else:
        base = Path(os.environ.get("XDG_DATA_HOME", Path.home() / ".local" / "share"))
    d = base / "pycount"
    d.mkdir(parents=True, exist_ok=True)
    return d


def default_config_path() -> Path:
    return user_data_dir() / "config.toml"


if __name__ == "__main__":
    print("frozen:", is_frozen())
    print("app_dir:", app_dir())
    print("data_dir:", user_data_dir())

9.5 让程序自我更新（可选进阶）

"""一个克制的「检查更新」实现：只提示，不擅自安装。"""
from __future__ import annotations

import json
import urllib.request
from importlib.metadata import PackageNotFoundError, version

PYPI_URL = "https://pypi.org/pypi/pycount/json"


def _parse(v: str) -> tuple[int, ...]:
    return tuple(int(x) for x in v.split(".")[:3] if x.isdigit())


def latest_version() -> str | None:
    try:
        with urllib.request.urlopen(PYPI_URL, timeout=3) as resp:
            return json.load(resp)["info"]["version"]
    except (OSError, ValueError, KeyError):
        return None      # 网络不通就静默降级，绝不让工具挂掉


def check_update() -> str | None:
    try:
        cur = version("pycount")
    except PackageNotFoundError:
        return None
    latest = latest_version()
    if latest and _parse(latest) > _parse(cur):
        return f"有新版本 {latest}（当前 {cur}），运行 pipx upgrade pycount"
    return None

端到端整合：一份完整可运行的工具

把前面的模块拼起来，就是一个真正能用的 pycount。本节给出最终骨架与验证步骤，建议你按顺序敲一遍。

10.1 目录与启动流程

pycount/
├── pyproject.toml
├── README.md
├── src/pycount/
│   ├── __init__.py       # __version__ = "1.2.0"
│   ├── __main__.py       # python -m pycount
│   ├── cli.py            # main(argv) -> int
│   ├── argparser.py      # build_parser()
│   ├── config.py         # Settings / load_settings
│   ├── logsetup.py       # setup_logging / get_logger
│   ├── core.py           # analyze_source / complexity_signals / merge_stats
│   ├── scanners.py       # iter_source_files / LANG_BY_SUFFIX
│   ├── render.py         # render_json / render_csv / render_table
│   └── paths.py          # is_frozen / user_data_dir
└── tests/
    ├── conftest.py
    ├── test_core.py
    ├── test_config.py
    └── test_cli.py

1. __main__.py 或安装后的 pycount 调 cli.main()；
2. argparser.build_parser() 解析 argv；
3. logsetup.setup_logging() 按 -v 决定级别；
4. config.load_settings() 合并四层配置；
5. scanners.iter_source_files() 遍历并过滤；
6. core.analyze_source() 逐文件统计（纯函数，可并行）；
7. render.* 输出到 stdout 或文件；
8. 按阈值返回退出码。

10.2 参数全集（最终版）

usage: pycount [-h] [-v] [--config PATH] [--no-color] [--log-file PATH]
              [-e NAME ...] [-m KB] [--long-line N]
              {scan,report,config} ...

pycount —— 代码行数统计与质量体检

子命令:
  scan      扫描路径并输出统计（默认）
  report    生成 Markdown 周报并追加到文件
  config    查看当前生效的配置及其来源

通用选项:
  -h, --help              显示帮助并退出
  -v, --verbose           输出 DEBUG 日志
  --config PATH           指定 TOML 配置文件
  --no-color              禁用彩色输出
  --log-file PATH         同时写入滚动日志文件
  -e, --exclude NAME      排除目录名（可多个）
  --version               显示版本号

scan 子命令:
  paths...                 目标文件或目录
  -f, --format {table,json,csv,markdown}
  -o, --output PATH        写入文件而不是标准输出
  -m, --min-size KB       只统计大于该体积的文件
  --long-line N            长行阈值（默认 100）
  --no-progress            不显示进度条

10.3 一次完整的使用演示

$ pycount scan ./src -f json -m 1 --no-progress | head -30
{
  "files": [
    {
      "branches": 12,
      "comment": 18,
      "defs": 6,
      "defs_count": 6,
      "language": "Python",
      "long_lines": [],
      "path": "src/pycount/core.py",
      "size_kb": 5.12
    }
  ],
  "summary": {
    "blank": 96,
    "code": 412,
    "comment": 74,
    "files": 7,
    "total": 582
  }
}

$ pycount scan . -f table
语言        文件   代码行   注释率
────────────────────────────────────
Python        128    24310     18.4%
Markdown        12      1860      5.2%
合计：140 个文件，26170 行代码，5000 行注释，3200 行空白

$ echo $?
0

$ pycount scan . --max-lines 500
2024-05-20 16:30:12 WARNING 发现 3 个超过 500 行的巨型文件
2024-05-20 16:30:12 WARNING   src/legacy/parser.py (1284 行)
$ echo $?
1

$ pycount scan /nope
2024-05-20 16:31:02 ERROR 路径不存在: /nope
$ echo $?
4

进阶话题与常见工程决策

11.1 大目录的加速：并行化

扫描 10 万个文件时，IO 成为瓶颈。CPU 密集的 analyze_source 可以直接喂给进程池：

"""并行分析：CPU 密集用 ProcessPool，IO 密集用 ThreadPool。"""
from __future__ import annotations

import os
from concurrent.futures import ProcessPoolExecutor, as_completed

from .core import FileStat, analyze_source, complexity_signals


def analyze_one(job: tuple[str, str, str]) -> FileStat:
    """必须定义成模块级函数，进程池才能 pickle 它。"""
    path, text, lang = job
    st = analyze_source(text, path, lang)
    st.defs_count, st.branches = complexity_signals(text)
    return st


def analyze_many(jobs: list[tuple[str, str, str]],
                 workers: int | None = None) -> list[FileStat]:
    workers = workers or min(8, (os.cpu_count() or 2))
    if workers <= 1 or len(jobs) < 20:
        return [analyze_one(j) for j in jobs]      # 小任务别付进程启动成本

    results: list[FileStat] = []
    with ProcessPoolExecutor(max_workers=workers) as pool:
        futures = {pool.submit(analyze_one, j): j[0] for j in jobs}
        for fut in as_completed(futures):
            try:
                results.append(fut.result())
            except Exception as exc:             # noqa: BLE001
                print(f"分析 {futures[fut]} 失败: {exc}")
    return results

11.2 交互式确认：危险操作的保护栏

"""删除类操作的安全确认。"""
from __future__ import annotations

import sys
from pathlib import Path


def confirm(prompt: str, *, assume_yes: bool = False,
            dry_run: bool = False) -> bool:
    if dry_run:
        print("[dry-run] 跳过确认，不做任何实际修改")
        return False
    if assume_yes:
        return True
    if not sys.stdin.isatty():
        # 非交互环境（如 CI）绝不默认执行破坏性操作
        print("非交互环境：请显式传入 --yes 以确认执行", file=sys.stderr)
        return False
    while True:
        try:
            ans = input(f"{prompt} [y/N] ").strip().lower()
        except EOFError:
            return False
        if ans in {"y", "yes"}:
            return True
        if ans in {"", "n", "no"}:
            return False
        print("请输入 y 或 n")

11.3 Shell 补全

# 生成 bash 补全脚本（argparse 自带）
$ python -m pycount --generate-completion bash > /etc/bash_completion.d/pycount

# 里面大致是：
_pycount_scan() {
    local cur prev
    COMPREPLY=()
    cur="${COMP_WORDS[COMP_CWORD]}"
    prev="${COMP_WORDS[COMP_CWORD-1]}"
    case "${prev}" in
        -f|--format) COMPREPLY=( $(compgen -W "table json csv markdown" -- "${cur}") ); return 0 ;;
        -e|--exclude) COMPREPLY=( $(compgen -d -- "${cur}") ); return 0 ;;
    esac
}
complete -F _pycount_scan pycount

11.4 GitHub Actions 中的 CI 用法

# .github/workflows/quality.yml
name: code-quality
on: [push, pull_request]

jobs:
  pycount:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      - name: 安装工具
        run: |
          python -m pip install --upgrade pip
          pip install -e ".[dev]"
      - name: 跑测试
        run: pytest --cov=pycount --cov-report=xml
      - name: 静态检查
        run: ruff check .
      - name: 代码体检（超阈值即失败）
        run: pycount scan src -f markdown --max-lines 500 --no-progress

11.5 常见工程决策速查

本章小结

• 工具与脚本的分水岭在入口、输出、结构、依赖、质量、分发六个维度：参数化、结构化、标准日志、声明式依赖、自动化测试、可安装分发。
• argparse 的核心是 add_argument 的五个维度（名字、类型、动作、默认、帮助）与 add_subparsers() + set_defaults(func=...) 的子命令分发模式。
• 项目骨架推荐 src 布局与四层结构：cli（编排）/ config（配置）/ core（纯领域）/ render + scanners（IO 适配）。core 保持纯粹是测试成本低的关键。
• 配置优先级阶梯：代码默认值 < 配置文件 < 环境变量 < 命令行参数；argparse 用 default=None 标记「用户未指定」，否则会击穿阶梯。
• 日志走 stderr，数据走 stdout，这样 pycount -f json | jq 才能工作；用 RotatingFileHandler 防日志撑爆磁盘。
• 终端呈现要处理三件事：颜色需探测 isatty/NO_COLOR/FORCE_COLOR；进度条要区分 TTY 与重定向；中文对齐必须用 unicodedata.east_asian_width 计算显示宽度。
• 退出码是 CLI 的对外契约：0 成功、1 业务失败、2 用法错误、3 配置错误、4 输入缺失、130 中断。用异常携带退出码让处理收敛成一行。
• 测试重点是边界与异常路径：空文件、非 UTF-8、二进制、超长行、权限错误、参数非法；用 tmp_path、monkeypatch、capsys 保证隔离。
• 打包：PyPI 源码包是主路径（python -m build + twine upload，版本不可覆盖）；PyInstaller 用于免依赖场景，注意 sys._MEIPASS 资源路径与 spawn 模式的 pickle 限制。

练习题

1. 把单文件脚本改造成子命令工具。手头有一个 300 行的分析脚本，功能混在一起。用子命令模式重构为 stats scan ... / stats export ... / stats clean ...，要求每个子命令有独立参数、独立的 --help、以及各自的处理函数，通过 set_defaults(func=...) 分发。至少写 6 个针对参数解析的测试（缺参数、非法取值、互斥冲突、默认值、类型转换失败）。
   参考答案要点主解析器只保留通用选项与 add_subparsers(required=True, dest="command")；每个子解析器 set_defaults(func=...)；测试用 pytest.raises(SystemExit) 断言退出码 2，并检查 captured.err 里包含子命令名。
2. 实现四层配置合并并写测试矩阵。完成 load_settings，让「默认值 / 配置文件 / 环境变量 / 命令行」任意组合都得到正确结果。写 @pytest.mark.parametrize 覆盖 8 种组合，重点验证：配置文件设了 format、环境变量设了 min_size_kb、命令行同时给 -f 和 -m 时的最终值。
   参考答案要点argparse 全部 default=None；合并函数按顺序 dataclasses.replace；monkeypatch.setenv("PYCOUNT_FORMAT", "csv") 注入环境变量；tmp_path 写 TOML 再传 --config。注意 frozen dataclass 修改要整体替换。
3. 实现一个支持中文的表格渲染器并验证对齐。写 render_table(headers, rows)，要求在终端中每一列严格对齐，包含中文（2 列宽）、英文（1 列宽）、含 ANSI 颜色的单元格（不计入宽度）。写测试断言输出行中相同列的分隔竖线位于相同的字符下标。
   参考答案要点display_width() 用 unicodedata.east_asian_width；对已着色的字符串先剥离 ANSI 再算宽（可用正则 re.sub(r"\033\[[0-9;]*m", "", s)）；测试用 capsys 取输出后按行索引竖线位置断言。
4. 进度条与非交互环境。写一个 ProgressBar，要求：TTY 下原地刷新（\r）并显示百分比与 ETA；非 TTY（isatty() 为 False）时每 10% 输出一整行，不含 \r；结束后一定换行。分别用伪终端与重定向两种方式验证输出，并说明为什么进度条必须写 stderr。
   参考答案要点用 unittest.mock.patch("sys.stdout.isatty", return_value=False) 模拟非 TTY；ETA 用 elapsed/ratio - elapsed；进度条走 stderr 是因为它属于「诊断信息」，放 stdout 会污染 | jq 管道。结束用 finally: bar.close() 保证异常路径也换行。
5. 退出码契约与 CI 集成。给工具定义退出码表（0/1/2/3/4/130），写一个 ToolError 异常携带退出码，并构造 6 种失败场景验证码值正确。再写一个 GitHub Actions 工作流：跑测试 + 跑 ruff + 跑 pycount --max-lines 400 作为质量闸门（超阈值让 CI 变红）。
   参考答案要点except ToolError as e: log.error(...); return e.code；argparse 自身的用法错误天然是 2；KeyboardInterrupt 返回 130。YAML 里质量闸门直接靠退出码，不用写 if。
6. 实现 analyze_source 的健壮版本。要求：用 tokenize 正确识别所有形式的注释与字符串（覆盖单双引号、三引号、f-string、行内 # 在字符串内的情况），正确统计 f-string 内的 {a if b else c}（这会让朴素的 ast 计数出错吗？验证一下），并对语法错误文件返回 line=0 而不抛异常。写至少 10 个用例。
   参考答案要点用 tokenize.generate_tokens(io.StringIO(src).readline)，tokenize.NAME 之后是 COMMENT、STRING 就跳过；tokenize.TokenError/IndentationError 都要 catch。f-string 在 3.12 里由 FSTRING_START/MIDDLE/END 三个 token 组成，if/else 在 FSTRING_MIDDLE 里，ast 的 BoolOp 能正确覆盖但 JoinedStr 需要展开处理。
7. 用 PyInstaller 打包并在无 Python 环境的机器验证。打出单文件可执行程序，要求：体积尽量小（排除未用模块）；配置与缓存写入用户目录而不是 exe 同级；pycount --version 正确。用 python -m build 同时产出 PyPI 包，并在 pipx install dist/*.whl 后验证命令行入口。
   参考答案要点--exclude-module tkinter --exclude-module unittest --exclude-module pydoc 可明显瘦身；用 is_frozen() + sys._MEIPASS 判定模式；user_data_dir() 按平台返回 AppData/Library Support/XDG_DATA_HOME；验证时把 PATH 里的 python 摘掉再跑。
8. （选做）给 pycount 加一个 watch 子命令。监听目录变化（可用 watchdog），文件变化时增量重新统计并输出差异（新增/修改/删除的文件及其行数变化）。要求支持 --debounce 去抖、--once 单次运行、以及 Ctrl-C 时干净退出。
   参考答案要点watchdog.observers.Observer 监听 on_modified/on_created/on_deleted；用 threading.Timer 做去抖（改动后 debounce 秒内再触发）；用「路径 -> 行数」的字典做 diff；事件回调里更新状态要加锁，或用 queue.Queue 把事件传给主线程处理，避免竞态。Ctrl-C 用 observer.stop(); observer.join() 保证退出。