568 Python CLI 工具开发：argparse 实战¶

# 改代码方式（每次都要编辑脚本）
# 1. 打开 filter_abundance.py
# 2. 找到第 15 行，把 input_file = "old_data.tsv" 改成 "new_data.tsv"
# 3. 运行 python filter_abundance.py

# CLI 工具方式（直接在终端用）
python filter_abundance.py --input new_data.tsv --threshold 0.01 --output filtered.tsv
# 甚至可以直接：
# filter_abundance --input new_data.tsv --threshold 0.01 --output filtered.tsv

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""最简单的 argparse 示例"""

import argparse  # 导入命令行参数解析模块

# 1. 创建解析器对象，description 是 --help 时显示的描述
parser = argparse.ArgumentParser(
    description="一个简单的生信 CLI 工具示例"
)

# 2. 添加参数
parser.add_argument("input_file", help="输入文件路径")  # 位置参数（必填，不加 --）

# 3. 解析参数
args = parser.parse_args()  # 把命令行输入的参数解析成对象

# 4. 使用参数
print(f"你输入的文件是：{args.input_file}")  # 通过 args.属性名 访问

$ python demo.py sample.fastq
你输入的文件是：sample.fastq

$ python demo.py --help
usage: demo.py [-h] input_file

一个简单的生信 CLI 工具示例

positional arguments:
  input_file  输入文件路径

options:
  -h, --help  show this help message and exit

import argparse

parser = argparse.ArgumentParser(description="参数类型示例")

# 位置参数（positional argument）：不加 -- 前缀，按顺序填写，必须提供
parser.add_argument("input_file", help="输入文件路径")

# 可选参数（optional argument）：加 -- 前缀，可以不提供
parser.add_argument("--output", "-o", help="输出文件路径")  # -o 是短选项
parser.add_argument("--threshold", "-t",
                    type=float,       # 参数类型，默认是字符串
                    default=0.01,     # 默认值
                    help="丰度过滤阈值（默认：0.01）")
parser.add_argument("--verbose", "-v",
                    action="store_true",  # 只要出现就是 True，不需要跟值
                    help="显示详细信息")

args = parser.parse_args()

import argparse

parser = argparse.ArgumentParser(description="互斥参数示例")

# 创建互斥组：组内参数只能选一个
group = parser.add_mutually_exclusive_group()
group.add_argument("--json", action="store_true", help="输出 JSON 格式")
group.add_argument("--csv", action="store_true", help="输出 CSV 格式")

args = parser.parse_args()
# 如果同时写 --json --csv，argparse 会自动报错

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
丰度表过滤工具 - 过滤低丰度物种
用法：python filter_abundance.py --input abundance.tsv --threshold 0.01 --output filtered.tsv
"""

import argparse  # 命令行参数解析
import sys       # 系统相关（用于退出码）
import csv       # CSV/TSV 文件读写


def parse_args():
    """解析命令行参数"""
    parser = argparse.ArgumentParser(
        description="过滤宏基因组丰度表中的低丰度物种",
        epilog="示例：python filter_abundance.py -i abundance.tsv -t 0.01 -o filtered.tsv"
    )

    # 必填参数
    parser.add_argument(
        "--input", "-i",
        required=True,           # 必须提供
        metavar="FILE",          # --help 中显示 FILE 而不是 INPUT
        help="输入丰度表文件（TSV 格式，第一列物种名，第二列丰度）"
    )

    # 可选参数，有默认值
    parser.add_argument(
        "--threshold", "-t",
        type=float,              # 自动转换为浮点数
        default=0.01,            # 默认阈值 1%
        help="丰度过滤阈值，低于此值的物种将被过滤（默认：0.01）"
    )

    parser.add_argument(
        "--output", "-o",
        default=None,            # 默认不指定，输出到终端
        metavar="FILE",
        help="输出文件路径（不指定则输出到终端）"
    )

    parser.add_argument(
        "--keep-header",
        action="store_true",     # 出现就是 True
        help="保留表头行"
    )

    return parser.parse_args()  # 返回解析后的参数对象


def filter_abundance(input_path, threshold, keep_header):
    """
    读取丰度表并过滤低丰度物种

    参数：
        input_path: 输入文件路径
        threshold: 丰度阈值
        keep_header: 是否保留表头

    返回：
        filtered_rows: 过滤后的行列表
        total_count: 总物种数
        kept_count: 保留的物种数
    """
    filtered_rows = []   # 存放过滤后的数据
    total_count = 0      # 总物种计数
    kept_count = 0       # 保留的物种计数
    header = None        # 表头

    # 打开文件，指定编码防止中文乱码
    with open(input_path, "r", encoding="utf-8") as f:
        reader = csv.reader(f, delimiter="\t")  # 用 tab 分隔读取

        for i, row in enumerate(reader):        # 逐行遍历
            if i == 0 and keep_header:           # 第一行且需要保留表头
                header = row
                continue

            if len(row) < 2:                     # 跳过格式不对的行
                continue

            species = row[0]                     # 第一列：物种名
            try:
                abundance = float(row[1])        # 第二列：丰度值（转浮点数）
            except ValueError:
                continue                         # 转换失败就跳过这行

            total_count += 1                     # 总数 +1

            if abundance >= threshold:           # 丰度大于等于阈值才保留
                filtered_rows.append(row)
                kept_count += 1                  # 保留数 +1

    return header, filtered_rows, total_count, kept_count


def main():
    """主函数：解析参数 → 过滤数据 → 输出结果"""
    args = parse_args()  # 获取命令行参数

    # 执行过滤
    header, filtered_rows, total, kept = filter_abundance(
        args.input,
        args.threshold,
        args.keep_header
    )

    # 统计信息输出到 stderr（不影响重定向）
    print(f"总物种数：{total}", file=sys.stderr)
    print(f"保留物种数：{kept}（阈值 >= {args.threshold}）", file=sys.stderr)
    print(f"过滤掉：{total - kept} 个低丰度物种", file=sys.stderr)

    # 决定输出目标：文件 or 终端
    if args.output:
        out_file = open(args.output, "w", encoding="utf-8", newline="")
    else:
        out_file = sys.stdout  # 不指定文件就输出到终端

    writer = csv.writer(out_file, delimiter="\t")  # TSV 格式写入

    if header:                    # 如果有表头，先写表头
        writer.writerow(header)

    for row in filtered_rows:     # 逐行写入过滤后的数据
        writer.writerow(row)

    if args.output:               # 如果写了文件，关闭文件
        out_file.close()
        print(f"结果已保存到：{args.output}", file=sys.stderr)


# 入口：只有直接运行此脚本时才执行 main()
if __name__ == "__main__":
    main()

# 基本用法
python filter_abundance.py -i abundance.tsv -t 0.01 -o filtered.tsv

# 保留表头
python filter_abundance.py -i abundance.tsv -t 0.005 -o filtered.tsv --keep-header

# 输出到终端（方便用管道继续处理）
python filter_abundance.py -i abundance.tsv -t 0.01 | head -5

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Kraken2 报告解析工具
用法：python parse_kraken2.py --report kraken2.report --level S --top 20
"""

import argparse   # 命令行参数解析
import csv         # CSV 读写
import json        # JSON 输出
import sys         # 系统相关


# Kraken2 报告中的分类级别代码对照表
LEVEL_MAP = {
    "D": "Domain",     # 域（如 Bacteria）
    "P": "Phylum",     # 门
    "C": "Class",      # 纲
    "O": "Order",      # 目
    "F": "Family",     # 科
    "G": "Genus",      # 属
    "S": "Species",    # 种
}


def parse_args():
    """解析命令行参数"""
    parser = argparse.ArgumentParser(
        description="解析 Kraken2 报告，提取指定分类级别的物种信息",
        formatter_class=argparse.RawDescriptionHelpFormatter,  # 保留 epilog 的换行格式
        epilog="""
分类级别代码：
  D - Domain（域）    P - Phylum（门）    C - Class（纲）
  O - Order（目）     F - Family（科）    G - Genus（属）
  S - Species（种）

示例：
  python parse_kraken2.py --report sample.report --level S --top 20
  python parse_kraken2.py -r sample.report -l G -n 10 --format csv -o result.csv
        """
    )

    parser.add_argument(
        "--report", "-r",
        required=True,
        metavar="FILE",
        help="Kraken2 报告文件路径"
    )

    parser.add_argument(
        "--level", "-l",
        default="S",
        choices=list(LEVEL_MAP.keys()),  # 只能选这些值
        help="分类级别代码（默认：S，即 Species 种）"
    )

    parser.add_argument(
        "--top", "-n",
        type=int,
        default=20,
        help="显示丰度最高的前 N 个物种（默认：20）"
    )

    parser.add_argument(
        "--format", "-f",
        default="table",
        choices=["table", "csv", "json"],  # 限定三种格式
        help="输出格式（默认：table）"
    )

    parser.add_argument(
        "--output", "-o",
        default=None,
        metavar="FILE",
        help="输出文件路径（不指定则输出到终端）"
    )

    parser.add_argument(
        "--min-reads",
        type=int,
        default=0,
        help="最低 reads 数过滤（默认：0，不过滤）"
    )

    return parser.parse_args()


def parse_kraken2_report(report_path, level, min_reads):
    """
    解析 Kraken2 报告文件

    Kraken2 报告格式（6 列，tab 分隔）：
    1. 该分类占总 reads 的百分比
    2. 该分类及子分类的 reads 数
    3. 仅该分类的 reads 数
    4. 分类级别代码（如 S, G, F）
    5. NCBI TaxID
    6. 物种名（前面有缩进空格表示层级）
    """
    results = []  # 存放解析结果

    with open(report_path, "r", encoding="utf-8") as f:
        for line in f:
            line = line.rstrip("\n")       # 去掉行尾换行符
            parts = line.split("\t")       # 按 tab 分割

            if len(parts) < 6:             # 不够 6 列就跳过
                continue

            percentage = float(parts[0].strip())   # 第 1 列：百分比
            reads_clade = int(parts[1].strip())     # 第 2 列：该分类及子分类 reads
            reads_taxon = int(parts[2].strip())     # 第 3 列：仅该分类 reads
            rank_code = parts[3].strip()            # 第 4 列：分类级别代码
            taxid = parts[4].strip()                # 第 5 列：TaxID
            name = parts[5].strip()                 # 第 6 列：物种名（去掉前后空格）

            # 只保留指定级别的结果
            if rank_code != level:
                continue

            # 过滤低 reads 数的结果
            if reads_clade < min_reads:
                continue

            results.append({
                "name": name,
                "percentage": percentage,
                "reads_clade": reads_clade,
                "reads_taxon": reads_taxon,
                "taxid": taxid,
            })

    # 按百分比从大到小排序
    results.sort(key=lambda x: x["percentage"], reverse=True)

    return results


def output_table(results, out_file):
    """以表格格式输出结果"""
    # 表头
    header = f"{'排名':<5}{'物种名':<40}{'丰度%':<10}{'Reads数':<12}{'TaxID':<12}"
    print(header, file=out_file)
    print("-" * len(header), file=out_file)  # 分隔线

    for i, item in enumerate(results, 1):    # 从 1 开始编号
        line = f"{i:<5}{item['name']:<40}{item['percentage']:<10.4f}{item['reads_clade']:<12}{item['taxid']:<12}"
        print(line, file=out_file)


def output_csv(results, out_file):
    """以 CSV 格式输出结果"""
    writer = csv.writer(out_file)
    writer.writerow(["rank", "name", "percentage", "reads_clade", "reads_taxon", "taxid"])

    for i, item in enumerate(results, 1):
        writer.writerow([
            i,
            item["name"],
            item["percentage"],
            item["reads_clade"],
            item["reads_taxon"],
            item["taxid"],
        ])


def output_json(results, out_file):
    """以 JSON 格式输出结果"""
    # ensure_ascii=False 让中文正常显示，indent=2 让格式好看
    json.dump(results, out_file, ensure_ascii=False, indent=2)
    print(file=out_file)  # 末尾加换行


def main():
    """主函数"""
    args = parse_args()

    # 解析报告
    results = parse_kraken2_report(args.report, args.level, args.min_reads)

    # 截取 top N
    results = results[:args.top]

    # 输出统计信息到 stderr
    level_name = LEVEL_MAP.get(args.level, args.level)
    print(f"分类级别：{args.level}（{level_name}）", file=sys.stderr)
    print(f"显示前 {args.top} 个结果", file=sys.stderr)

    # 决定输出目标
    if args.output:
        out_file = open(args.output, "w", encoding="utf-8", newline="")
    else:
        out_file = sys.stdout

    # 根据格式选择输出函数
    if args.format == "table":
        output_table(results, out_file)
    elif args.format == "csv":
        output_csv(results, out_file)
    elif args.format == "json":
        output_json(results, out_file)

    if args.output:
        out_file.close()
        print(f"结果已保存到：{args.output}", file=sys.stderr)


if __name__ == "__main__":
    main()

# 查看种级别 top 20（默认）
python parse_kraken2.py --report sample.report

# 查看属级别 top 10，CSV 格式输出
python parse_kraken2.py -r sample.report -l G -n 10 -f csv -o genus_top10.csv

# JSON 格式 + 过滤低 reads
python parse_kraken2.py -r sample.report -l S -n 50 -f json --min-reads 100

# 查看帮助
python parse_kraken2.py --help

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""丰度过滤工具 - Click 版本"""

import click  # 需要先 pip install click
import csv    # CSV 读写
import sys    # 系统相关


@click.command()  # 把函数变成 CLI 命令
@click.option("--input", "-i", "input_file",  # input_file 是变量名（避免和内置 input 冲突）
              required=True,
              type=click.Path(exists=True),    # 自动检查文件是否存在！
              help="输入丰度表文件（TSV 格式）")
@click.option("--threshold", "-t",
              type=float,
              default=0.01,
              show_default=True,               # 在 --help 中自动显示默认值
              help="丰度过滤阈值")
@click.option("--output", "-o",
              type=click.Path(),               # 输出路径（不检查是否存在）
              default=None,
              help="输出文件路径")
def filter_abundance(input_file, threshold, output):
    """过滤宏基因组丰度表中的低丰度物种。"""  # 这个 docstring 会自动变成 --help 的描述
    kept = 0       # 保留计数
    total = 0      # 总数计数
    rows = []      # 存放结果

    with open(input_file, "r", encoding="utf-8") as f:
        reader = csv.reader(f, delimiter="\t")
        for row in reader:
            if len(row) < 2:
                continue
            try:
                abundance = float(row[1])
            except ValueError:
                continue
            total += 1
            if abundance >= threshold:
                rows.append(row)
                kept += 1

    # click.echo 是 Click 推荐的 print 替代，兼容性更好
    click.echo(f"保留 {kept}/{total} 个物种", err=True)

    if output:
        with open(output, "w", encoding="utf-8", newline="") as f:
            writer = csv.writer(f, delimiter="\t")
            for row in rows:
                writer.writerow(row)
        click.echo(f"结果已保存到：{output}", err=True)
    else:
        for row in rows:
            click.echo("\t".join(row))


if __name__ == "__main__":
    filter_abundance()  # Click 会自动处理参数解析

pip install click  # 安装 Click 框架（当前最新版 8.3.3）

def main():
    """工具的主逻辑"""
    # ... 你的代码 ...

# 只有直接运行时才执行，被 import 时不执行
if __name__ == "__main__":
    main()

# 第一步：确保脚本第一行是 shebang
# #!/usr/bin/env python3   ← 这行就是 shebang

# 第二步：给脚本加上可执行权限
chmod +x filter_abundance.py  # 让脚本可以直接运行

# 第三步：直接运行（不需要在前面写 python）
./filter_abundance.py --input data.tsv --threshold 0.01

# 方法 1：创建个人 bin 目录
mkdir -p ~/bin                          # 创建 bin 目录（如果不存在）
cp filter_abundance.py ~/bin/filter_abundance  # 复制脚本（去掉 .py 后缀更专业）
chmod +x ~/bin/filter_abundance         # 加可执行权限

# 把 ~/bin 加入 PATH（写入 .bashrc 永久生效）
echo 'export PATH="$HOME/bin:$PATH"' >> ~/.bashrc
source ~/.bashrc                        # 立即生效

# 现在可以在任何目录直接使用
filter_abundance --input data.tsv --threshold 0.01

# 方法 2：用 pip install -e . 安装（见知识库 569 包发布篇）

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""带日志功能的 CLI 工具示例"""

import argparse  # 命令行参数解析
import logging   # 日志模块
import sys       # 系统相关


def setup_logging(verbose):
    """
    设置日志系统

    参数：
        verbose: 是否开启详细模式（True 显示 DEBUG 级别信息）
    """
    # 根据 verbose 参数决定日志级别
    level = logging.DEBUG if verbose else logging.INFO

    logging.basicConfig(
        level=level,                           # 日志级别
        format="%(asctime)s [%(levelname)s] %(message)s",  # 日志格式
        datefmt="%Y-%m-%d %H:%M:%S",          # 时间格式
        stream=sys.stderr,                     # 输出到 stderr，不影响数据重定向
    )


def main():
    parser = argparse.ArgumentParser(description="带日志的生信工具示例")
    parser.add_argument("--input", "-i", required=True, help="输入文件")
    parser.add_argument("--verbose", "-v", action="store_true",
                        help="显示详细调试信息")
    args = parser.parse_args()

    # 初始化日志系统
    setup_logging(args.verbose)

    # 获取 logger 对象
    logger = logging.getLogger(__name__)

    logger.info("开始处理文件：%s", args.input)       # INFO 级别：总是显示
    logger.debug("打开文件中...")                      # DEBUG 级别：只有 -v 时显示
    logger.debug("读取完成，共 1000 行")
    logger.warning("发现 5 行格式异常，已跳过")        # WARNING 级别：总是显示
    logger.info("处理完成")


if __name__ == "__main__":
    main()

# 普通模式：只显示 INFO 和 WARNING
$ python tool.py -i data.tsv
2026-05-09 10:30:00 [INFO] 开始处理文件：data.tsv
2026-05-09 10:30:01 [WARNING] 发现 5 行格式异常，已跳过
2026-05-09 10:30:01 [INFO] 处理完成

# 详细模式：额外显示 DEBUG
$ python tool.py -i data.tsv -v
2026-05-09 10:30:00 [INFO] 开始处理文件：data.tsv
2026-05-09 10:30:00 [DEBUG] 打开文件中...
2026-05-09 10:30:01 [DEBUG] 读取完成，共 1000 行
2026-05-09 10:30:01 [WARNING] 发现 5 行格式异常，已跳过
2026-05-09 10:30:01 [INFO] 处理完成

error: unrecognized arguments: --input

error: the following arguments are required: --input

error: argument --threshold: invalid float value: 'abc'

FileNotFoundError: [Errno 2] No such file or directory: 'data.tsv'

import os
if not os.path.exists(args.input):
    print(f"错误：文件 {args.input} 不存在", file=sys.stderr)
    sys.exit(1)  # 以错误码退出

#!/usr/bin/env python3
"""工具描述"""
import argparse

def main():
    parser = argparse.ArgumentParser(description="工具描述")
    parser.add_argument("--input", "-i", required=True, help="输入文件")
    parser.add_argument("--output", "-o", default="out.tsv", help="输出文件")
    parser.add_argument("--threshold", "-t", type=float, default=0.01, help="阈值")
    parser.add_argument("--verbose", "-v", action="store_true", help="详细模式")
    args = parser.parse_args()
    # 你的逻辑写在这里

if __name__ == "__main__":
    main()