ARTICLE DETAIL

资讯详情

深耕编程入门与网站建设的一线实战洞察。

CLI-Anything:命令行工作流引擎打造自动化终端体验

CLI-Anything:命令行工作流引擎打造自动化终端体验 你有没有过这种感觉每天打开电脑真正花在“干活”上的时间没多少反而把大量时间耗在切窗口、找命令、改参数、复制粘贴这些琐碎操作上。我有一段时间被这种碎片化工作折磨得够呛痛定思痛之后做了个决定把几乎所有重复性任务都收敛到命令行里用一个统一的框架来调度它们。这个项目最后被我命名为CLI-Anything意思很直白——只要是能用命令行做的事就不用手工一步步去点。它不是一个单一功能的工具而是一个命令行工作流引擎你可以往里面自由添加命令、挂插件、调 API甚至让大模型直接替你执行复杂的终端操作。CLI-Anything 适合谁如果你是开发、运维、数据分析或者任何日常需要和终端打交道的人被“一堆脚本谁也不服谁”的局面困扰过那么这篇文章值得看完。我会把它从立项到落地的完整思路、核心代码、真实坑点全部整理出来包括很多硬啃文档未必能发现的细节。全文不会出现空泛的架构图只有能直接抄走的代码和配置。1. 先聊聊为什么要做 CLI-Anything1.1 终端老兵的痛点命令太散心智负担太重做技术的人多少都有这种经历自己电脑上没少存脚本有 Shell、有 Python可能还有几段 Ruby 或者 Perl 时代的遗产。每个脚本的用法都不一样有的要改文件开头的变量有的通过环境变量传参有的只接收固定参数。时间一长你根本记不清convert.py和batch_convert.py到底哪个是生效版本更别说别人接手你机器时的那种崩溃感。我当时的日常是早上要拉代码、看昨天的构建结果、写日报中午要批量处理图片、压缩资源、转格式晚上要跑数据统计、发通知、归档日志。每次做这些事情都在不同的工具、不同的 web 页面、不同的配置项之间来回切换。真正用来做决策和输出的时间极少大量精力耗在了“怎么把命令拼对”上。CLI-Anything 想解决的正是这个问题。它不做具体业务而是提供一个统一入口只要在终端里敲cli-anything do-something剩下的事情由框架去调度插件、读取配置、处理输出。你不需要记住每个脚本的独有参数所有命令的用法都由--help统一说明。1.2 从“脚本堆”到“命令框架”的转变有人会问那我写一个大 Shell 脚本把所有事情包起来不就行了我试过不可行。脚本一旦变大参数解析、错误处理、输出格式、并发调度全都耦合在一起改一处崩三处。尤其当我想让某个命令支持“今天只处理三天内的文件”这种运行时过滤条件时Shell 脚本的代码量会迅速失控。CLI-Anything 采用的是插件化设计。每个功能都是一条独立命令插件框架层统一负责四件事参数解析、配置注入、日志输出、权限确认。插件作者只需要实现一个函数返回一个结果其余都不用管。这样设计的好处非常明显新增功能只需新增一个文件不需要改动框架核心。删除或禁用某个功能只需改配置或删目录。所有命令的交互方式一致cli-anything name [args]上手成本极低。后续可以支持热加载插件的更新不需要重启整个框架。一句话总结CLI-Anything 不是又一个脚本合集而是一个把“散装命令”变成“统一命令体系”的容器。2. 整体架构与设计思路2.1 核心模块拆解一次命令执行的生命周期很多工具一上来就画复杂架构图实际根本没那个必要。CLI-Anything 的核心模块只有六个每个职责单一没有过度抽象。模块职责关键设计点入口模块读取用户输入分发到命令注册表只做参数预解析不掺业务逻辑注册表扫描插件目录建立命令名到处理函数的映射支持按目录、按文件名前缀过滤配置加载器读取 YAML 配置合并环境变量配置分层全局、插件级、运行时级执行器调用插件处理函数管理超时和取消支持阻塞执行和流式输出两种模式通知器把结果发送到 webhook、邮件或本地通知失败自动重试可单独禁用调度器按 cron 表达式触发命令防止同一条命令并发重入一次命令执行的生命周期大致是用户在终端输入cli-anything demo --flag→ 入口模块解析出命令名demo→ 注册表查找到对应插件 → 配置加载器把 YAML 配置和命令行参数合并 → 执行器调用插件处理函数 → 插件返回结果 → 通知器按规则发送通知。整个过程中插件不感知框架细节框架也不关心插件内部怎么实现。2.2 为什么选 YAML 做配置而不是 JSON / TOML这是我实际对比过才定的。JSON 写起来啰嗦而且不能加注释——只要配置超过三行没人能记住每个字段的含义。TOML 虽然规范但在表达层级嵌套结构时体验一般。YAML 天然适合“命令定义插件参数”这个场景它允许注释、支持多行字符串、层级关系用缩进就能表达可读性远超其他格式。下面是我实际在用的一个配置片段# config.yaml global: log_level: info timeout: 120 dry_run: false plugins: report: enabled: true schedule: 0 9 * * 5 recipients: - opsexample.com ai_model: qwen-plus image: enabled: true default_quality: 85 output_format: webp这里有几个细节enabled字段用来控制插件是否加载schedule字段由调度器解析ai_model是报告插件要用的模型名。配置里不能写死任何密钥一律用环境变量引用比如api_key: ${LLM_API_KEY}加载的时候替换成真实值。这样配置文件可以放进 Git不用担心密钥泄露。2.3 插件协议怎么定极简但够用插件协议是整个框架的“宪法”定得太重没人愿意写插件定得太轻又没法满足不同需求。CLI-Anything 的约定是每个插件就是一个普通 Python 文件放在plugins/目录下文件里定义一个register()函数即可。# plugins/demo.py from cli_anything import CommandMeta def handler(ctx): 具体的业务逻辑 name ctx.args.get(name, world) return {message: fhello, {name}} def register() - CommandMeta: return CommandMeta( namedemo, description一个示例插件, handlerhandler, )就这么多。CommandMeta里还可以带上arguments列表用来声明该命令支持哪些参数。框架启动时扫描plugins/目录对每个文件执行register()然后把命令名注册到映射表里。我还留了两个钩子pre_hook和post_hook分别在执行前和执行后触发。比如你想在每次执行命令前检查磁盘空间或者在执行后自动把结果同步到某个系统就可以写一个公共插件把这些逻辑挂到全局钩子上不需要每个插件自己去实现。3. 核心实现把“Anything”变成可运行的命令3.1 命令注册机制用装饰器干掉样板代码如果每个插件都手写register()函数写多了还是会觉得重复。所以我封装了一个装饰器command插件的代码量进一步减少# cli_anything/decorators.py import functools from .registry import registry def command(name, descriptionNone): def decorator(func): registry.register( namename, descriptiondescription or func.__doc__, handlerfunc, ) functools.wraps(func) def wrapper(*args, **kwargs): return func(*args, **kwargs) return wrapper return decorator # plugins/report.py from cli_anything.decorators import command command(namereport, description生成工作周报) def report(ctx): # 业务逻辑 return {status: ok}装饰器的好处是直观插件作者定义函数的时候顺手就完成了注册不用记忆额外的初始化流程。而且func.__doc__可以直接作为命令描述写文档的成本几乎为零。框架启动时注册表会对所有扫描到的函数做校验如果发现命令名重复直接在启动阶段报错而不是等到运行的时候才报“command not found”。3.2 配置加载与变量注入安全第一灵活第二配置加载器要做的事情不只是yaml.safe_load()。我在实现时加了几个关键环节# cli_anything/config.py import os import re import yaml _ENV_PATTERN re.compile(r\$\{([A-Za-z_][A-Za-z0-9_]*)\}) def _resolve_env(value: str) - str: if isinstance(value, str): def replace(match): env_name match.group(1) if env_name not in os.environ: raise RuntimeError(f环境变量 {env_name} 未设置) return os.environ[env_name] return _ENV_PATTERN.sub(replace, value) return value def load_config(path: str, overrides: dict | None None): with open(path, r, encodingutf-8) as f: raw yaml.safe_load(f) # 深度遍历把所有 ${ENV_NAME} 换成真实环境变量 resolved _deep_map(raw, _resolve_env) if overrides: resolved _deep_merge(resolved, overrides) return resolved注意我用的是yaml.safe_load()而不是yaml.load()这能避免恶意 YAML 触发任意代码执行。第二个细节是环境变量缺失时直接抛异常而不是静默替换成空字符串——因为配置里缺一个 key 可能意味着线上事故。第三个是支持命令行参数覆盖配置文件的同名 key举个例子cli-anything report --dry-run会临时把global.dry_run改成True适合正式执行前先演练一遍。3.3 AI 调用模块流式输出、超时和重试一个都不能少CLI-Anything 里最受欢迎的功能是让大语言模型去处理那些“说不清但很费时间”的杂活。比如把一段混乱的日志整理成结构化的摘要或者根据 git 提交记录生成周报。这里涉及一个通用的 LLM 调用模块我把它单独抽了出来# cli_anything/llm.py import json import os import time import requests class LLMClient: def __init__(self, model: str, api_key: str, base_url: str): self.model model self.api_key api_key self.base_url base_url def chat(self, messages: list[dict], stream: bool True, temperature: float 0.2): url f{self.base_url}/chat/completions headers { Authorization: fBearer {self.api_key}, Content-Type: application/json, } payload { model: self.model, messages: messages, temperature: temperature, stream: stream, } # 核心逻辑支持重试、超时、流式输出 for attempt in range(3): try: resp requests.post(url, jsonpayload, headersheaders, timeout(10, 120), streamstream) resp.raise_for_status() if not stream: return resp.json()[choices][0][message][content] return self._read_stream(resp) except (requests.exceptions.Timeout, requests.exceptions.ConnectionError): wait 2 ** attempt print(f请求失败{wait}s 后重试..., filesys.stderr) time.sleep(wait) raise RuntimeError(LLM 调用在重试 3 次后仍然失败) def _read_stream(self, resp): buffer for line in resp.iter_lines(): if not line: continue line line.decode(utf-8) if line.startswith(data: ): data line[6:] if data [DONE]: break try: chunk json.loads(data) delta chunk[choices][0][delta].get(content, ) if delta: buffer delta print(delta, end, flushTrue) except json.JSONDecodeError: continue print() return buffer流式输出这里有一个关键细节Python 的print默认带缓冲如果不加flushTrue在终端里看到的效果就是字数一段一段往外蹦体验非常差。所以我统一用print(..., end, flushTrue)强制实时刷新。超时设置也值得注意我用了(10, 120)这种连接超时和读取超时分开的写法——连接 10 秒读取 120 秒比一锤子买卖的单一超时合理得多。3.4 定时任务与通知让命令自己跑起来有些命令不该由人手动敲比如每周五上午生成本周报告并发送到群里。CLI-Anything 接了一个轻量调度器解析 cron 表达式到点就触发对应的命令。# cli_anything/scheduler.py import threading import time class Scheduler: def __init__(self): self._jobs [] self._lock threading.Lock() def every(self, cron_expr: str, command_name: str, args: dict | None None): # 这里用 croniter 或 cron 解析库省略实现 self._jobs.append({ cron: cron_expr, command: command_name, args: args or {}, last_run: None, }) def run(self): while True: now time.localtime() with self._lock: for job in self._jobs: if self._match_cron(job[cron], now) and self._can_run(job, now): thread threading.Thread(targetself._run_job, args(job,)) thread.start() time.sleep(30) def _can_run(self, job, now): if job[last_run] is None: return True # 避免同一分钟内重复执行 run_key f{now.tm_min}_{now.tm_hour}_{now.tm_mday}_{now.tm_mon}_{now.tm_wday} last_key f{job[last_run].tm_min}_{job[last_run].tm_hour}_{job[last_run].tm_mday}_{job[last_run].tm_mon}_{job[last_run].tm_wday} return run_key ! last_key通知部分就更简单了钉钉、企业微信、飞书这类平台都有现成的 webhook 接口本质就是发起一个 POST JSON 请求。我封装了一个Notifier类执行器在命令成功后自动推送结果摘要。如果命令失败通知器会标记statusfailed并附带前 200 个字符的错误日志方便第一时间定位问题。4. 实操案例三条命令解决日常三大场景4.1 一键生成项目周报周报是每个团队都绕不开的痛点。用 CLI-Anything 实现之后我每周五早上只需要敲一条命令cli-anything report --week插件内部的执行逻辑是先调用git log --since7.days.ago收集本周所有提交记录把每次提交的 message、作者、日期整理成文本再拼进 prompt 里发给 LLM最后让模型输出一段结构化的工作总结追加到本地weekly-report.md同时通过 webhook 发到团队群。这里有个细节提交信息往往又乱又长直接塞给模型会浪费 token 还会跑偏。我做了个预处理把 commit message 里的“wip”“fix typo”这类高频低价值内容过滤掉再按模块归类。Prompt 里也强调“只输出纯文本不要 Markdown控制在 500 字以内按 123 列出工作项”。实测下来报告质量比很多同事手写的还规整。4.2 批量图片压缩与格式转换运营或者前端同学经常遇到要批量处理图片的需求。以前我得一个个打开 Photoshop后来我用 CLI-Anything 写了一个图片插件核心代码大概长这样# plugins/image.py from pathlib import Path from concurrent.futures import ThreadPoolProcessor from PIL import Image from cli_anything.decorators import command command(nameimg, description图片批量压缩/转换) def image_compress(ctx): src_dir Path(ctx.args.get(src, images)) dst_dir Path(ctx.args.get(dst, compressed)) quality int(ctx.args.get(quality, 85)) fmt ctx.args.get(format, webp) dst_dir.mkdir(parentsTrue, exist_okTrue) files [f for f in src_dir.iterdir() if f.suffix.lower() in {.png, .jpg, .jpeg}] def process_one(path: Path): out dst_dir / path.with_suffix(. fmt).name with Image.open(path) as img: # 先处理 EXIF 旋转避免手机照片方向不对 img ImageOps.exif_transpose(img) img.save(out, formatfmt, qualityquality) return out with ThreadPoolExecutor(max_workers4) as pool: results list(pool.map(process_one, files)) return {total: len(results), output_dir: str(dst_dir)}两个关键点。第一一定要用ImageOps.exif_transpose(img)处理手机照片的 EXIF 方向信息不然你会得到一批歪着脑袋的图片。第二压缩 PNG 到 JPEG 之前要记得处理透明通道直接把带 alpha 的 PNG 保存成 JPEG 会得到黑色背景正确做法是先合成到白色背景上。这两个坑我都是踩过之后才写进注释里的。4.3 用自然语言直接操作终端AI Agent 模式这是 CLI-Anything 最亮眼的玩法输入一句人话它帮你拆解成命令并执行。比如我想“统计当前目录下所有 Python 文件的总行数并输出到 report.txt”传统做法是自己去查wc -l的语法再编写管道组合Agent 模式下只需要cli-anything agent 统计当前目录下所有 .py 文件的总行数并输出到 report.txt内部流程是先把这句话发给 LLM让它返回一段可执行的 Shell 命令框架在真正执行之前会把命令打印到终端并等待用户按y确认。这个确认步骤绝对不能省因为大模型生成的命令偶尔会出现幻觉尤其是涉及删除、覆盖文件的操作风险极大。我还加了一个黑名单机制包含rm -rf、mkfs、dd这类高风险命令的候选指令会被直接拒掉并提示用户改成显式指定路径的写法。这是把 AI 的能力和命令执行的安全边界结合到一起的典型案例建议所有做命令行 AI 工具的人都参考一下。5. 常见问题与排查技巧实录5.1 命令找不到原因可能不在注册表有次我新增了一个插件执行cli-anything newcmd却提示 command not found。排查了一圈最后发现问题不在注册表而在文件权限插件目录是从别的机器 clone 过来的Python 文件没有可执行权限扫描器在导入模块时被拒了。讽刺的是报错信息里完全没有提权限问题只显示“无法加载模块 demo.py”。在 Linux 环境里遇到命令找不到先按这个顺序排查插件文件是否有读权限、目录名是否以_开头被扫描器忽略了、register()函数是否真的被调用、命令名是否不小心带了下划线或空格。建议在启动时加一条 debug 日志打印出注册表里所有命令名几十秒就能定位问题。5.2 流式输出乱码尤其在 Windows 上Windows 默认控制台的编码经常和 UTF-8 打架。最开始我们在 Windows 上运行 CLI-Anything 的流式输出中文直接乱码成锟斤拷。这不是内容问题而是 stdout 编码问题。解决方案是在入口处强制设置 Python 标准输出编码为 UTF-8import sys if hasattr(sys.stdout, reconfigure): sys.stdout.reconfigure(encodingutf-8)同时尽量别在 Windows 上用 CMD 跑复杂输出改用 Windows Terminal 或者 VS Code 的集成终端表现会稳定很多。如果你还在维护老代码记得把PYTHONIOENCODINGutf-8写入环境变量作为兜底。5.3 配置热更新失败改了不生效CLI-Anything 支持配置热更新但有人反馈说我改了 YAML 就是不生效。查下来发现是因为配置加载器做了缓存修改文件后缓存没有失效。代码里加了一个基于文件 mtime 的判断每次读取配置时先比较文件的最后修改时间变了才重新加载。def load_config_if_modified(path: str, last_mtime: float) - tuple[dict, float] | None: current_mtime Path(path).stat().st_mtime if current_mtime last_mtime: return None # 未修改返回旧值和旧时间 config _do_load(path) return config, current_mtime这个方案能省掉重启进程的麻烦但在使用的时候也要小心热更新会对当前正在运行的命令生效可能造成执行结果不一致。所以我只对enabled这种“开关类”配置做热更新业务参数变化还是要求重启。5.4 跨平台坑路径不要自己拼同一份插件代码在 macOS 上运行正常一放到 Linux 服务器上就开始报错。我排查后发现是路径拼接用了/硬编码而在一台 Windows 机器上则直接崩了。经过这些教训后我在项目规范里规定所有文件路径一律用pathlib.Path处理禁止手写/或\拼接。场景错误做法正确做法拼接路径src_dir / namePath(src_dir) / name判断后缀name.endswith(.py)Path(name).suffix .py获取家目录读取HOME环境变量Path.home()跨平台这个问题看似低级实际上是最容易反复踩的坑。早日使用pathlib你的插件寿命会长很多。6. 再往前一步CLI-Anything 的扩展玩法6.1 做成插件市场一键安装别人写好的命令现在插件都是本地文件共享起来要靠 Git 仓库。后续可以做一层“插件市场”概念定义一种cli-plugin.yaml的元数据格式包含插件名、版本、作者、依赖项然后通过cli-anything install plugin-name从远程仓库拉取并自动安装。这里有两个安全设计必须提前想好插件包要做签名校验至少校验 SHA256 哈希防止下载内容被篡改插件安装后默认跑在沙箱模式里禁止访问网络和敏感目录除非用户在安装时明确授予相应权限。命令行工具做好之后会非常容易做得“飞起来”安全边界就格外重要。6.2 从单机工具到团队自动化中心CLI-Anything 完全可以充当一个团队的自动化操作中心。除了个人使用你可以把调度器放到一台常驻服务器上让团队的定时任务都通过它来统一管理。再加上权限设计普通成员只能执行只读类命令核心成员可以执行写操作类命令所有执行记录都写入审计日志。这听起来像一个大工程但基于插件化的设计其实只是时间问题。我个人实际使用中最喜欢的一个小习惯是把临时起意的重复操作直接变成一次性插件不想保留的话就删除文件。这个项目真正的价值不在于某一个命令有多强而在于它把“本来要折腾十分钟的事情”变成了“一条命令加一个回车”。如果这篇文章让你对 CLI-Anything 的搭建有了具体想法我的建议是别等完善了框架再开始用先把你手头最烦的三个重复性工作做成插件跑通一个最小的闭环。你会发现第二十条命令永远比第一条更好写而“用命令行搞定一切”的惯性会替你省下远超想象的时间。
返回列表