dupeGuru 源码解析：hscommon.jobprogress.job 作业进度管理机制（Job / NullJob 全解）

hscommon.jobprogress.job 是 dupeGuru 中统一的长任务进度管理模块，负责把"扫描、比对、删除"等耗时操作包装成可汇报进度、可被用户取消的作业单元。本文以 job.rst 所列的 Job 与 NullJob 两个公开类为主线，结合 job.py、performer.py 及 core 包中的真实调用，完整讲解其回调协议、进度加权算法、子任务拆分、取消机制与线程化执行方式。读完本文，你将能够在自己的扫描、批处理类代码中直接复用这套轻量进度框架。

一、模块定位：为什么需要 Job？

dupeGuru 中大量操作耗时较长（全盘扫描、图片指纹计算、批量删除等），需要一个通用机制让 GUI 随时知道"进展到哪了"、并允许用户中途取消。在 developer/index.rst 中可以看到，dupeGuru 内置了"generalized threaded job mechanism"：core.app.DupeGuru 持有一个 ThreadedJobPerformer 实例作为进度出口，当 App 需要执行长任务时调用 _start_job()，而平台 GUI 子类负责启动任务与刷新进度窗口。

hscommon.jobprogress.job 正是这套机制的最底层：它不依赖任何 GUI 框架，只通过一个简单的回调函数对外通报进度。其设计刻意保持"轻量"——Job 只持有一个回调，而不是维护 Parent/子对象树，代价是它"not foolproof"（源码 docstring 原话）：例如在子任务进行中又直接对父任务调用 add_progress() 并不会报错，但会打乱进度计算；这些限制由调用方自行遵守。

二、核心 API：Job 类的完整接口

Job 类的完整定义位于 hscommon/jobprogress/job.py，源码中为每个方法都写了 docstring，本文按构造、进度更新、任务划分三类逐一说明。

1. 构造：Job(job_proportions, callback)

def __init__(self, job_proportions: Union[List[int], int], callback: Callable) -> None

• job_proportions：声明你要执行几个"子作业"以及它们之间的权重比例。传整数 n 等价于 [1] * n（n 个等权重的作业）；传列表如 [2, 8] 表示两个作业，权重分别为 2 与 8（后续进度按权重加权折算）。
• callback：每次进度更新时被调用的函数。协议是 callback(progress)，若上层需要描述文本则调用 callback(progress, desc)。返回值必须为真，返回 False 表示"用户要求取消"，Job 会立刻抛出 JobCancelled。

构造时若 callback 不可调用会抛出 TypeError("'callback' MUST be set when creating a Job")。

2. 进度汇报三件套：start_job / add_progress / set_progress

def start_job(self, max_progress: int = 100, desc: str = "") -> None
def add_progress(self, progress: int = 1, desc: str = "") -> None
def set_progress(self, progress: int, desc: str = "") -> None

• start_job(max_progress, desc)：开始下一个作业。max_progress 是你在这个作业内部将要推进的工作单位上限（默认 100）；desc 是作业描述。注意：start_job 的调用次数不能超过构造时声明的作业数，超过会抛出 JobCountError。
• add_progress(n=1, desc="")：相当于 set_progress(self._progress + n, desc)，即"再前进 n 个单位"，适合在循环体内逐步推进。
• set_progress(progress, desc)：把当前作业内部进度直接设为某个值并触发回调。进度超过 max_progress 时会被钳制（self._progress > self._currmax 则置为 _currmax）。

典型用法可见 core/app.py 的 _do_delete：先 j.start_job(self.results.mark_count) 把总删除数作为工作单位上限，然后在每个待删副本的处理函数里 j.add_progress()，实现"每删一个文件进度 +1"。

3. 取消检查：check_if_cancelled()

def check_if_cancelled(self) -> None

调用它会触发一次 _do_update("")，即强制调用一次 callback。如果 callback 返回 False，则抛出 JobCancelled。适合在"无法按固定单位推进、但希望周期性检查取消标志"的场景。典型例子是目录遍历：core/directories.py 与同文件 L135 在遍历循环里调用 j.check_if_cancelled()，让超长目录扫描也能及时响应取消。

三、进度是如何算出来的？_do_update 加权算法

进度换算发生在私有方法 _do_update（job.py）：

if self._current_job:
    passed_progress = self._passed_jobs * self._currmax
    current_progress = self._current_job * self._progress
    total_progress = self._jobcount * self._currmax
    progress = ((passed_progress + current_progress) * 100) // total_progress
else:
    progress = -1  # indeterminate

• 内部状态：_jobcount = 所有作业权重之和；_passed_jobs = 已经完成的作业权重累计；_current_job = 当前作业权重；_progress = 当前作业内部进度（0～_currmax）。
• 公式本质是"已完成权重 × 当前作业满进度 + 当前权重 × 内部进度"，再除以总权重 × 满进度，乘 100 后取整，得到 0～100 的整数百分比。
• 特殊分支：当 _current_job 为 0（即尚未 start_job）时，进度传 -1，源码注释为 indeterminate（不确定态），GUI 层可据此显示"忙"而非百分比。
• 回调适配：desc 为空时只传一个参数 callback(progress)，非空时传 callback(progress, desc)，这样既兼容"只关心进度"的简单回调，也兼容带描述文本的 GUI 回调。
• 若 callback 返回假值，立即 raise JobCancelled()，取消信号由此在调用栈中向上传播。

四、子任务拆分：start_subjob 与嵌套权重

def start_subjob(self, job_proportions: Union[List[int], int], desc: str = "") -> "Job"

当你"知道子任务的数量、但不知道每个子任务内部的工作单位数"时，用 start_subjob 最合适。它的实现是：

1. 先在父任务上 start_job(100, desc)（把父任务当前这一档视为一个内部进度 0～100 的作业）；
2. 返回一个新的 Job(job_proportions, self._subjob_callback)，其中 _subjob_callback 会把子任务的进度透传给父任务：

def _subjob_callback(self, progress, desc=""):
    self.set_progress(progress, desc)
    return True  # JobCancelled 会在最上层抛出

也就是说，子 Job 的每一次进度更新都会映射为父任务内部的 set_progress()，从而实现"父进度 = 各子任务进度的加权汇总"。仓库里大量使用了这一模式：

• core/scanner.py：j = j.start_subjob([2, 8])，将扫描任务拆成"读元数据"与"比对"两个权重 2:8 的阶段；
• core/pe/matchblock.py：先 j.start_subjob([3, 7])，再在子任务内 j = j.start_subjob([9, 1], tr("Preparing for matching"))，实现两层嵌套；
• core/engine.py：j = j.start_subjob(2) 后再 sj = j.start_subjob(2)，把"建词表 → 合并相似词 → 分组"拆成多个子阶段。

Job 的 docstring 特别提醒：子任务进行期间不应再直接对父任务调 add_progress()，否则会打乱进度折算；它不阻止你这么做，只是为了保持类轻量（不需要 Parent 指针与一堆空检查）。

五、便捷迭代器：iter_with_progress

def iter_with_progress(self, iterable, desc_format=None, every=1, count=None)

这是把"遍历集合 + 推进进度"合二为一的生成器：遍历 iterable，每 every 个元素自动调用一次 add_progress(every, desc)。要点：

• 必须能取得长度：若 iterable 不是可 len() 的序列（如生成器），必须通过 count 显式指定总数，否则 len() 会失败；
• desc_format 支持 % 格式化，形如 "Prepared %d/%d files"，其中两个占位符依次为"当前已处理数、总数"；每 every 个元素更新一次描述文本，遍历结束时会以 (count, count) 收尾并 set_progress(100)；
• 返回值为生成器，用 for ... in j.iter_with_progress(...) 消费即可。

实际用法：

• core/engine.py build_word_dict：j.iter_with_progress(objects, "Prepared %d/%d files", JOB_REFRESH_RATE)，其中 JOB_REFRESH_RATE 控制每多少个文件刷新一次，避免过度回调拖慢性能；
• core/pe/matchexif.py：j.iter_with_progress(files, tr("Read EXIF of %d/%d pictures"))；
• core/pe/matchblock.py：prepare_pictures 中分析图片时使用；
• core/results.py：加载结果 XML 时 j.iter_with_progress(group_elems, every=100)，故意加大 every 以降低回调频率。

六、取消与异常体系

模块顶部定义了三个异常（job.py）：

异常	触发时机	语义
JobCancelled	callback 返回 False 时由 _do_update 抛出	用户取消了任务
JobInProgressError	已有任务在跑时又试图创建/执行任务	同一时刻只允许一个作业
JobCountError	start_job 次数超过构造时声明的作业数	作业计数越界

取消语义在 core/app.py 的 _start_job 中可见一斑：任务通过 progress_window.run(...) 启动，若上层试图在已有任务运行时再启动新任务，捕获 job.JobInProgressError 后会给用户弹提示："A previous action is still hanging in there..."（原文见源码中的 tr() 消息）。

测试侧对取消行为的验证：core/tests/app_test.py 直接构造 app.JOB = Job(1, lambda *args, **kw: False)——callback 恒返回 False，从而立即触发取消路径。

七、NullJob：空实现与默认参数惯例

class NullJob(Job):
    def __init__(self, *args, **kwargs): pass   # Null job does nothing
    # add_progress / check_if_cancelled / start_job / set_progress 全部 pass
    def start_subjob(self, *args, **kwargs) -> "NullJob":
        return NullJob()

NullJob 继承 Job 但把所有方法都做成 no-op，start_subjob 返回自身类型的新实例。模块末尾还提供了单例 nulljob = NullJob()。

它的价值在于默认参数模式：库函数把 j=job.nulljob 作为形参默认值，调用方如果不关心进度，直接省略该参数即可，代码路径零开销；关心进度的调用方传入真实 Job 即可。仓库中此类惯例遍布各处：

• core/engine.py：def build_word_dict(objects, j=job.nulljob)；
• core/engine.py：def getmatches_by_contents(files, bigsize=0, j=job.nulljob)；
• core/pe/matchblock.py：def prepare_pictures(pictures, cache_path, with_dimensions, match_rotated, j=job.nulljob)；
• core/directories.py：def get_files(self, fileclasses=None, j=job.nulljob)；
• core/results.py：from hscommon.jobprogress.job import nulljob 后 def load_from_xml(self, infile, get_file, j=nulljob)。

因此 NullJob 不是"没用的占位类"，而是让进度参数变成可选 API 的关键设计。测试基础设施同样受益：core/tests/base.py 中 JOB = nulljob 作为测试基类的默认作业，使单元测试无须关心进度回调。

八、与 ThreadedJobPerformer 的配合：真正的"后台任务"

Job 本身是同步的，真正让它跑在后台线程的是 hscommon/jobprogress/performer.py 中的 ThreadedJobPerformer。其使用模板（源码 docstring 示例）：

j = self._create_job()
self._run_threaded(self.some_work_func, (arg1, arg2, j))

关键机制：

• create_job()：若 _job_running 为真则抛 JobInProgressError；否则重置 last_progress/last_desc/job_cancelled，返回 Job(1, self._update_progress)——即一个权重为 1 的作业，其回调 _update_progress(newprogress, newdesc) 把进度写入实例属性并返回 not self.job_cancelled，GUI 层轮询或绑定这些属性即可显示进度；
• run_threaded(target, args)：同样先检查 _job_running，然后 Thread(target=self._async_run, args=(target,)+args).start()；
• _async_run(*args)：在线程内执行工作函数；JobCancelled 被静默吞掉（取消是正常流程），其他异常则记录到 last_error 与 last_traceback；finally 中复位 _job_running 并把 last_progress 置 None（表示任务结束）；
• reraise_if_error()：主线程检测到 _job_running 变回 False 后调用它，可把工作线程中的异常带 traceback 重新抛回主线程。

dupeGuru 的 GUI（如 qt 端的 progress_window）正是基于这一组合，把"进度条刷新 + 取消按钮 + 后台线程"串成完整链路，而底层进度数据全部来自 Job 的回调协议。

九、仓库中的完整调用链示例

把以上机制串起来，可以看到 dupeGuru 扫描结果集成的全貌：

1. core/scanner.py get_dupe_groups(files, ignore_list=None, j=job.nulljob) 接收外部传入的 Job；
2. 内部用 j.start_subjob(<a href="https://link.gitcode.com/i/a411655b00f8a4535c072fd902592c8f" target="_blank">2, 8]) 划分阶段，在遍历文件时用 j.iter_with_progress(files, tr("Read metadata of %d/%d files")) 报告"已读 N/M 个文件"（[core/scanner.py）；
3. 比对接入 core/engine.py getmatches_by_contents：j.start_job(len(possible_matches), PROGRESS_MESSAGE % (0, 0)) 定义本轮单位数，每次找到匹配组 j.add_progress(desc=PROGRESS_MESSAGE % (...))（L296、L315）；
4. 图片模式则进入 core/pe/matchblock.py getmatches，通过两层 start_subjob 加权分配"分析图片：比对"的进度占比；
5. 删除阶段回到 core/app.py _do_delete，以标记副本数为 start_job 单位，逐个 add_progress()。

测试端同样验证了进度行为：core/tests/engine_test.py 构造 j = job.Job(1, do_progress) 传入 getmatches 系函数，断言回调被正确触发，并注释说明"因为 iter_with_progress 使用了 every > 1，所以没有中间日志"；core/tests/scanner_test.py 也以 job.Job(1, do_progress) 驱动 get_dupe_groups。

十、使用建议与注意事项（来自源码 docstring）

• 回调必须可调用，构造时即校验；
• 不要对同一个 Job 并发调用多个作业流程：同一时刻只能有一个 start_job 处于活动状态；
• 子任务模式下，进度应只由当前子任务驱动；对父任务直接 add_progress 会破坏加权折算；
• 不要在 start_job 次数上超过构造声明的作业数（会抛 JobCountError）；
• iter_with_progress 必须保证可 len()，或显式传 count；
• 大数据集建议调大 every（如 core/results.py 的 every=100），减少回调频率以降低开销；
• 需要线程化时，优先组合 ThreadedJobPerformer.create_job() + run_threaded()，并在结束点调用 reraise_if_error() 把线程异常带回主线程。

这套设计把"进度计算、取消传播、子任务拆分"收敛在约 180 行纯 Python 内，不依赖 GUI 框架，是 dupeGuru 全平台（PE/SE/ME 三种模式）共用的长任务基础设施，非常适合作为同类桌面工具或批处理脚本的进度管理模板复用。