最近我把手上一堆零散的自动化脚本统一收拢进了一个叫 colibri 的小项目里。名字取自法语里的“蜂鸟”当时起名的时候没想太多就是觉得这鸟体型小、翅膀扇得快、还能在空中悬停挺符合我对一个工具的全部期待轻量、敏捷、按需出现用完了就飞走不占地方。这个项目本质上是一个本地任务执行框架用 Python 写的核心功能是把“定时清理日志”“批量压缩图片”“监控磁盘水位”“处理临时文件”这类日常琐事统一管理起来。你可以把它理解成一个轻量版的任务调度器但它没有常驻后台服务没有 Web 界面不上 Kubernetes就是一个命令行工具加一套插件机制。触发方式可以是手动执行也可以通过系统自带的定时任务来调。写这篇文章是想把整个设计思路、核心代码、实操步骤和踩过的坑一次性说清楚。如果你想搭一套自己的本地自动化工具链或者正在犹豫要不要为了几个小脚本引入一套重型调度框架那这篇应该能给你省点时间。内容不挑基础哪怕你刚接触 Python按着步骤走也能跑起来。1. 为什么叫 colibri以及它到底解决什么问题1.1 蜂鸟给我的三条设计原则蜂鸟这种生物很有意思它每秒翅膀能扇几十次可以做出其他鸟类做不到的悬停动作但体重通常只有几克。它不需要像鹰那样飞几个小时只需要在花丛之间快速穿梭采完蜜就走。我当时想日常开发里的很多自动化需求其实也这样任务本身很小但执行频次高、种类杂不值得为了它们搭建一套重量级系统。于是 colibri 在设计之初就定了三条硬性原则。第一单次运行时间要短。所有任务必须是“跑完即退”的进程模型任务结束进程就退出不驻留、不监听端口、不写常驻内存状态。这样出问题的时候最多就是这一次执行失败不会拖垮整台机器。第二依赖要少。能用一个标准库解决的问题绝不引入第三方框架。整个项目运行起来只需要 Python 3.10 和少量必要依赖装完之后一个虚拟环境也就几十兆拷贝到别的机器上也能直接用。第三每一个任务都是一等公民。任务之间尽量不互相依赖这样你可以随时往 tasks 目录里丢一个新的 .py 文件来增加能力不需要改动框架代码。插件化是我最坚持的一点因为日常需求变化太快今天要清理日志明天要统计接口数据如果每加一个任务都要改主程序那维护成本很快就会失控。1.2 常见工具对比为什么不自建轮子在动手写 colibri 之前我把市面上能想到的方案都过了一遍。最后选择的路线是“系统定时任务 轻量框架”而不是去部署一套现成的调度系统。下面这张表是我当时对比的核心结论方案依赖与复杂度运行方式适合场景我不选它的原因裸脚本 crontab极低零依赖每次独立进程一两个固定脚本脚本一多没法管理参数、日志、状态全靠自觉cron 自研小框架colibri 路线低仅运行时依赖每次独立进程十几个到几十个本地任务——Airflow / DolphinScheduler高需要数据库和 Web 服务常驻服务跨系统、跨团队的数据管道运维成本高本地小任务用不上systemd timer中需系统权限独立服务Linux 上的服务类任务不适合跨平台写起来也偏底层云厂商定时触发器中云函数有云依赖的在线任务本地文件操作根本没法做这张表看得比较清楚colibri 的定位就是“比裸脚本多一点秩序比重型调度轻一个量级”。如果你已经有 Airflow 在跑那你不需要 colibri如果你只是有几个脚本散落在各处想统一管起来那正好是它发挥价值的地方。1.3 适用与不适用的场景清单任何工具都有边界colibri 也不例外。我在使用过程中总结了一个很简单的判断标准任务是不是“无状态、单机、可重试”的。如果答案是肯定的那就非常适合用 colibri如果不是建议老老实实上一套正经的调度系统。适合的场景包括定时清理日志和临时文件、对图片或视频做批处理、拉取接口数据后落盘、生成每日统计报表、检查磁盘和内存水位、定期备份某个目录、把一种文件格式批量转换成另一种。这些任务有个共性执行时间短、不需要跨机器协调、失败了下次重跑就行。不适合的场景也很明显需要分布式执行的批处理、任务之间有复杂依赖关系比如 A 完成后 B 才能开始、需要实时监控和告警平台联动、需要 Web UI 给非技术人员操作。这些需求一旦出现那就是 Airflow、Temporal 这类系统的领域硬往轻量框架里塞是塞不进去的。2. 核心模块拆解与实现要点2.1 项目目录结构先看整体结构。这个项目我从一开始就打算让它保持“打开即懂”的状态所以目录划分非常直白colibri/ ├── pyproject.toml ├── config.toml ├── colibri/ │ ├── __init__.py │ ├── cli.py │ ├── config.py │ ├── runner.py │ ├── registry.py │ └── utils.py ├── tasks/ │ ├── __init__.py │ ├── clean_logs.py │ ├── compress_images.py │ └── disk_usage_report.py └── data/ ├── colibri.db └── logs/核心框架代码就五个模块每个模块职责单一。cli.py负责解析命令行参数是用户入口config.py负责读取 TOML 配置文件registry.py维护任务注册表runner.py负责真正执行任务utils.py放一些通用的公共函数。tasks/目录放所有实际任务每个文件就是一个独立任务。data/目录存放运行日志和 SQLite 历史数据库。2.2 配置体系为什么选 TOML 而不选 YAML配置是一个工具的灵魂一个好的配置文件应该让用户在不读源码的情况下就能完成 80% 的定制。我当时在 TOML 和 YAML 之间犹豫了一段时间。YAML 功能更强但缩进敏感写错一个空格就解析失败而且类型推断过于智能有时候反而让人摸不着头脑。TOML 语法更接近 INI但支持完整的数据类型可读性对新手特别友好。下面是项目的 config.toml我加了详细注释# 全局配置 [general] # 默认时区所有任务计算今天时用它避免定时执行时差8小时 timezone Asia/Shanghai # 历史记录保留天数超过会被自动清理 history_retention_days 90 [storage] # SQLite 数据库文件路径 database data/colibri.db # 日志目录 log_dir data/logs # 各任务的自定义配置段任务自己按需读取 [task.clean_logs] # 清理 /tmp/applogs 下的日志保留 7 天 target_dir /tmp/applogs pattern *.log keep_days 7 # 空目录是否一并删除 remove_empty_dirs true [task.compress_images] # 批量压缩 images 目录下的 jpg压缩后质量 85 target_dir data/images recursive true quality 85配置文件里最重要的一个设计就是[task.xxx]段。每个任务的名字作为段的二级键任务的代码只负责读取自己对应的配置段互不干扰。这样即使你加了新任务也只需要在配置文件里加一段框架层面完全不需要动。2.3 任务执行管线与插件注册机制这一节是核心。整个框架的任务管理思路可以概括为八个字定义即注册注册即发现。每个任务文件通过装饰器把自己注册进全局注册表框架启动时扫描 tasks 目录动态导入所有任务模块注册表里就自然有了完整的任务清单。先看任务基类和注册器。我用了一个非常轻量的装饰器方案# colibri/registry.py 任务注册表维护任务名称到执行函数的映射。 from __future__ import annotations import importlib import pkgutil from dataclasses import dataclass, field from typing import Any, Callable from . import tasks as tasks_pkg # 全局注册表 _REGISTRY: dict[str, dict[str, Any]] {} dataclass class TaskMeta: 任务元信息描述一个任务的属性。 name: str description: str # 是否允许并发执行同一任务默认不允许 allow_concurrent: bool False def register_task( name: str, description: str , allow_concurrent: bool False, ) - Callable: 类装饰器把任务类注册到全局注册表。 用法示例 register_task(clean_logs, 清理过期日志) class CleanLogsTask: def run(self, ctx): ... def decorator(cls): if name in _REGISTRY: raise ValueError(ftask {name} already registered) _REGISTRY[name] { class: cls, meta: TaskMeta( namename, descriptiondescription, allow_concurrentallow_concurrent, ), } return cls return decorator def scan_tasks() - None: 扫描 tasks 包下的所有模块触发模块级装饰器执行。 for mod_info in pkgutil.iter_modules(tasks_pkg.__path__): if mod_info.name.startswith(_): continue importlib.import_module(fcolibri.tasks.{mod_info.name}) def get_task_names() - list[str]: return sorted(_REGISTRY.keys()) def get_task(name: str) - dict[str, Any]: if name not in _REGISTRY: raise KeyError(funknown task: {name}) return _REGISTRY[name] def all_tasks() - dict[str, dict[str, Any]]: return dict(_REGISTRY)这里有一个非常关键的细节scan_tasks()用的是pkgutil.iter_modules配合importlib.import_module。这样做的目的是让所有任务模块在启动时被导入导入的过程中模块底部的装饰器就会被执行任务自动进入注册表。以后增加新任务只需要在colibri/tasks/目录下新建一个文件什么都不用改重启命令后新任务就自动出现了。再来看命令入口cli.py。命令行交互我用的是argparse没有引入 click 或 typer因为在这个场景下argparse完全够用而且少一个依赖就少一分维护成本# colibri/cli.py 命令行入口colibri run 任务名 --config config.toml import argparse import sys from . import registry from .config import load_config from .runner import TaskContext, run_task def main(argv: list[str] | None None) - int: parser argparse.ArgumentParser( progcolibri, description轻量级本地任务执行框架, ) subparsers parser.add_subparsers(destcommand, requiredTrue) # colibri list: 列出所有已注册任务 list_parser subparsers.add_parser(list, help列出所有任务) list_parser.add_argument(--verbose, actionstore_true, help显示描述信息) # colibri run task_name run_parser subparsers.add_parser(run, help执行指定任务) run_parser.add_argument(task, help任务名称可用 colibri list 查看) run_parser.add_argument(--config, defaultconfig.toml, help配置文件路径) run_parser.add_argument(--dry-run, actionstore_true, help只打印将要执行的操作不实际改变文件) args parser.parse_args(argv) # 扫描任务模块触发注册 registry.scan_tasks() if args.command list: for name in registry.get_task_names(): # short 模式下只输出名字 if not args.verbose: print(name) else: meta registry.get_task(name)[meta] print(f{name:20s} {meta.description}) return 0 if args.command run: if args.task not in registry.all_tasks(): print(f错误: 未找到任务 {args.task}可用任务列表如下, filesys.stderr) for name in registry.get_task_names(): print(f {name}, filesys.stderr) return 1 cfg load_config(args.config) ctx TaskContext(configcfg, dry_runargs.dry_run) return run_task(args.task, ctx) return 1 if __name__ __main__: sys.exit(main())run_task的执行逻辑在runner.py里。这里我特别注意了异常隔离一个任务崩溃不能影响框架本身的稳定性同时要把关键状态写进数据库# colibri/runner.py 任务执行器负责实例化任务、注入上下文、管理执行生命周期。 import sqlite3 import time import traceback from dataclasses import dataclass, field from datetime import datetime, timezone from pathlib import Path from . import registry dataclass class TaskContext: 任务上下文包含配置和运行状态。 config: dict dry_run: bool False # 供任务写运行过程中产生的临时状态 state: dict field(default_factorydict) def _init_db(db_path: Path) - None: 初始化 SQLite 数据库表结构。 db_path.parent.mkdir(parentsTrue, exist_okTrue) with sqlite3.connect(db_path) as conn: conn.execute( CREATE TABLE IF NOT EXISTS task_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, task_name TEXT NOT NULL, status TEXT NOT NULL, started_at TEXT NOT NULL, finished_at TEXT NOT NULL, duration_ms INTEGER NOT NULL, error_message TEXT ) ) # 开启 WAL减少并发读写时的锁等待 conn.execute(PRAGMA journal_modeWAL) def run_task(task_name: str, ctx: TaskContext) - int: 执行单个任务捕获异常并记录历史。 entry registry.get_task(task_name) task_cls entry[class] task_instance task_cls() db_path Path(ctx.config[storage][database]) _init_db(db_path) started time.monotonic() started_at datetime.now(timezone.utc).isoformat() status success error_msg None print(f[{task_name}] 开始执行...) try: # 任务实例的 run 方法是真正的业务入口 result task_instance.run(ctx) print(f[{task_name}] 执行成功: {result if result else }) except Exception: status failed error_msg traceback.format_exc() print(f[{task_name}] 执行失败详情见日志, filesys.stderr) print(error_msg, filesys.stderr) finished_at datetime.now(timezone.utc).isoformat() duration_ms int((time.monotonic() - started) * 1000) with sqlite3.connect(db_path) as conn: conn.execute( INSERT INTO task_history (task_name, status, started_at, finished_at, duration_ms, error_message) VALUES (?, ?, ?, ?, ?, ?), (task_name, status, started_at, finished_at, duration_ms, error_msg), ) return 0 if status success else 1有几个细节值得展开说一下。第一SQLite 数据库不做跨进程锁依赖因为设计上就要求“同一时间只有一个执行实例”所以单个连接写入非常安全。第二我把异常信息完整记录了 traceback而不是只存一句“failed”。这样任务失败后排查问题时可以直接从数据库里捞到完整堆栈不用再去翻日志文件省了非常多时间。2.4 历史记录与日志出了事能找到原因历史记录是我为自己的“忘记症”设计的。以前用裸脚本的时候任务跑没跑、跑成没跑成全靠记忆。现在有了 SQLite 历史表每一条执行记录都能追溯。任务历史表结构就是上面代码里的四个关键字段status标记成功或失败duration_ms记录耗时error_message保存异常信息。查询最近 10 次某个任务执行情况的 SQL 很简单SELECT task_name, status, started_at, duration_ms, substr(error_message, 1, 100) FROM task_history WHERE task_name clean_logs ORDER BY id DESC LIMIT 10;除了数据库我还保留了文件日志这样一旦数据库本身出问题比如磁盘写满、文件锁死还有最后一道防线。日志写入用的是标准库logging的RotatingFileHandler按单个文件 1MB 大小轮转最多保留 5 份。这样单机小任务完全不需要引入 ELK 那一套重型设施。3. 实操过程从零搭一个自己的 colibri3.1 初始化项目开始之前先把环境准备好。我默认你本地已经装了 Python 3.10 或更高版本。先创建一个虚拟环境把目录结构拉起来mkdir colibri cd colibri python3 -m venv .venv source .venv/bin/activate pip install --upgrade pip pip install Pillow # 只有图片压缩任务需要其他任务用标准库就够了然后按前面目录结构创建文件夹和文件。最快捷的方式是先跑一遍 CLI 的 list 命令验证框架能正常启动python -m colibri list如果输出是空的说明框架已经正常加载只是还没有任何任务。下一步我们就开始写第一个真正的任务。3.2 写第一个任务清理过期日志日志清理是最典型、最不容易出错的入门任务。它不涉及复杂的第三方依赖纯粹是文件系统操作非常适合作为第一个练手项目。我在colibri/tasks/clean_logs.py里写下如下代码# colibri/tasks/clean_logs.py 清理过期日志文件。 import time from datetime import datetime, timedelta from pathlib import Path from colibri.registry import register_task register_task(clean_logs, 清理指定目录下超过保留天数的日志文件) class CleanLogsTask: def run(self, ctx): task_cfg ctx.config.get(task, {}).get(clean_logs, {}) target_dir Path(task_cfg.get(target_dir, tmp/logs)) pattern task_cfg.get(pattern, *.log) keep_days int(task_cfg.get(keep_days, 7)) if not target_dir.exists(): return f目录不存在: {target_dir} now time.time() cutoff now - keep_days * 86400 # 86400 24 * 3600 deleted_count 0 freed_bytes 0 for f in target_dir.glob(pattern): if f.is_file(): # 获取文件最后修改时间 mtime f.stat().st_mtime if mtime cutoff: if ctx.dry_run: print(f[dry-run] 将删除 {f}) else: print(f删除 {f}) freed_bytes f.stat().st_size f.unlink() deleted_count 1 # 可选清理空目录 if task_cfg.get(remove_empty_dirs, False): for d in sorted(target_dir.rglob(*), keylambda p: len(p.parts), reverseTrue): if d.is_dir() and not any(d.iterdir()): if ctx.dry_run: print(f[dry-run] 将删除空目录 {d}) else: print(f删除空目录 {d}) d.rmdir() return f共删除 {deleted_count} 个文件释放 {freed_bytes / 1024 / 1024:.2f} MB这里有个小细节值得新手注意文件删除前一定要先取st_size再unlink()顺序反了就会拿到 0 字节。另外rglob(*)遍历出来的目录列表要按层级从深到浅排一下否则你先删了父目录子目录就删不掉了。上面那行keylambda p: len(p.parts), reverseTrue干的就是这件事。对于keep_days这个参数我默认给 7 天这是很常见的日志保留周期你可以根据自己的磁盘空间调整。如果你日志量特别大建议先--dry-run跑一遍确认删的范围没问题再正式执行。3.3 写第二个任务批量压缩图片第二个任务我选的是图片压缩。为什么会选它因为图片处理是日常需求里出现频率很高、但用裸脚本处理起来又略麻烦的场景。抓个现成的 Pillow 库就能搞定。# colibri/tasks/compress_images.py 批量压缩图片文件。 from pathlib import Path from PIL import Image from colibri.registry import register_task register_task(compress_images, 批量压缩 JPG/PNG 图片) class CompressImagesTask: def run(self, ctx): task_cfg ctx.config.get(task, {}).get(compress_images, {}) target_dir Path(task_cfg.get(target_dir, data/images)) recursive bool(task_cfg.get(recursive, False)) quality int(task_cfg.get(quality, 85)) if not target_dir.exists(): return f目录不存在: {target_dir} iterator target_dir.rglob(*) if recursive else target_dir.glob(*) processed 0 saved_bytes 0 for img_path in iterator: if img_path.suffix.lower() not in (.jpg, .jpeg, .png): continue if not img_path.is_file(): continue original_size img_path.stat().st_size out_path img_path.with_suffix(.compressed.jpg) if ctx.dry_run: print(f[dry-run] 将压缩 {img_path} - {out_path}) continue try: with Image.open(img_path) as im: # 转换为 RGB统一 JPEG 输出格式 rgb im.convert(RGB) rgb.save(out_path, JPEG, qualityquality, optimizeTrue) new_size out_path.stat().st_size saved_bytes original_size - new_size processed 1 print(f压缩 {img_path.name}: {original_size / 1024:.1f}KB - {new_size / 1024:.1f}KB) except Exception as e: # 文件损坏或格式问题跳过但不中断整个任务 print(f跳过 {img_path}: {e}) return f处理 {processed} 张图片节省 {saved_bytes / 1024 / 1024:.2f} MB质量参数给 85 是我实测下来的经验值对大多数场景是一个压缩比和画质都比较均衡的档位。下面是同一张图在不同质量参数下的对比我用它来说明这个选择背后的逻辑压缩质量输出大小体积变化肉眼观感1002.1 MB100%原图画质几乎无损901.2 MB约 57%基本无感知差异850.9 MB约 43%放大后能发现细微差异750.6 MB约 29%正常观看可接受细节有明显损失600.4 MB约 19%文字边缘出现明显锯齿不适合存档如果是电商图、博客配图这类场景85 是安全选择如果是精修摄影图要长期存档建议直接 90 以上只有缩略图这种对画质不敏感的用途才建议压到 75 以下。这个任务运行完原始图片文件不会被覆盖压缩后的新文件会以.compressed.jpg后缀保存方便你先检查再决定原图是否删掉。3.4 接入定时调度让任务自己跑起来手动执行很有意思但 colibri 的更大价值在于无人值守地定时执行。这一步就是把我们的 CLI 命令挂到操作系统的定时机制上。在 Linux 和 macOS 上用的是 crontab。执行crontab -e打开编辑界面加入下面这几行# 每天凌晨 2 点清理日志 0 2 * * * /home/user/colibri/.venv/bin/python -m colibri run clean_logs --config /home/user/colibri/config.toml /home/user/colibri/data/logs/cron.log 21 # 每周日凌晨 3 点压缩图片 0 3 * * 0 /home/user/colibri/.venv/bin/python -m colibri run compress_images --config /home/user/colibri/config.toml /home/user/colibri/data/logs/cron.log 21有几个细节这里必须说明白都是我在真实环境里踩过的坑。第一cron 默认的 PATH 非常精简直接写python很可能定位到系统自带的旧版 Python而不是虚拟环境里的 Python。所以必须用绝对路径指向.venv/bin/python。第二--config参数也要用绝对路径因为 cron 执行时的工作目录不是你当时敲crontab -e的目录。第三标准输出和错误输出都要重定向到日志文件否则任务一旦有输出cron 会尝试发邮件通知你如果系统没配邮件服务输出会直接丢失排错的时候什么线索都看不到。在 Windows 上则用“任务计划程序”来做同样的事情。基础设置只需要三步创建基本任务、触发器选每天或每周、操作选“启动程序”程序填虚拟环境里的python.exe绝对路径参数填-m colibri run clean_logs --config D:\path\to\config.toml起始于填项目根目录。3.5 查看运行结果从命令行和数据库两个维度观察任务跑完之后查看结果有两个入口。第一个入口是命令行直接跑一次适合调试python -m colibri list --verbose python -m colibri run clean_logs --dry-run python -m colibri run clean_logs第二个入口是查数据库。我已经把每次执行的历史都记录在data/colibri.db里用sqlite3命令可以直接查sqlite3 data/colibri.db进入交互模式后SELECT task_name, status, started_at, duration_ms FROM task_history ORDER BY id DESC LIMIT 10;这个查询会告诉你最近 10 次执行情况。如果发现某次任务执行失败再把error_message字段拉出来看SELECT error_message FROM task_history WHERE status failed ORDER BY id DESC LIMIT 1;不管是通过日志文件还是数据库出问题的时候都能快速定位是环境问题、配置问题还是代码问题。4. 常见问题与排查技巧实录4.1 定时任务里找不到命令和环境变量我遇到过的第一个“诡异问题”就是任务在手动执行时一切正常但一到 cron 里就跑不起来还报“module not found”或者“command not found”。原因基本都出在环境上cron 和 GUI 登录会话的 PATH 完全不同它只保留极少数系统目录路径。排查思路很简单。第一步把任务里的 Python 解释器路径换成绝对路径这是最常见的修复手段。第二步如果任务内部还调用了系统命令比如ffmpeg、git你还需要在任务代码里显式指定这些命令的绝对路径或者在任务执行前先export PATH/usr/local/bin:/usr/bin:/bin。第三步在代码里打印sys.executable和os.environ.get(PATH)确认实际执行环境和你预期的一致。这个坑的根本原因在于“定时环境”和“交互环境”的隔离从设计上理解它就不会觉得很玄。4.2 中文路径和编码问题Windows 下跑任务时会遇到一类很常见的问题路径里带中文或者配置文件里有中文注释命令行直接报 UnicodeDecodeError 或者 GBK 编解码错误。原因是 Windows 控制台默认用 GBK代码页 936而 Python 3 在读取文件和输出字符串时默认用 UTF-8。我的解决办法是第一所有源文件头部加# -*- coding: utf-8 -*-或者干脆把编辑器默认编码设成 UTF-8。第二如果脚本要输出中文先测试控制台能否正常显示不行就临时设置环境变量PYTHONIOENCODINGutf-8。第三路径处理统一用pathlib.Path不要用字符串拼接这样能避免很多 Windows 反斜杠带来的转义烦恼。4.3 SQLite 数据库文件被锁SQLite 是轻量级的但它有写入锁机制同一时刻只允许一个写事务。如果你的某个任务耗时很长而你又同时用sqlite3命令去查同一份历史表就可能看到database is locked报错。解决方式我分两层处理。第一层在连接 SQLite 时使用默认的连接超时默认 5 秒并在建表时开启 WAL 日志模式PRAGMA journal_modeWALWAL 允许读写并发能大幅降低锁冲突概率。第二层也是最根本的保证同一个任务不要并发执行。我在设计任务注册表里保留了allow_concurrent字段并在注册装饰器时显式暴露出来就是提醒自己除非确定任务内部做了并发安全处理否则不要让两个进程同时跑同一个任务。4.4 新增任务后 list 里面看不到这是所有插件化框架都会遇到的问题明明在colibri/tasks/目录下新建了文件但colibri list输出里什么都没有。原因通常有三个。第一个是文件命名问题。scan_tasks会跳过所有以_开头的模块名如果你的文件名是_clean_logs.py它就不会被导入。第二个是导入时的异常被吞掉了比如任务文件里 import 了一个没安装的依赖但异常发生在了模块顶层而importlib.import_module在模块导入失败时会抛出ModuleNotFoundError这时需要去检查任务文件顶层的 import 语句以及pyproject.toml里的依赖声明。第三个是装饰器没写对register_task的第一个参数必须和你在配置里的[task.xxx]段名保持一一对应否则即使注册成功执行时也会因为取不到配置而报错。4.5 任务执行失败排查速查表现象可能原因检查手段手动执行成功定时执行失败环境变量 PATH 不全、工作目录不对用绝对路径打印sys.executable和当前工作目录中文内容乱码编码不一致控制台 GBK 与文件 UTF-8 冲突设置PYTHONIOENCODINGutf-8统一用 pathlib任务显示成功但文件没变化配置项路径写错或匹配不到文件加print打印实际扫描范围用--dry-run验证数据库锁死多个进程同时写同一个 SQLite开启 WAL限制任务并发延长连接超时新增任务在 list 中不显示文件名下划线开头、依赖缺失、装饰器未执行直接 import 任务文件看报错用python -c import colibri.tasks.xxx验证图片压缩后文件比原文件大JPEG 转 JPEG 正常但大尺寸 PNG 可能不缩反涨对 PNG 先转 RGB 再用 JPEG 保存同时合理设置 quality检查原图是否本身已是高压缩任务重复执行cron 配置了多个入口或手动和 cron 同时跑查看 cron 列表在 history 表里查执行时间间隔日志文件增长过大没有合理轮转长时间累积配置 RotatingFileHandler定期清理 data/logs这张表是我在真实使用中整理出来的高频问题清单。大部分问题都不是代码逻辑写错而是环境、路径、并发这几类边缘情况。把这些想明白了工具用起来就会稳很多。最后再分享一点个人体会。用 colibri 一段时间后我有一个很深的感触很多看似需要复杂系统解决的问题在一台普通的电脑上用一个不到一千行的小框架就能处理得游刃有余。当你真的把任务从“临时跑一下的脚本”升级成“有配置、有插件、有历史记录的框架任务”之后整个心智负担会小很多。不要再为三五个小脚本去搭一套又重又复杂的系统了工具不在于大而在于贴身。蜂鸟虽小该采的花蜜一点都不会少。
最近我把手上一堆零散的自动化脚本统一收拢进了一个叫 colibri 的小项目里。名字取自法语里的“蜂鸟”当时起名的时候没想太多就是觉得这鸟体型小、翅膀扇得快、还能在空中悬停挺符合我对一个工具的全部期待轻量、敏捷、按需出现用完了就飞走不占地方。这个项目本质上是一个本地任务执行框架用 Python 写的核心功能是把“定时清理日志”“批量压缩图片”“监控磁盘水位”“处理临时文件”这类日常琐事统一管理起来。你可以把它理解成一个轻量版的任务调度器但它没有常驻后台服务没有 Web 界面不上 Kubernetes就是一个命令行工具加一套插件机制。触发方式可以是手动执行也可以通过系统自带的定时任务来调。写这篇文章是想把整个设计思路、核心代码、实操步骤和踩过的坑一次性说清楚。如果你想搭一套自己的本地自动化工具链或者正在犹豫要不要为了几个小脚本引入一套重型调度框架那这篇应该能给你省点时间。内容不挑基础哪怕你刚接触 Python按着步骤走也能跑起来。1. 为什么叫 colibri以及它到底解决什么问题1.1 蜂鸟给我的三条设计原则蜂鸟这种生物很有意思它每秒翅膀能扇几十次可以做出其他鸟类做不到的悬停动作但体重通常只有几克。它不需要像鹰那样飞几个小时只需要在花丛之间快速穿梭采完蜜就走。我当时想日常开发里的很多自动化需求其实也这样任务本身很小但执行频次高、种类杂不值得为了它们搭建一套重量级系统。于是 colibri 在设计之初就定了三条硬性原则。第一单次运行时间要短。所有任务必须是“跑完即退”的进程模型任务结束进程就退出不驻留、不监听端口、不写常驻内存状态。这样出问题的时候最多就是这一次执行失败不会拖垮整台机器。第二依赖要少。能用一个标准库解决的问题绝不引入第三方框架。整个项目运行起来只需要 Python 3.10 和少量必要依赖装完之后一个虚拟环境也就几十兆拷贝到别的机器上也能直接用。第三每一个任务都是一等公民。任务之间尽量不互相依赖这样你可以随时往 tasks 目录里丢一个新的 .py 文件来增加能力不需要改动框架代码。插件化是我最坚持的一点因为日常需求变化太快今天要清理日志明天要统计接口数据如果每加一个任务都要改主程序那维护成本很快就会失控。1.2 常见工具对比为什么不自建轮子在动手写 colibri 之前我把市面上能想到的方案都过了一遍。最后选择的路线是“系统定时任务 轻量框架”而不是去部署一套现成的调度系统。下面这张表是我当时对比的核心结论方案依赖与复杂度运行方式适合场景我不选它的原因裸脚本 crontab极低零依赖每次独立进程一两个固定脚本脚本一多没法管理参数、日志、状态全靠自觉cron 自研小框架colibri 路线低仅运行时依赖每次独立进程十几个到几十个本地任务——Airflow / DolphinScheduler高需要数据库和 Web 服务常驻服务跨系统、跨团队的数据管道运维成本高本地小任务用不上systemd timer中需系统权限独立服务Linux 上的服务类任务不适合跨平台写起来也偏底层云厂商定时触发器中云函数有云依赖的在线任务本地文件操作根本没法做这张表看得比较清楚colibri 的定位就是“比裸脚本多一点秩序比重型调度轻一个量级”。如果你已经有 Airflow 在跑那你不需要 colibri如果你只是有几个脚本散落在各处想统一管起来那正好是它发挥价值的地方。1.3 适用与不适用的场景清单任何工具都有边界colibri 也不例外。我在使用过程中总结了一个很简单的判断标准任务是不是“无状态、单机、可重试”的。如果答案是肯定的那就非常适合用 colibri如果不是建议老老实实上一套正经的调度系统。适合的场景包括定时清理日志和临时文件、对图片或视频做批处理、拉取接口数据后落盘、生成每日统计报表、检查磁盘和内存水位、定期备份某个目录、把一种文件格式批量转换成另一种。这些任务有个共性执行时间短、不需要跨机器协调、失败了下次重跑就行。不适合的场景也很明显需要分布式执行的批处理、任务之间有复杂依赖关系比如 A 完成后 B 才能开始、需要实时监控和告警平台联动、需要 Web UI 给非技术人员操作。这些需求一旦出现那就是 Airflow、Temporal 这类系统的领域硬往轻量框架里塞是塞不进去的。2. 核心模块拆解与实现要点2.1 项目目录结构先看整体结构。这个项目我从一开始就打算让它保持“打开即懂”的状态所以目录划分非常直白colibri/ ├── pyproject.toml ├── config.toml ├── colibri/ │ ├── __init__.py │ ├── cli.py │ ├── config.py │ ├── runner.py │ ├── registry.py │ └── utils.py ├── tasks/ │ ├── __init__.py │ ├── clean_logs.py │ ├── compress_images.py │ └── disk_usage_report.py └── data/ ├── colibri.db └── logs/核心框架代码就五个模块每个模块职责单一。cli.py负责解析命令行参数是用户入口config.py负责读取 TOML 配置文件registry.py维护任务注册表runner.py负责真正执行任务utils.py放一些通用的公共函数。tasks/目录放所有实际任务每个文件就是一个独立任务。data/目录存放运行日志和 SQLite 历史数据库。2.2 配置体系为什么选 TOML 而不选 YAML配置是一个工具的灵魂一个好的配置文件应该让用户在不读源码的情况下就能完成 80% 的定制。我当时在 TOML 和 YAML 之间犹豫了一段时间。YAML 功能更强但缩进敏感写错一个空格就解析失败而且类型推断过于智能有时候反而让人摸不着头脑。TOML 语法更接近 INI但支持完整的数据类型可读性对新手特别友好。下面是项目的 config.toml我加了详细注释# 全局配置 [general] # 默认时区所有任务计算今天时用它避免定时执行时差8小时 timezone Asia/Shanghai # 历史记录保留天数超过会被自动清理 history_retention_days 90 [storage] # SQLite 数据库文件路径 database data/colibri.db # 日志目录 log_dir data/logs # 各任务的自定义配置段任务自己按需读取 [task.clean_logs] # 清理 /tmp/applogs 下的日志保留 7 天 target_dir /tmp/applogs pattern *.log keep_days 7 # 空目录是否一并删除 remove_empty_dirs true [task.compress_images] # 批量压缩 images 目录下的 jpg压缩后质量 85 target_dir data/images recursive true quality 85配置文件里最重要的一个设计就是[task.xxx]段。每个任务的名字作为段的二级键任务的代码只负责读取自己对应的配置段互不干扰。这样即使你加了新任务也只需要在配置文件里加一段框架层面完全不需要动。2.3 任务执行管线与插件注册机制这一节是核心。整个框架的任务管理思路可以概括为八个字定义即注册注册即发现。每个任务文件通过装饰器把自己注册进全局注册表框架启动时扫描 tasks 目录动态导入所有任务模块注册表里就自然有了完整的任务清单。先看任务基类和注册器。我用了一个非常轻量的装饰器方案# colibri/registry.py 任务注册表维护任务名称到执行函数的映射。 from __future__ import annotations import importlib import pkgutil from dataclasses import dataclass, field from typing import Any, Callable from . import tasks as tasks_pkg # 全局注册表 _REGISTRY: dict[str, dict[str, Any]] {} dataclass class TaskMeta: 任务元信息描述一个任务的属性。 name: str description: str # 是否允许并发执行同一任务默认不允许 allow_concurrent: bool False def register_task( name: str, description: str , allow_concurrent: bool False, ) - Callable: 类装饰器把任务类注册到全局注册表。 用法示例 register_task(clean_logs, 清理过期日志) class CleanLogsTask: def run(self, ctx): ... def decorator(cls): if name in _REGISTRY: raise ValueError(ftask {name} already registered) _REGISTRY[name] { class: cls, meta: TaskMeta( namename, descriptiondescription, allow_concurrentallow_concurrent, ), } return cls return decorator def scan_tasks() - None: 扫描 tasks 包下的所有模块触发模块级装饰器执行。 for mod_info in pkgutil.iter_modules(tasks_pkg.__path__): if mod_info.name.startswith(_): continue importlib.import_module(fcolibri.tasks.{mod_info.name}) def get_task_names() - list[str]: return sorted(_REGISTRY.keys()) def get_task(name: str) - dict[str, Any]: if name not in _REGISTRY: raise KeyError(funknown task: {name}) return _REGISTRY[name] def all_tasks() - dict[str, dict[str, Any]]: return dict(_REGISTRY)这里有一个非常关键的细节scan_tasks()用的是pkgutil.iter_modules配合importlib.import_module。这样做的目的是让所有任务模块在启动时被导入导入的过程中模块底部的装饰器就会被执行任务自动进入注册表。以后增加新任务只需要在colibri/tasks/目录下新建一个文件什么都不用改重启命令后新任务就自动出现了。再来看命令入口cli.py。命令行交互我用的是argparse没有引入 click 或 typer因为在这个场景下argparse完全够用而且少一个依赖就少一分维护成本# colibri/cli.py 命令行入口colibri run 任务名 --config config.toml import argparse import sys from . import registry from .config import load_config from .runner import TaskContext, run_task def main(argv: list[str] | None None) - int: parser argparse.ArgumentParser( progcolibri, description轻量级本地任务执行框架, ) subparsers parser.add_subparsers(destcommand, requiredTrue) # colibri list: 列出所有已注册任务 list_parser subparsers.add_parser(list, help列出所有任务) list_parser.add_argument(--verbose, actionstore_true, help显示描述信息) # colibri run task_name run_parser subparsers.add_parser(run, help执行指定任务) run_parser.add_argument(task, help任务名称可用 colibri list 查看) run_parser.add_argument(--config, defaultconfig.toml, help配置文件路径) run_parser.add_argument(--dry-run, actionstore_true, help只打印将要执行的操作不实际改变文件) args parser.parse_args(argv) # 扫描任务模块触发注册 registry.scan_tasks() if args.command list: for name in registry.get_task_names(): # short 模式下只输出名字 if not args.verbose: print(name) else: meta registry.get_task(name)[meta] print(f{name:20s} {meta.description}) return 0 if args.command run: if args.task not in registry.all_tasks(): print(f错误: 未找到任务 {args.task}可用任务列表如下, filesys.stderr) for name in registry.get_task_names(): print(f {name}, filesys.stderr) return 1 cfg load_config(args.config) ctx TaskContext(configcfg, dry_runargs.dry_run) return run_task(args.task, ctx) return 1 if __name__ __main__: sys.exit(main())run_task的执行逻辑在runner.py里。这里我特别注意了异常隔离一个任务崩溃不能影响框架本身的稳定性同时要把关键状态写进数据库# colibri/runner.py 任务执行器负责实例化任务、注入上下文、管理执行生命周期。 import sqlite3 import time import traceback from dataclasses import dataclass, field from datetime import datetime, timezone from pathlib import Path from . import registry dataclass class TaskContext: 任务上下文包含配置和运行状态。 config: dict dry_run: bool False # 供任务写运行过程中产生的临时状态 state: dict field(default_factorydict) def _init_db(db_path: Path) - None: 初始化 SQLite 数据库表结构。 db_path.parent.mkdir(parentsTrue, exist_okTrue) with sqlite3.connect(db_path) as conn: conn.execute( CREATE TABLE IF NOT EXISTS task_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, task_name TEXT NOT NULL, status TEXT NOT NULL, started_at TEXT NOT NULL, finished_at TEXT NOT NULL, duration_ms INTEGER NOT NULL, error_message TEXT ) ) # 开启 WAL减少并发读写时的锁等待 conn.execute(PRAGMA journal_modeWAL) def run_task(task_name: str, ctx: TaskContext) - int: 执行单个任务捕获异常并记录历史。 entry registry.get_task(task_name) task_cls entry[class] task_instance task_cls() db_path Path(ctx.config[storage][database]) _init_db(db_path) started time.monotonic() started_at datetime.now(timezone.utc).isoformat() status success error_msg None print(f[{task_name}] 开始执行...) try: # 任务实例的 run 方法是真正的业务入口 result task_instance.run(ctx) print(f[{task_name}] 执行成功: {result if result else }) except Exception: status failed error_msg traceback.format_exc() print(f[{task_name}] 执行失败详情见日志, filesys.stderr) print(error_msg, filesys.stderr) finished_at datetime.now(timezone.utc).isoformat() duration_ms int((time.monotonic() - started) * 1000) with sqlite3.connect(db_path) as conn: conn.execute( INSERT INTO task_history (task_name, status, started_at, finished_at, duration_ms, error_message) VALUES (?, ?, ?, ?, ?, ?), (task_name, status, started_at, finished_at, duration_ms, error_msg), ) return 0 if status success else 1有几个细节值得展开说一下。第一SQLite 数据库不做跨进程锁依赖因为设计上就要求“同一时间只有一个执行实例”所以单个连接写入非常安全。第二我把异常信息完整记录了 traceback而不是只存一句“failed”。这样任务失败后排查问题时可以直接从数据库里捞到完整堆栈不用再去翻日志文件省了非常多时间。2.4 历史记录与日志出了事能找到原因历史记录是我为自己的“忘记症”设计的。以前用裸脚本的时候任务跑没跑、跑成没跑成全靠记忆。现在有了 SQLite 历史表每一条执行记录都能追溯。任务历史表结构就是上面代码里的四个关键字段status标记成功或失败duration_ms记录耗时error_message保存异常信息。查询最近 10 次某个任务执行情况的 SQL 很简单SELECT task_name, status, started_at, duration_ms, substr(error_message, 1, 100) FROM task_history WHERE task_name clean_logs ORDER BY id DESC LIMIT 10;除了数据库我还保留了文件日志这样一旦数据库本身出问题比如磁盘写满、文件锁死还有最后一道防线。日志写入用的是标准库logging的RotatingFileHandler按单个文件 1MB 大小轮转最多保留 5 份。这样单机小任务完全不需要引入 ELK 那一套重型设施。3. 实操过程从零搭一个自己的 colibri3.1 初始化项目开始之前先把环境准备好。我默认你本地已经装了 Python 3.10 或更高版本。先创建一个虚拟环境把目录结构拉起来mkdir colibri cd colibri python3 -m venv .venv source .venv/bin/activate pip install --upgrade pip pip install Pillow # 只有图片压缩任务需要其他任务用标准库就够了然后按前面目录结构创建文件夹和文件。最快捷的方式是先跑一遍 CLI 的 list 命令验证框架能正常启动python -m colibri list如果输出是空的说明框架已经正常加载只是还没有任何任务。下一步我们就开始写第一个真正的任务。3.2 写第一个任务清理过期日志日志清理是最典型、最不容易出错的入门任务。它不涉及复杂的第三方依赖纯粹是文件系统操作非常适合作为第一个练手项目。我在colibri/tasks/clean_logs.py里写下如下代码# colibri/tasks/clean_logs.py 清理过期日志文件。 import time from datetime import datetime, timedelta from pathlib import Path from colibri.registry import register_task register_task(clean_logs, 清理指定目录下超过保留天数的日志文件) class CleanLogsTask: def run(self, ctx): task_cfg ctx.config.get(task, {}).get(clean_logs, {}) target_dir Path(task_cfg.get(target_dir, tmp/logs)) pattern task_cfg.get(pattern, *.log) keep_days int(task_cfg.get(keep_days, 7)) if not target_dir.exists(): return f目录不存在: {target_dir} now time.time() cutoff now - keep_days * 86400 # 86400 24 * 3600 deleted_count 0 freed_bytes 0 for f in target_dir.glob(pattern): if f.is_file(): # 获取文件最后修改时间 mtime f.stat().st_mtime if mtime cutoff: if ctx.dry_run: print(f[dry-run] 将删除 {f}) else: print(f删除 {f}) freed_bytes f.stat().st_size f.unlink() deleted_count 1 # 可选清理空目录 if task_cfg.get(remove_empty_dirs, False): for d in sorted(target_dir.rglob(*), keylambda p: len(p.parts), reverseTrue): if d.is_dir() and not any(d.iterdir()): if ctx.dry_run: print(f[dry-run] 将删除空目录 {d}) else: print(f删除空目录 {d}) d.rmdir() return f共删除 {deleted_count} 个文件释放 {freed_bytes / 1024 / 1024:.2f} MB这里有个小细节值得新手注意文件删除前一定要先取st_size再unlink()顺序反了就会拿到 0 字节。另外rglob(*)遍历出来的目录列表要按层级从深到浅排一下否则你先删了父目录子目录就删不掉了。上面那行keylambda p: len(p.parts), reverseTrue干的就是这件事。对于keep_days这个参数我默认给 7 天这是很常见的日志保留周期你可以根据自己的磁盘空间调整。如果你日志量特别大建议先--dry-run跑一遍确认删的范围没问题再正式执行。3.3 写第二个任务批量压缩图片第二个任务我选的是图片压缩。为什么会选它因为图片处理是日常需求里出现频率很高、但用裸脚本处理起来又略麻烦的场景。抓个现成的 Pillow 库就能搞定。# colibri/tasks/compress_images.py 批量压缩图片文件。 from pathlib import Path from PIL import Image from colibri.registry import register_task register_task(compress_images, 批量压缩 JPG/PNG 图片) class CompressImagesTask: def run(self, ctx): task_cfg ctx.config.get(task, {}).get(compress_images, {}) target_dir Path(task_cfg.get(target_dir, data/images)) recursive bool(task_cfg.get(recursive, False)) quality int(task_cfg.get(quality, 85)) if not target_dir.exists(): return f目录不存在: {target_dir} iterator target_dir.rglob(*) if recursive else target_dir.glob(*) processed 0 saved_bytes 0 for img_path in iterator: if img_path.suffix.lower() not in (.jpg, .jpeg, .png): continue if not img_path.is_file(): continue original_size img_path.stat().st_size out_path img_path.with_suffix(.compressed.jpg) if ctx.dry_run: print(f[dry-run] 将压缩 {img_path} - {out_path}) continue try: with Image.open(img_path) as im: # 转换为 RGB统一 JPEG 输出格式 rgb im.convert(RGB) rgb.save(out_path, JPEG, qualityquality, optimizeTrue) new_size out_path.stat().st_size saved_bytes original_size - new_size processed 1 print(f压缩 {img_path.name}: {original_size / 1024:.1f}KB - {new_size / 1024:.1f}KB) except Exception as e: # 文件损坏或格式问题跳过但不中断整个任务 print(f跳过 {img_path}: {e}) return f处理 {processed} 张图片节省 {saved_bytes / 1024 / 1024:.2f} MB质量参数给 85 是我实测下来的经验值对大多数场景是一个压缩比和画质都比较均衡的档位。下面是同一张图在不同质量参数下的对比我用它来说明这个选择背后的逻辑压缩质量输出大小体积变化肉眼观感1002.1 MB100%原图画质几乎无损901.2 MB约 57%基本无感知差异850.9 MB约 43%放大后能发现细微差异750.6 MB约 29%正常观看可接受细节有明显损失600.4 MB约 19%文字边缘出现明显锯齿不适合存档如果是电商图、博客配图这类场景85 是安全选择如果是精修摄影图要长期存档建议直接 90 以上只有缩略图这种对画质不敏感的用途才建议压到 75 以下。这个任务运行完原始图片文件不会被覆盖压缩后的新文件会以.compressed.jpg后缀保存方便你先检查再决定原图是否删掉。3.4 接入定时调度让任务自己跑起来手动执行很有意思但 colibri 的更大价值在于无人值守地定时执行。这一步就是把我们的 CLI 命令挂到操作系统的定时机制上。在 Linux 和 macOS 上用的是 crontab。执行crontab -e打开编辑界面加入下面这几行# 每天凌晨 2 点清理日志 0 2 * * * /home/user/colibri/.venv/bin/python -m colibri run clean_logs --config /home/user/colibri/config.toml /home/user/colibri/data/logs/cron.log 21 # 每周日凌晨 3 点压缩图片 0 3 * * 0 /home/user/colibri/.venv/bin/python -m colibri run compress_images --config /home/user/colibri/config.toml /home/user/colibri/data/logs/cron.log 21有几个细节这里必须说明白都是我在真实环境里踩过的坑。第一cron 默认的 PATH 非常精简直接写python很可能定位到系统自带的旧版 Python而不是虚拟环境里的 Python。所以必须用绝对路径指向.venv/bin/python。第二--config参数也要用绝对路径因为 cron 执行时的工作目录不是你当时敲crontab -e的目录。第三标准输出和错误输出都要重定向到日志文件否则任务一旦有输出cron 会尝试发邮件通知你如果系统没配邮件服务输出会直接丢失排错的时候什么线索都看不到。在 Windows 上则用“任务计划程序”来做同样的事情。基础设置只需要三步创建基本任务、触发器选每天或每周、操作选“启动程序”程序填虚拟环境里的python.exe绝对路径参数填-m colibri run clean_logs --config D:\path\to\config.toml起始于填项目根目录。3.5 查看运行结果从命令行和数据库两个维度观察任务跑完之后查看结果有两个入口。第一个入口是命令行直接跑一次适合调试python -m colibri list --verbose python -m colibri run clean_logs --dry-run python -m colibri run clean_logs第二个入口是查数据库。我已经把每次执行的历史都记录在data/colibri.db里用sqlite3命令可以直接查sqlite3 data/colibri.db进入交互模式后SELECT task_name, status, started_at, duration_ms FROM task_history ORDER BY id DESC LIMIT 10;这个查询会告诉你最近 10 次执行情况。如果发现某次任务执行失败再把error_message字段拉出来看SELECT error_message FROM task_history WHERE status failed ORDER BY id DESC LIMIT 1;不管是通过日志文件还是数据库出问题的时候都能快速定位是环境问题、配置问题还是代码问题。4. 常见问题与排查技巧实录4.1 定时任务里找不到命令和环境变量我遇到过的第一个“诡异问题”就是任务在手动执行时一切正常但一到 cron 里就跑不起来还报“module not found”或者“command not found”。原因基本都出在环境上cron 和 GUI 登录会话的 PATH 完全不同它只保留极少数系统目录路径。排查思路很简单。第一步把任务里的 Python 解释器路径换成绝对路径这是最常见的修复手段。第二步如果任务内部还调用了系统命令比如ffmpeg、git你还需要在任务代码里显式指定这些命令的绝对路径或者在任务执行前先export PATH/usr/local/bin:/usr/bin:/bin。第三步在代码里打印sys.executable和os.environ.get(PATH)确认实际执行环境和你预期的一致。这个坑的根本原因在于“定时环境”和“交互环境”的隔离从设计上理解它就不会觉得很玄。4.2 中文路径和编码问题Windows 下跑任务时会遇到一类很常见的问题路径里带中文或者配置文件里有中文注释命令行直接报 UnicodeDecodeError 或者 GBK 编解码错误。原因是 Windows 控制台默认用 GBK代码页 936而 Python 3 在读取文件和输出字符串时默认用 UTF-8。我的解决办法是第一所有源文件头部加# -*- coding: utf-8 -*-或者干脆把编辑器默认编码设成 UTF-8。第二如果脚本要输出中文先测试控制台能否正常显示不行就临时设置环境变量PYTHONIOENCODINGutf-8。第三路径处理统一用pathlib.Path不要用字符串拼接这样能避免很多 Windows 反斜杠带来的转义烦恼。4.3 SQLite 数据库文件被锁SQLite 是轻量级的但它有写入锁机制同一时刻只允许一个写事务。如果你的某个任务耗时很长而你又同时用sqlite3命令去查同一份历史表就可能看到database is locked报错。解决方式我分两层处理。第一层在连接 SQLite 时使用默认的连接超时默认 5 秒并在建表时开启 WAL 日志模式PRAGMA journal_modeWALWAL 允许读写并发能大幅降低锁冲突概率。第二层也是最根本的保证同一个任务不要并发执行。我在设计任务注册表里保留了allow_concurrent字段并在注册装饰器时显式暴露出来就是提醒自己除非确定任务内部做了并发安全处理否则不要让两个进程同时跑同一个任务。4.4 新增任务后 list 里面看不到这是所有插件化框架都会遇到的问题明明在colibri/tasks/目录下新建了文件但colibri list输出里什么都没有。原因通常有三个。第一个是文件命名问题。scan_tasks会跳过所有以_开头的模块名如果你的文件名是_clean_logs.py它就不会被导入。第二个是导入时的异常被吞掉了比如任务文件里 import 了一个没安装的依赖但异常发生在了模块顶层而importlib.import_module在模块导入失败时会抛出ModuleNotFoundError这时需要去检查任务文件顶层的 import 语句以及pyproject.toml里的依赖声明。第三个是装饰器没写对register_task的第一个参数必须和你在配置里的[task.xxx]段名保持一一对应否则即使注册成功执行时也会因为取不到配置而报错。4.5 任务执行失败排查速查表现象可能原因检查手段手动执行成功定时执行失败环境变量 PATH 不全、工作目录不对用绝对路径打印sys.executable和当前工作目录中文内容乱码编码不一致控制台 GBK 与文件 UTF-8 冲突设置PYTHONIOENCODINGutf-8统一用 pathlib任务显示成功但文件没变化配置项路径写错或匹配不到文件加print打印实际扫描范围用--dry-run验证数据库锁死多个进程同时写同一个 SQLite开启 WAL限制任务并发延长连接超时新增任务在 list 中不显示文件名下划线开头、依赖缺失、装饰器未执行直接 import 任务文件看报错用python -c import colibri.tasks.xxx验证图片压缩后文件比原文件大JPEG 转 JPEG 正常但大尺寸 PNG 可能不缩反涨对 PNG 先转 RGB 再用 JPEG 保存同时合理设置 quality检查原图是否本身已是高压缩任务重复执行cron 配置了多个入口或手动和 cron 同时跑查看 cron 列表在 history 表里查执行时间间隔日志文件增长过大没有合理轮转长时间累积配置 RotatingFileHandler定期清理 data/logs这张表是我在真实使用中整理出来的高频问题清单。大部分问题都不是代码逻辑写错而是环境、路径、并发这几类边缘情况。把这些想明白了工具用起来就会稳很多。最后再分享一点个人体会。用 colibri 一段时间后我有一个很深的感触很多看似需要复杂系统解决的问题在一台普通的电脑上用一个不到一千行的小框架就能处理得游刃有余。当你真的把任务从“临时跑一下的脚本”升级成“有配置、有插件、有历史记录的框架任务”之后整个心智负担会小很多。不要再为三五个小脚本去搭一套又重又复杂的系统了工具不在于大而在于贴身。蜂鸟虽小该采的花蜜一点都不会少。
