Python 实用工具：Cleo 库深度解析与实战指南

pip install cleo

from cleo import Application, Command

class HelloCommand(Command):
    name = "hello"  # 命令名称
    description = "Print a greeting message"  # 命令描述

    def handle(self):
        # 使用 output 对象输出内容，支持颜色和样式
        self.line("<info>Hello, World!</info>")
        self.line("This is a Cleo-powered CLI tool.")

if __name__ == "__main__":
    app = Application(name="my_cli", version="1.0.0")  # 创建应用程序实例
    app.add(HelloCommand())  # 注册命令
    app.run()  # 启动应用程序

python my_script.py hello

class GreetCommand(Command):
    name = "greet"
    description = "Greets a person by name"

    def configure(self):
        # 定义位置参数：name（必填），age（可选，默认值为 18）
        self.add_argument("name", description="The person's name")
        self.add_argument("age", description="The person's age", default=18, optional=True)

    def handle(self):
        name = self.argument("name")
        age = self.argument("age")
        self.line(f"Hello, <comment>{name}</comment>! You are <fg=green>{age}</fg=green> years old.")

# 传递必填参数和可选参数
python my_script.py greet Alice 25
# 仅传递必填参数（age 使用默认值）
python my_script.py greet Bob

Hello, Alice! You are 25 years old.
Hello, Bob! You are 18 years old.

class ListFilesCommand(Command):
    name = "ls"
    description = "List files in a directory"

    def configure(self):
        # 添加选项：--path（默认值为当前目录），-v/--verbose 显示详细信息
        self.add_option(
            "path",
            "p",  # 短选项
            description="Directory path",
            default=".",
            value_required=True  # 选项需要值
        )
        self.add_option(
            "verbose",
            "v",
            description="Show detailed information",
            action="store_true"  # 标记选项（无值，存在即 True）
        )

    def handle(self):
        path = self.option("path")
        verbose = self.option("verbose")
        files = os.listdir(path)  # 简化的文件列表获取逻辑

        if verbose:
            self.line(f"Listing files in <fg=cyan>{path}</fg=cyan> (verbose mode):")
            for file in files:
                file_size = os.path.getsize(os.path.join(path, file))
                self.line(f"- <fg=green>{file}</fg=green> ({file_size} bytes)")
        else:
            self.line(f"Files in <fg=cyan>{path}</fg=cyan>:")
            self.line(", ".join(files))

# 常规模式
python my_script.py ls --path ./docs
# 详细模式
python my_script.py ls -v -p ./src

Listing files in ./src (verbose mode):
- main.py (4521 bytes)
- utils.py (2894 bytes)
- config.json (128 bytes)

# 尝试调用时不传递 name 参数
python my_script.py greet

 [Error] The "greet" command requires that you provide a value for the "name" argument.

self.line("<fg=magenta bg=white bold>WARNING:</bg=white></fg=magenta> This is a test message.")
self.line("<error>Operation failed! Please check the logs.</error>")

from cleo.formatters.table import Table, Style

class UsersCommand(Command):
    name = "users:list"
    description = "List registered users"

    def handle(self):
        users = [
            {"id": 1, "name": "Alice", "email": "[email protected]"},
            {"id": 2, "name": "Bob", "email": "[email protected]"},
            {"id": 3, "name": "Charlie", "email": "[email protected]"}
        ]

        table = Table(self.io)
        table.set_style(Style().set_header_color("cyan").set_border_color("magenta"))  # 设置表格样式
        table.set_header_row(["ID", "Name", "Email"])  # 设置表头

        for user in users:
            table.add_row([str(user["id"]), user["name"], user["email"]])

        table.render()  # 渲染表格

 ╒════╤════════╤══════════════════╕
 │ ID │ Name   │ Email            │
 ╞════╪════════╪══════════════════╡
 │ 1  │ Alice  │ [email protected]│
 ├────┼────────┼──────────────────┤
 │ 2  │ Bob    │ [email protected]  │
 ├────┼────────┼──────────────────┤
 │ 3  │ Charlie│ [email protected]│
 ╘════╧════════╧══════════════════╛

def handle(self):
    name = self.ask("What is your name?", default="Guest")  # 带默认值的提问
    age = self.ask("How old are you?", type=int)  # 类型验证（仅允许输入整数）
    confirm = self.confirm(f"Confirm user: {name} ({age} years old)?", default=True)  # 确认提问

    if confirm:
        self.line("<info>User confirmed.</info>")
    else:
        self.line("<error>Operation cancelled.</error>")

What is your name? (Guest) Alice
How old are you? 25
Confirm user: Alice (25 years old)? [y/n] y
User confirmed.

from cleo import Command
import logging

class BaseCommand(Command):
    def __init__(self):
        super().__init__()
        self.logger = logging.getLogger(self.name)  # 根据命令名创建日志器

    def configure(self):
        # 添加公共选项：--debug 开启调试日志
        self.add_option("debug", None, description="Enable debug mode", action="store_true")

    def handle(self):
        if self.option("debug"):
            logging.basicConfig(level=logging.DEBUG)
            self.logger.debug("Debug mode enabled")
        # 其他公共逻辑...

class DatabaseCommand(BaseCommand):
    name = "db:connect"
    description = "Connect to the database"

    def configure(self):
        super().configure()  # 继承父类配置
        # 添加子类特有的参数和选项
        self.add_argument("host", description="Database host")
        self.add_option("port", "p", description="Database port", default=3306)

    def handle(self):
        super().handle()  # 执行父类逻辑（如调试日志）
        host = self.argument("host")
        port = self.option("port")
        self.line(f"Connecting to database at <fg=green>{host}:{port}</fg=green>...")
        # 数据库连接逻辑...

app = Application()
database_group = app.create_group("database", "Database-related commands")
database_group.add(DatabaseCommand())
database_group.add(AnotherDatabaseCommand())  # 添加其他数据库命令

Usage:
  command [options] [arguments]

Groups:
  database    Database-related commands
  system      System management commands

import os
from datetime import datetime
from cleo import Command

class FileStatsCommand(Command):
    name = "stats"
    description = "Show file or directory statistics"

    def configure(self):
        self.add_argument("path", description="File or directory path")

    def handle(self):
        path = self.argument("path")
        if not os.path.exists(path):
            self.line(f"<error>{path} does not exist.</error>")
            return 1

        stats = os.stat(path)
        is_dir = os.path.isdir(path)
        created_time = datetime.fromtimestamp(stats.st_ctime).strftime("%Y-%m-%d %H:%M:%S")
        modified_time = datetime.fromtimestamp(stats.st_mtime).strftime("%Y-%m-%d %H:%M:%S")

        self.line(f"<bold>File/Directory: </bold>{path}")
        self.line(f"<comment>Type: </comment>{'Directory' if is_dir else 'File'}")
        self.line(f"<comment>Size: </comment>{stats.st_size} bytes")
        self.line(f"<comment>Created: </comment>{created_time}")
        self.line(f"<comment>Modified: </comment>{modified_time}")

import shutil
import glob
from cleo import Command

class FileCopyCommand(Command):
    name = "copy"
    description = "Copy files or directories"

    def configure(self):
        self.add_argument("src", description="Source file/directory or glob pattern")
        self.add_argument("dest", description="Destination path")
        self.add_option(
            "recursive",
            "r",
            description="Copy directories recursively",
            action="store_true"
        )
        self.add_option(
            "force",
            "f",
            description="Overwrite existing files",
            action="store_true"
        )

    def handle(self):
        src = self.argument("src")
        dest = self.argument("dest")
        recursive = self.option("recursive")
        force = self.option("force")

        # 处理通配符路径
        sources = glob.glob(src, recursive=recursive)
        if not sources:
            self.line(f"<error>No files matching pattern: {src}</error>")
            return 1

        for source in sources:
            try:
                if os.path.isdir(source):
                    if not recursive:
                        self.line(f"<warning>Skipping directory {source} (use -r to copy recursively)</warning>")
                        continue
                    # 复制目录
                    dest_path = os.path.join(dest, os.path.basename(source))
                    if os.path.exists(dest_path) and not force:
                        self.line(f"<warning>Directory {dest_path} exists (use -f to overwrite)</warning>")
                        continue
                    shutil.copytree(source, dest_path, dirs_exist_ok=force)
                    self.line(f"<info>Copied directory: {source} -> {dest_path}</info>")
                else:
                    # 复制文件
                    dest_file = os.path.join(dest, os.path.basename(source)) if os.path.isdir(dest) else dest
                    if os.path.exists(dest_file) and not force:
                        self.line(f"<warning>File {dest_file} exists (use -f to overwrite)</warning>")
                        continue
                    shutil.copy2(source, dest_file)  # 保留元数据
                    self.line(f"<info>Copied file: {source} -> {dest_file}</info>")
            except Exception as e:
                self.line(f"<error>Failed to copy {source}: {str(e)}</error>")

import os
import glob
from cleo import Command

class FileCleanCommand(Command):
    name = "clean"
    description = "Clean temporary files in directory"

    def configure(self):
        self.add_argument("dir", description="Directory to clean")
        self.add_option(
            "patterns",
            "p",
            description="File patterns to delete (comma-separated)",
            default="*.tmp,*~"  # 默认清理.tmp文件和~结尾文件
        )
        self.add_option(
            "dry-run",
            None,
            description="Show what would be deleted without actual removal",
            action="store_true"
        )
        self.add_option(
            "confirm",
            "c",
            description="Ask for confirmation before deletion",
            action="store_true"
        )

    def handle(self):
        target_dir = self.argument("dir")
        if not os.path.isdir(target_dir):
            self.line(f"<error>{target_dir} is not a valid directory</error>")
            return 1

        patterns = self.option("patterns").split(",")
        dry_run = self.option("dry-run")
        confirm = self.option("confirm")
        files_to_delete = []

        # 收集匹配的文件
        for pattern in patterns:
            pattern_path = os.path.join(target_dir, pattern)
            files_to_delete.extend(glob.glob(pattern_path))

        if not files_to_delete:
            self.line("<info>No files matching cleanup patterns found</info>")
            return 0

        # 显示待删除文件
        self.line(f"<comment>Found {len(files_to_delete)} files to delete:</comment>")
        for file in files_to_delete:
            self.line(f"- {file}")

        # 确认流程
        if confirm:
            if not self.confirm("Proceed with deletion?", default=False):
                self.line("<info>Deletion cancelled</info>")
                return 0

        # 执行删除
        deleted = 0
        for file in files_to_delete:
            try:
                if dry_run:
                    self.line(f"<info>[Dry run] Would delete: {file}</info>")
                else:
                    os.remove(file)
                    self.line(f"<info>Deleted: {file}</info>")
                    deleted += 1
            except Exception as e:
                self.line(f"<error>Failed to delete {file}: {str(e)}</error>")

        self.line(f"\n<comment>Summary: {deleted}/{len(files_to_delete)} files processed</comment>")

from cleo import Application
from commands.stats import FileStatsCommand
from commands.copy import FileCopyCommand
from commands.clean import FileCleanCommand

if __name__ == "__main__":
    app = Application(name="file_tool", version="1.0.0")
    app.add(FileStatsCommand())
    app.add(FileCopyCommand())
    app.add(FileCleanCommand())
    app.run()

# pyproject.toml 示例（使用poetry）

file_tool stats ./docs
file_tool copy ./images/*.png ./backup -f
file_tool clean ./tmp -p "*.log,*.bak" -c

from cleo.events import EventDispatcher, ConsoleCommandEvent, ConsoleEvents

def before_command(event: ConsoleCommandEvent):
    command = event.command
    if command.name in ["db:connect", "db:migrate"]:
        # 数据库命令权限验证
        if not has_db_access():
            event.exit_code = 1
            event.io.write_line("<error>Permission denied: DB access required</error>")

dispatcher = EventDispatcher()
dispatcher.add_listener(ConsoleEvents.COMMAND, before_command)
app = Application(event_dispatcher=dispatcher)

# 生成bash补全脚本
file_tool completion bash > /etc/bash_completion.d/file_tool
# 生成zsh补全脚本
file_tool completion zsh > ~/.zsh/completions/_file_tool

from cleo.testing import CommandTester

def test_greet_command():
    command = GreetCommand()
    tester = CommandTester(command)
    tester.execute("Alice 30")
    assert "Hello, Alice! You are 30 years old." in tester.io.fetch_output()