Python 命令行工具
命令行工具就是在终端里通过命令运行的小程序。它没有图形界面,但非常适合做自动化任务,例如批量重命名文件、清洗数据、导出报表、巡检服务、初始化数据库、触发离线任务。
零基础学习命令行工具时,不要只会写 input()。商业项目里的 CLI 更强调:参数可解释、失败有退出码、危险操作可预览、日志可追踪、核心逻辑可测试、脚本可被 CI 或定时任务调用。
学完本页后,你应该能做到:
- 看懂一条命令由程序、位置参数、选项参数、开关参数组成。
- 使用
argparse写出支持--help的命令行工具。 - 理解标准输入、标准输出、标准错误、退出码的作用。
- 会设计
--dry-run、--json、--verbose、--config等商业常用参数。 - 会把 CLI 参数解析和业务逻辑分层,方便测试和复用。
- 能写一个“医疗资产 CSV 导入预校验”命令行 Demo。
- 能排查路径、编码、权限、退出码、定时任务环境差异等问题。
命令行工具适合做什么
| 场景 | 例子 | 为什么适合 CLI |
|---|---|---|
| 数据导入 | 导入医院资产 CSV | 可重复执行,可记录成功失败 |
| 数据清洗 | 清洗订单、日志、对账文件 | 文件输入输出明确 |
| 巡检脚本 | 检查数据库、Redis、接口健康 | 可被定时任务和 CI 调用 |
| 批处理 | 批量生成报表、批量重命名 | 不需要图形界面 |
| 运维辅助 | 初始化索引、执行迁移、导出配置 | 可参数化、可审计 |
| AI 工程 | 批量生成 embedding、评估 RAG 数据 | 适合离线长任务 |
命令行工具不是“临时脚本随便写”。越是会批量修改数据、删除文件、触发任务的工具,越要严谨。
从 input 到参数化
最简单的命令行交互程序:
name = input("请输入你的名字:")
print(f"你好,{name}")运行:
python hello.py这适合学习,但不适合自动化。因为 CI、定时任务、Shell 脚本无法方便地“人工输入”。真实 CLI 更常见的是:
python import_assets.py ./assets.csv --output ./result --dry-run参数化的好处:
- 命令可以复制、审计、记录。
- 可以被脚本、定时任务、流水线调用。
- 可以通过
--help自解释。 - 可以用退出码告诉调用方成功或失败。
一条命令由什么组成
例如:
python backup.py ./docs --output ./backup --zip --dry-run拆开看:
| 部分 | 类型 | 含义 |
|---|---|---|
python | 执行程序 | 使用 Python 解释器运行 |
backup.py | 脚本文件 | 要执行的 Python 文件 |
./docs | 位置参数 | 要备份的目录 |
--output ./backup | 选项参数 | 输出目录 |
--zip | 开关参数 | 是否压缩 |
--dry-run | 开关参数 | 只预览,不真正执行 |
执行流程:
flowchart TD
A["用户输入命令"] --> B["Shell 拆分命令和参数"]
B --> C["启动 Python 解释器"]
C --> D["执行脚本文件"]
D --> E["argparse 解析参数"]
E --> F["校验参数是否合法"]
F --> G{"参数是否正确"}
G -->|否| H["打印错误和帮助信息"]
H --> I["返回非 0 退出码"]
G -->|是| J["执行业务逻辑"]
J --> K["输出结果"]
K --> L["返回退出码"]标准输入、标准输出、标准错误
CLI 程序有三个常见通道:
| 通道 | 英文 | 作用 |
|---|---|---|
| 标准输入 | stdin | 接收输入,例如管道传入内容 |
| 标准输出 | stdout | 输出正常结果 |
| 标准错误 | stderr | 输出错误、警告、诊断信息 |
为什么要区分 stdout 和 stderr?
如果用户把命令结果重定向到文件,正常结果应该进入文件,错误信息应该仍然显示在终端或日志里。
import sys
print("正常结果") # stdout
print("错误信息", file=sys.stderr) # stderr示例:
python tool.py > result.txt 2> error.log含义:
result.txt保存正常输出。error.log保存错误输出。
退出码
命令行程序应该用退出码告诉调用方执行结果。
| 退出码 | 含义 |
|---|---|
0 | 成功 |
1 | 通用失败 |
2 | 参数错误,argparse 默认常用 |
| 其他非 0 | 按项目约定表示不同失败类型 |
示例:
def main() -> int:
try:
print("执行成功")
return 0
except Exception as exc:
print(f"执行失败:{exc}")
return 1
if __name__ == "__main__":
raise SystemExit(main())为什么用 raise SystemExit(main())?
main() 返回整数,SystemExit 把它变成进程退出码。这样 Shell、CI、定时任务能判断命令是否成功。
python tool.py
echo $?Windows PowerShell:
python tool.py
$LASTEXITCODE使用 argparse
argparse 是 Python 标准库,不需要额外安装,适合大多数脚本和基础工具。
import argparse
from pathlib import Path
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
description="统计目录下的指定后缀文件数量"
)
parser.add_argument("directory", help="要统计的目录")
parser.add_argument("--suffix", default=".md", help="文件后缀,默认 .md")
parser.add_argument("--recursive", action="store_true", help="是否递归子目录")
return parser
def count_files(directory: Path, suffix: str, recursive: bool) -> int:
pattern = f"**/*{suffix}" if recursive else f"*{suffix}"
return sum(1 for path in directory.glob(pattern) if path.is_file())
def main() -> int:
parser = build_parser()
args = parser.parse_args()
root = Path(args.directory)
if not root.exists():
parser.error(f"目录不存在:{root}")
if not root.is_dir():
parser.error(f"不是目录:{root}")
total = count_files(root, args.suffix, args.recursive)
print(f"共找到 {total} 个 {args.suffix} 文件")
return 0
if __name__ == "__main__":
raise SystemExit(main())运行:
python count_files.py ./docs --recursive查看帮助:
python count_files.py --helpparser.error() 会输出错误信息并返回参数错误退出码,适合处理参数非法。
参数类型详解
| 类型 | 示例 | 说明 |
|---|---|---|
| 位置参数 | ./docs | 必填,通常表示核心输入 |
| 选项参数 | --output ./dist | 可选,通常带值 |
| 开关参数 | --recursive | 出现就是 True |
| 多值参数 | --tags java python | 一次传多个值 |
| 枚举参数 | --format json | 限制固定取值 |
| 子命令 | user add | 一个工具包含多个功能 |
位置参数
parser.add_argument("filename", help="文件名")位置参数必须传,否则程序会提示错误。
选项参数
parser.add_argument("--output", default="./dist", help="输出目录")选项参数可以设置默认值,用户不传时使用默认值。
开关参数
parser.add_argument("--dry-run", action="store_true", help="只预览,不真正执行")--dry-run 非常重要,适合删除文件、发布服务、修改数据库、批量导入这类危险操作。
类型转换
parser.add_argument("--limit", type=int, default=100, help="最多处理多少条")如果用户传 --limit abc,argparse 会提示参数类型错误。
枚举值 choices
parser.add_argument(
"--format",
choices=["text", "json", "csv"],
default="text",
help="输出格式",
)限制取值可以减少运行到业务深处才报错。
多值参数
parser.add_argument("--systems", nargs="+", help="来源系统列表")运行:
python tool.py --systems HIS LIS PACS子命令
当工具有多个能力时,使用子命令。
import argparse
def main() -> int:
parser = argparse.ArgumentParser(description="资产工具")
subparsers = parser.add_subparsers(dest="command", required=True)
import_parser = subparsers.add_parser("import", help="导入资产")
import_parser.add_argument("file")
check_parser = subparsers.add_parser("check", help="检查文件")
check_parser.add_argument("file")
args = parser.parse_args()
if args.command == "import":
print(f"导入 {args.file}")
elif args.command == "check":
print(f"检查 {args.file}")
return 0
if __name__ == "__main__":
raise SystemExit(main())运行:
python asset_tool.py import assets.csv
python asset_tool.py check assets.csvCLI 分层设计
不要把所有逻辑都写在 main() 里。推荐拆成:
asset_tool/
cli.py # 解析命令行参数,处理输出和退出码
service.py # 核心业务逻辑
validators.py # 参数和数据校验
report.py # 输出结果
tests/
test_service.py调用流程:
flowchart TD
A["cli.py 解析参数"] --> B["校验路径和选项"]
B --> C["调用 service.py"]
C --> D["执行业务逻辑"]
D --> E["返回结构化结果"]
E --> F["cli.py 输出文本或 JSON"]
F --> G["返回退出码"]分层的好处:
- 参数解析和业务逻辑互不干扰。
- 单元测试可以直接测试
service.py,不用模拟命令行。 - 以后改成 Web API 或定时任务时,核心逻辑还能复用。
- 错误处理更清晰:CLI 负责把异常转成用户提示和退出码。
配置文件和环境变量
命令行参数适合每次执行会变化的值,例如输入文件、输出目录、是否 dry-run。环境变量适合密钥、连接串这类敏感配置。
| 来源 | 适合放什么 | 是否提交 Git |
|---|---|---|
| 命令参数 | 输入文件、输出格式、limit | 命令本身可记录 |
| 配置文件 | 默认目录、业务阈值 | 模板可提交 |
| 环境变量 | 密码、Token、数据库连接 | 不提交 |
读取环境变量:
import os
database_url = os.environ["DATABASE_URL"]如果关键配置缺失,建议启动就失败,不要等处理到一半才发现。
输出给人看和输出给程序看
CLI 输出分两种:
| 输出类型 | 特点 | 示例 |
|---|---|---|
| 人类可读 | 友好、解释充分 | 成功 10 条,失败 2 条 |
| 机器可读 | 结构稳定,便于脚本处理 | JSON、CSV |
支持 --json:
import json
def print_result(result: dict, as_json: bool) -> None:
if as_json:
print(json.dumps(result, ensure_ascii=False))
else:
print(f"成功 {result['success']} 条,失败 {result['failed']} 条")如果命令会被 CI 或其他程序调用,提供 JSON 输出会更稳。
商业 Demo:医疗资产 CSV 预校验工具
需求:用户导入医院资产 CSV 前,先用 CLI 做预校验。
命令:
python asset_import_check.py assets.csv --output ./out --dry-run --json规则:
- CSV 必须包含
asset_code、asset_name、department、price。 asset_code必填且不能重复。asset_name必填。department必填。price必须是大于等于 0 的数字。--dry-run只输出统计,不写文件。- 非 dry-run 时输出
clean_assets.csv和error_assets.csv。
完整 Demo:
import argparse
import csv
import json
import sys
from dataclasses import dataclass
from decimal import Decimal, InvalidOperation
from pathlib import Path
@dataclass(frozen=True)
class AssetRow:
asset_code: str
asset_name: str
department: str
price: Decimal
@dataclass(frozen=True)
class CheckResult:
clean_rows: list[AssetRow]
error_rows: list[dict]
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description="医疗资产 CSV 导入预校验工具")
parser.add_argument("file", help="待校验 CSV 文件")
parser.add_argument("--output", default="./out", help="输出目录")
parser.add_argument("--dry-run", action="store_true", help="只预览,不写结果文件")
parser.add_argument("--json", action="store_true", help="以 JSON 格式输出统计")
return parser
def parse_price(value: str) -> Decimal:
try:
price = Decimal(value)
except (InvalidOperation, TypeError) as exc:
raise ValueError("price 必须是数字") from exc
if price < 0:
raise ValueError("price 不能小于 0")
return price
def validate_row(row: dict, line_no: int, seen_codes: set[str]) -> AssetRow:
asset_code = (row.get("asset_code") or "").strip()
asset_name = (row.get("asset_name") or "").strip()
department = (row.get("department") or "").strip()
price_text = (row.get("price") or "").strip()
if not asset_code:
raise ValueError("asset_code 不能为空")
if asset_code in seen_codes:
raise ValueError(f"asset_code 重复: {asset_code}")
if not asset_name:
raise ValueError("asset_name 不能为空")
if not department:
raise ValueError("department 不能为空")
price = parse_price(price_text)
seen_codes.add(asset_code)
return AssetRow(asset_code, asset_name, department, price)
def check_file(path: Path) -> CheckResult:
if not path.exists():
raise FileNotFoundError(f"文件不存在: {path}")
if not path.is_file():
raise ValueError(f"不是普通文件: {path}")
clean_rows: list[AssetRow] = []
error_rows: list[dict] = []
seen_codes: set[str] = set()
required = {"asset_code", "asset_name", "department", "price"}
with path.open("r", encoding="utf-8", newline="") as file:
reader = csv.DictReader(file)
if not reader.fieldnames or not required.issubset(reader.fieldnames):
raise ValueError(f"CSV 表头必须包含: {sorted(required)}")
for line_no, row in enumerate(reader, start=2):
try:
clean_rows.append(validate_row(row, line_no, seen_codes))
except ValueError as exc:
row["line_no"] = line_no
row["error"] = str(exc)
error_rows.append(row)
return CheckResult(clean_rows=clean_rows, error_rows=error_rows)
def write_outputs(output_dir: Path, result: CheckResult) -> None:
output_dir.mkdir(parents=True, exist_ok=True)
with (output_dir / "clean_assets.csv").open("w", encoding="utf-8", newline="") as file:
writer = csv.DictWriter(
file,
fieldnames=["asset_code", "asset_name", "department", "price"],
)
writer.writeheader()
for row in result.clean_rows:
writer.writerow(
{
"asset_code": row.asset_code,
"asset_name": row.asset_name,
"department": row.department,
"price": str(row.price),
}
)
error_fields = ["asset_code", "asset_name", "department", "price", "line_no", "error"]
with (output_dir / "error_assets.csv").open("w", encoding="utf-8", newline="") as file:
writer = csv.DictWriter(file, fieldnames=error_fields, extrasaction="ignore")
writer.writeheader()
writer.writerows(result.error_rows)
def print_summary(result: CheckResult, as_json: bool) -> None:
summary = {
"success": len(result.clean_rows),
"failed": len(result.error_rows),
}
if as_json:
print(json.dumps(summary, ensure_ascii=False))
else:
print(f"校验通过 {summary['success']} 行,失败 {summary['failed']} 行")
def main() -> int:
parser = build_parser()
args = parser.parse_args()
try:
result = check_file(Path(args.file))
print_summary(result, args.json)
if not args.dry_run:
write_outputs(Path(args.output), result)
return 0 if not result.error_rows else 1
except Exception as exc:
print(f"执行失败: {exc}", file=sys.stderr)
return 2
if __name__ == "__main__":
raise SystemExit(main())执行流程:
flowchart TD
A["输入命令"] --> B["argparse 解析参数"]
B --> C["检查文件是否存在"]
C --> D["读取 CSV 表头"]
D --> E["逐行校验字段"]
E --> F{"当前行是否合法"}
F -->|是| G["加入 clean_rows"]
F -->|否| H["加入 error_rows"]
G --> I{"是否还有下一行"}
H --> I
I -->|有| E
I -->|没有| J["输出统计"]
J --> K{"是否 dry-run"}
K -->|是| L["不写文件"]
K -->|否| M["写 clean/error CSV"]
L --> N["返回退出码"]
M --> N这个 Demo 体现的工程能力:
- CLI 层只处理参数、输出和退出码。
check_file是核心业务,可以单独测试。--dry-run避免误写文件。--json方便被其他脚本或 CI 解析。stderr输出错误,不污染正常输出。- 返回码区分成功、校验失败、程序异常。
测试 CLI
核心逻辑优先直接测函数:
from pathlib import Path
def test_check_file_success(tmp_path: Path):
csv_file = tmp_path / "assets.csv"
csv_file.write_text(
"asset_code,asset_name,department,price\n"
"A001,心电监护仪,ICU,12000\n",
encoding="utf-8",
)
result = check_file(csv_file)
assert len(result.clean_rows) == 1
assert len(result.error_rows) == 0需要测试完整命令时,可以用 subprocess:
import subprocess
import sys
def test_cli_help():
result = subprocess.run(
[sys.executable, "asset_import_check.py", "--help"],
text=True,
capture_output=True,
)
assert result.returncode == 0
assert "医疗资产 CSV" in result.stdout不要只靠人工运行。CLI 很容易被定时任务和流水线调用,一旦参数或退出码改坏,会影响自动化链路。
click 和 typer
入门先学 argparse 就够了。项目变复杂后,可以了解第三方库。
| 工具 | 特点 | 适合场景 |
|---|---|---|
| argparse | 标准库,无需安装 | 简单脚本、基础工具 |
| click | 装饰器风格,生态成熟 | 多命令工具 |
| typer | 基于类型标注,体验友好 | 希望自动生成帮助和校验 |
Typer 示例:
import typer
app = typer.Typer()
@app.command()
def hello(name: str, upper: bool = False):
text = f"你好,{name}"
if upper:
text = text.upper()
print(text)
if __name__ == "__main__":
app()运行:
python app.py hello 张三 --upper选型建议:团队脚本少、追求零依赖,用 argparse;需要很多子命令和更友好的开发体验,再考虑 Typer 或 click。
定时任务和 CI 中的 CLI
CLI 经常被定时任务调用。定时任务环境和你手动运行不一样:
| 差异 | 影响 |
|---|---|
| 当前工作目录不同 | 相对路径找不到文件 |
| 环境变量不同 | 数据库连接、Token 缺失 |
| Python 解释器不同 | 包导入失败 |
| 权限不同 | 文件无法读写 |
| 输出不可见 | 失败信息没人看到 |
所以生产脚本要做到:
- 路径尽量使用绝对路径或明确基准目录。
- 关键配置启动时校验。
- 输出日志到文件或日志系统。
- 返回正确退出码。
- 支持幂等,重复执行不会破坏数据。
常见问题排查
flowchart TD
A["CLI 执行失败"] --> B{"参数是否正确"}
B -->|否| C["查看 --help 和 argparse 错误"]
B -->|是| D{"文件路径是否存在"}
D -->|否| E["检查当前工作目录和路径引号"]
D -->|是| F{"是否权限问题"}
F -->|是| G["检查运行用户和目录读写权限"]
F -->|否| H{"是否编码问题"}
H -->|是| I["确认文件编码,统一 UTF-8"]
H -->|否| J{"定时任务环境不同"}
J -->|是| K["检查解释器、环境变量、工作目录"]
J -->|否| L["查看 stderr、日志和退出码"]常见问题表:
| 问题 | 原因 | 解决 |
|---|---|---|
| 路径里有空格 | Shell 把路径拆成多个参数 | 路径加引号 |
| 中文输出乱码 | 终端或文件编码不一致 | 文件使用 UTF-8,必要时调整终端编码 |
| 定时任务能启动但找不到文件 | 工作目录不是项目目录 | 使用绝对路径或配置基准目录 |
| 命令看起来失败但 CI 通过 | 没有返回非 0 退出码 | raise SystemExit(main()) |
| 输出被下游脚本解析失败 | 文本格式不稳定 | 提供 --json |
| 批量删除误删 | 参数传错且没有预览 | 增加 --dry-run 和二次确认 |
面试标准回答
argparse 的作用是什么?
标准回答:argparse 是 Python 标准库,用于解析命令行参数。它可以定义位置参数、选项参数、布尔开关、默认值、类型转换、枚举值和帮助信息。使用它可以让脚本支持标准的 --help,并在参数错误时给出清晰提示。
命令行程序为什么要返回退出码?
标准回答:退出码是 CLI 和调用方之间的结果协议。0 表示成功,非 0 表示失败。Shell、CI、定时任务可以根据退出码决定是否继续执行、是否报警。如果程序失败但仍返回 0,自动化系统会误判为成功。
stdout 和 stderr 有什么区别?
标准回答:stdout 用于输出正常结果,stderr 用于输出错误和诊断信息。这样用户可以把正常结果重定向到文件,同时把错误单独保存或显示,避免错误信息污染机器可读输出。
为什么危险操作要支持 dry-run?
标准回答:命令行工具经常批量处理文件、数据和任务,参数一旦传错影响范围很大。--dry-run 可以只预览将要处理的对象,不真正执行修改,降低误删、误更新、误发布的风险。
CLI 为什么要和业务逻辑分层?
标准回答:CLI 层应该负责参数解析、输出和退出码,业务逻辑放到 Service 或普通函数里。这样核心逻辑可以被单元测试、Web API、定时任务复用,不会和命令行输入输出耦合。
练习
- 写一个命令,统计某个目录下
.md文件数量。 - 给它加上
--recursive参数,支持递归统计。 - 给它加上
--output result.txt,把结果写入文件。 - 给它加上
--dry-run,只打印将要做什么。 - 给它加上
--json,输出机器可读结果。 - 把核心统计逻辑拆到单独函数,并写一个简单测试。
- 故意传错路径,观察退出码和错误输出。
关联知识点
- Python 基础语法:理解函数、条件、循环、入口函数。
- Python 异常与文件:理解文件读写、CSV、编码、异常传播。
- Python 类型标注:给 CLI 参数和业务结果建立类型契约。
- Python 测试:为 CLI 的核心逻辑和命令入口写测试。
- Python 项目实践:把 CLI 放入完整项目结构中。
小结
Python 命令行工具的核心不是“能在终端跑”,而是能被人、脚本、CI、定时任务稳定调用。一个合格的 CLI 要有清晰参数、帮助信息、退出码、错误输出、dry-run、日志、可测试的业务函数和明确的安全边界。
