Archived
refactor: run_zip_job 四路终态收尾拆解（167 行 / 圈 37 → 32 行 / 圈 5）
为什么：run_zip_job 是全仓圈复杂度最高的函数（37），167 行里四个终态——正常交付、
全部失败、取消（含部分交付/全额退两条支路）、异常（同样两条支路）——各写一段
8~20 行的收尾内联在 try/except 里，且都伸手去读同一批局部变量。上一轮抽这个引擎时
就已指出「还可以再拆」，当时为控制单次改动面留了下来。
改法：把「一次运行」的状态收进 _ZipRun，四个终态各自成函数。
_ZipRun job / task_id / settle / tally / task / params /
cancel_ev / pause_ev / items / zip_tmp_path / zip_valid / t0
_progress_of 构造进度回调（原先是个内嵌闭包）
_start_zip_run 置 running + 组装清单 + 写 total + 计费补扣
_build_zip_archive 建临时 ZIP + 逐窗口填充 + 置 zip_valid + 记耗时
_publish_delivery 正常交付（结算 → 锁内复查 → 发布 result → on_delivered → 日志）
_finish_all_failed 全失败（丢弃空包 + 全额退费 + 转 error）
_finish_canceled 取消（部分交付 / 全额退两条支路）
_finish_business_error DownloadException（保留我们自己拼的用户可读文案）
_finish_unexpected 通用异常（先落 traceback 再泛化文案，两条支路）
复杂度（同口径 AST 实测，HEAD 与工作区由同一脚本测出）：
run_zip_job 167 行 / 圈 37 / 嵌套 4 → 32 行 / 圈 5 / 嵌套 2
文件内最大圈复杂度 37（就是它） → 13（_fill_zip，本次未动）
文件行数 439 → 522（+83，全是新签名与 docstring）· 函数数 19 → 27
顺手消掉一个**隐患**，不是纯搬家：``zip_valid`` 原先必须手工在函数头初始化为
False（注释写明「漏了就 NameError 二次崩线程」——它正是上上轮那个 bug 的残留
护栏）。收进 dataclass 后该约束由构造函数承担，而 ``_ZipRun(...)`` 的默认值就是
假值，物理上写不出「忘记初始化」。
等价性证据（两道，互相独立）：
1. 不靠注入脚本，直接把文件还原成 HEAD 版跑同一套测试（git stash push 单文件 →
跑 → pop）：tests/test_zip_workers.py + test_monitor_jobs.py 共 33 例**全过**。
这批测试是上一轮用反向注入验证过「忠实编码旧行为」的，故本轮它们对新实现
构成等价性验证。
2. AST 字面量多重集比对：字符串 56 种**一个不少**——没有任何用户可见文案或日志
格式被动过。数值唯一差异是 `0.0: 0 → 1`，即 _ZipRun.t0 的字段默认值。
刻意保留的语义（是决定，不是疏漏）：
- _publish_delivery 里「先 settle 再抛 DownloadCanceled」的顺序原样——抛出去后
_finish_canceled 会**再 settle 一次**（delivered 口径不同），依赖的是 ledger_settle
幂等；原实现同样如此，未动。
- 取消与异常两条支路里「部分交付」与「全额退」的判定条件（`delivered > 0 and
zip_tmp_path`）逐字保留。
- _finish_all_failed 不走 _apply_error：它要额外记 failed 计数，口径与 error 支路
不同（原实现即如此）。
- record_duration 仍在 zip_valid 置位之后、全失败判定之前（保持原顺序）。
验证：ruff 全过 · pyright 0/0 · pytest 294 passed（与改动前同数）· 生产 data/ 未被
触碰（全部用例走 tmp_path 沙箱）。
This commit is contained in:
snowy-yang committed
1 parent
9cbd341e2e
commit
da83201403
1 file changed
+177
-94
+177
-94
@@ -19,6 +19,10 @@
|
|||||||
:class:`zipfile.ZipFile` 不是线程安全的，且去重读改写 ``used_arcnames``
|
:class:`zipfile.ZipFile` 不是线程安全的，且去重读改写 ``used_arcnames``
|
||||||
需要串行——故下载与落包被拆成 ``download_one`` / ``write_entry`` 两个钩子，
|
需要串行——故下载与落包被拆成 ``download_one`` / ``write_entry`` 两个钩子，
|
||||||
只有前者进线程池。
|
只有前者进线程池。
|
||||||
|
|
||||||
|
``run_zip_job`` 的四个终态（成功 / 全失败 / 取消 / 异常）原先各写一段 8~20 行的
|
||||||
|
收尾内联在 try/except 里。此处把「一次运行」的状态收进 :class:`_ZipRun`，
|
||||||
|
四个终态各自成函数——改账本口径或补一种终态时，不必再在 167 行里找落点。
|
||||||
"""
|
"""
|
||||||
import os
|
import os
|
||||||
import tempfile
|
import tempfile
|
||||||
@@ -190,93 +194,128 @@ def _discard(path: str | None) -> None:
|
|||||||
pass
|
pass
|
||||||
|
|
||||||
|
|
||||||
def run_zip_job(job: ZipJob, task_id: str, *,
|
@dataclass
|
||||||
settle: Callable[[str, int, str], int] = _settle_batch_billing) -> None:
|
class _ZipRun:
|
||||||
"""执行一次批量打包任务（在 worker 线程内调用，自身不向外抛异常）。
|
"""一次 ``run_zip_job`` 的共享状态。
|
||||||
|
|
||||||
``settle`` 由调用方注入，默认走 :func:`services._taskcore._settle_batch_billing`。
|
四个终态收尾函数都要读同一批东西（任务 id、结算入口、计数、临时 ZIP 路径、
|
||||||
做成参数是为了让测试能在**模块级**替换结算函数（包装层在调用时读取模块全局，
|
ZIP 是否完整）。原先它们是散在 167 行函数体里的局部变量，被各 except 分支
|
||||||
替换即生效）。
|
分别访问——收成一个对象后，收尾函数不必逐个透传参数，也让「谁真的用到了
|
||||||
|
哪些状态」一眼可见。
|
||||||
"""
|
"""
|
||||||
zip_tmp_path: str | None = None
|
|
||||||
# zip_valid 必须在函数头初始化：ZIP 构建前的业务错误（参数缺失/无条目/
|
|
||||||
# 补扣失败）会 early raise，异常处理器若引用未绑定名会 NameError 二次崩线程
|
|
||||||
zip_valid = False
|
|
||||||
tally = _Tally()
|
|
||||||
items: list[Any] = []
|
|
||||||
try:
|
|
||||||
with DOWNLOAD_TASKS_LOCK:
|
|
||||||
task = DOWNLOAD_TASKS[task_id]
|
|
||||||
task['status'] = 'running'
|
|
||||||
task['msg'] = job.start_msg
|
|
||||||
params = task['params']
|
|
||||||
cancel_ev = task['cancel']
|
|
||||||
pause_ev = task['pause']
|
|
||||||
# 计耗时只算「下载+打包」这一段（不含排队），故不用 task['start']
|
|
||||||
t0 = time.time()
|
|
||||||
|
|
||||||
items = job.prepare(params, task_id, tally.failed)
|
job: ZipJob
|
||||||
with DOWNLOAD_TASKS_LOCK:
|
task_id: str
|
||||||
DOWNLOAD_TASKS[task_id]['total'] = len(items)
|
settle: Callable[[str, int, str], int]
|
||||||
job.supplement(task_id, params, len(items))
|
tally: _Tally = field(default_factory=_Tally)
|
||||||
|
task: dict[str, Any] = field(default_factory=dict)
|
||||||
|
params: dict[str, Any] = field(default_factory=dict)
|
||||||
|
cancel_ev: threading.Event | None = None
|
||||||
|
pause_ev: threading.Event | None = None
|
||||||
|
items: list[Any] = field(default_factory=list)
|
||||||
|
zip_tmp_path: str | None = None
|
||||||
|
zip_valid: bool = False
|
||||||
|
t0: float = 0.0
|
||||||
|
|
||||||
|
|
||||||
|
def _progress_of(run: _ZipRun) -> Callable[..., None]:
|
||||||
|
"""构造写任务进度的回调（窗口内每完成一条调用一次）。"""
|
||||||
|
|
||||||
def _progress(msg: str = '') -> None:
|
def _progress(msg: str = '') -> None:
|
||||||
with DOWNLOAD_TASKS_LOCK:
|
with DOWNLOAD_TASKS_LOCK:
|
||||||
t = DOWNLOAD_TASKS[task_id]
|
t = DOWNLOAD_TASKS[run.task_id]
|
||||||
t['done'] = tally.ok + len(tally.failed)
|
t['done'] = run.tally.ok + len(run.tally.failed)
|
||||||
t['failed'] = len(tally.failed)
|
t['failed'] = len(run.tally.failed)
|
||||||
if msg:
|
if msg:
|
||||||
t['msg'] = msg
|
t['msg'] = msg
|
||||||
|
|
||||||
# ZIP 构建放 downloads 卷（原实现落系统 %TEMP%——500 首无损可达 GB 级，
|
return _progress
|
||||||
# 小系统盘一次大单即塞爆），且建在任务专属目录（见 _task_work_dir）：
|
|
||||||
# downloads 根目录会被缓存清扫按 mtime 回收
|
|
||||||
work_dir = _task_work_dir(task_id)
|
def _start_zip_run(run: _ZipRun) -> None:
|
||||||
|
"""置 running、组装清单、写回 total、执行计费补扣。"""
|
||||||
|
with DOWNLOAD_TASKS_LOCK:
|
||||||
|
task = DOWNLOAD_TASKS[run.task_id]
|
||||||
|
task['status'] = 'running'
|
||||||
|
task['msg'] = run.job.start_msg
|
||||||
|
run.task = task
|
||||||
|
run.params = task['params']
|
||||||
|
run.cancel_ev = task['cancel']
|
||||||
|
run.pause_ev = task['pause']
|
||||||
|
# 计耗时只算「下载+打包」这一段（不含排队），故不用 task['start']
|
||||||
|
run.t0 = time.time()
|
||||||
|
|
||||||
|
run.items = run.job.prepare(run.params, run.task_id, run.tally.failed)
|
||||||
|
with DOWNLOAD_TASKS_LOCK:
|
||||||
|
DOWNLOAD_TASKS[run.task_id]['total'] = len(run.items)
|
||||||
|
run.job.supplement(run.task_id, run.params, len(run.items))
|
||||||
|
|
||||||
|
|
||||||
|
def _build_zip_archive(run: _ZipRun) -> None:
|
||||||
|
"""建临时 ZIP 并逐窗口填充。返回时 ``zip_valid`` 才置位。
|
||||||
|
|
||||||
|
ZIP 构建放 downloads 卷（原实现落系统 %TEMP%——500 首无损可达 GB 级，
|
||||||
|
小系统盘一次大单即塞爆），且建在任务专属目录（见 ``_task_work_dir``）：
|
||||||
|
downloads 根目录会被缓存清扫按 mtime 回收。
|
||||||
|
"""
|
||||||
|
job = run.job
|
||||||
|
assert run.cancel_ev is not None and run.pause_ev is not None
|
||||||
|
work_dir = _task_work_dir(run.task_id)
|
||||||
with tempfile.NamedTemporaryFile(prefix=job.tmp_prefix, suffix='.zip',
|
with tempfile.NamedTemporaryFile(prefix=job.tmp_prefix, suffix='.zip',
|
||||||
dir=str(work_dir), delete=False) as zip_tmp:
|
dir=str(work_dir), delete=False) as zip_tmp:
|
||||||
zip_tmp_path = zip_tmp.name
|
run.zip_tmp_path = zip_tmp.name
|
||||||
# 音频本身已压缩，条目按 ZIP_STORED（免压缩）大幅提速；文本条目仍压缩
|
# 音频本身已压缩，条目按 ZIP_STORED（免压缩）大幅提速；文本条目仍压缩
|
||||||
with zipfile.ZipFile(zip_tmp, 'w', zipfile.ZIP_DEFLATED) as zf:
|
with zipfile.ZipFile(zip_tmp, 'w', zipfile.ZIP_DEFLATED) as zf:
|
||||||
ctx = ZipContext(task_id=task_id, params=params, cancel_ev=cancel_ev,
|
ctx = ZipContext(task_id=run.task_id, params=run.params,
|
||||||
work_dir=work_dir, zf=zf, progress=_progress,
|
cancel_ev=run.cancel_ev, work_dir=work_dir, zf=zf,
|
||||||
total=len(items))
|
progress=_progress_of(run), total=len(run.items))
|
||||||
_fill_zip(job, items, ctx, tally, task, pause_ev)
|
_fill_zip(job, run.items, ctx, run.tally, run.task, run.pause_ev)
|
||||||
# zip_valid 只在 ZIP with 块正常退出（中央目录完整写出）后才置 True。
|
# zip_valid 只在 ZIP with 块正常退出（中央目录完整写出）后才置 True。
|
||||||
# 此前两个 worker 在这之后还各留了一段 `if not zip_valid:` 失败处理——
|
# 此前两个 worker 在这之后还各留了一段 `if not zip_valid:` 失败处理——
|
||||||
# 紧跟赋值、中间无任何可改zip_valid的语句，是**不可达**的复审残留，已删。
|
# 紧跟赋值、中间无任何可改zip_valid的语句，是**不可达**的复审残留，已删。
|
||||||
zip_valid = True
|
run.zip_valid = True
|
||||||
|
|
||||||
if job.record_duration and tally.ok > 0:
|
if job.record_duration and run.tally.ok > 0:
|
||||||
with BATCH_DURATIONS_LOCK:
|
with BATCH_DURATIONS_LOCK:
|
||||||
BATCH_DURATIONS.append(time.time() - t0)
|
BATCH_DURATIONS.append(time.time() - run.t0)
|
||||||
|
|
||||||
# 全部失败（含清单非空但一条都没成）：不发布空 ZIP、不按交付结算，
|
|
||||||
# 转 error 并全额退费。原音乐批量用 `ok==0 and failed`、播客用早抛
|
def _finish_all_failed(run: _ZipRun) -> None:
|
||||||
# DownloadException——这里统一为显式收敛，两边的账本终态一致
|
"""全部失败（含清单非空但一条都没成）：不发布空 ZIP、不按交付结算，转 error 并全额退费。
|
||||||
if tally.ok == 0:
|
|
||||||
_discard(zip_tmp_path)
|
原音乐批量用 ``ok==0 and failed``、播客用早抛 ``DownloadException``——这里
|
||||||
zip_tmp_path = None
|
统一为显式收敛，两边的账本终态一致。
|
||||||
refunded = settle(task_id, 0, 'error')
|
"""
|
||||||
_mark_task_terminal(task_id, 'error', job.all_failed_msg(tally.failed, refunded))
|
job, tally = run.job, run.tally
|
||||||
|
_discard(run.zip_tmp_path)
|
||||||
|
run.zip_tmp_path = None
|
||||||
|
refunded = run.settle(run.task_id, 0, 'error')
|
||||||
|
_mark_task_terminal(run.task_id, 'error', job.all_failed_msg(tally.failed, refunded))
|
||||||
with DOWNLOAD_TASKS_LOCK:
|
with DOWNLOAD_TASKS_LOCK:
|
||||||
t = DOWNLOAD_TASKS.get(task_id)
|
t = DOWNLOAD_TASKS.get(run.task_id)
|
||||||
if t is not None:
|
if t is not None:
|
||||||
t['failed'] = len(tally.failed)
|
t['failed'] = len(tally.failed)
|
||||||
t['refunded'] = refunded
|
t['refunded'] = refunded
|
||||||
api_service.logger.info(
|
api_service.logger.info(
|
||||||
f"{job.tag}全部失败: {task_id} 失败{len(tally.failed)} 退回{refunded}次")
|
f"{job.tag}全部失败: {run.task_id} 失败{len(tally.failed)} 退回{refunded}次")
|
||||||
return
|
|
||||||
|
|
||||||
# 正常交付：按入包成功数结算（'settled' 是 _settle_batch_billing 的默认值，
|
|
||||||
# 此处显式写出，让「成功/取消/失败」三种账本终态在本文件里一目了然）
|
def _publish_delivery(run: _ZipRun) -> None:
|
||||||
refunded = settle(task_id, tally.ok, 'settled')
|
"""正常交付：按入包成功数结算，锁内复查取消/终态后发布结果。
|
||||||
|
|
||||||
|
``'settled'`` 是 ``_settle_batch_billing`` 的默认值，此处显式写出，让
|
||||||
|
「成功/取消/失败」三种账本终态在本文件里一目了然。
|
||||||
|
"""
|
||||||
|
job, tally = run.job, run.tally
|
||||||
|
assert run.cancel_ev is not None
|
||||||
|
refunded = run.settle(run.task_id, tally.ok, 'settled')
|
||||||
with DOWNLOAD_TASKS_LOCK:
|
with DOWNLOAD_TASKS_LOCK:
|
||||||
t = DOWNLOAD_TASKS.get(task_id)
|
t = DOWNLOAD_TASKS.get(run.task_id)
|
||||||
# 发布前锁内复查取消/失效状态——取消请求落在「检查后、拿锁前」的窗口
|
# 发布前锁内复查取消/失效状态——取消请求落在「检查后、拿锁前」的窗口
|
||||||
# 也不得覆盖终态
|
# 也不得覆盖终态
|
||||||
if t is None or t.get('status') in ('canceled', 'error'):
|
if t is None or t.get('status') in ('canceled', 'error'):
|
||||||
raise DownloadCanceled("任务已取消")
|
raise DownloadCanceled("任务已取消")
|
||||||
if cancel_ev.is_set():
|
if run.cancel_ev.is_set():
|
||||||
raise DownloadCanceled("任务已取消")
|
raise DownloadCanceled("任务已取消")
|
||||||
t['status'] = 'done'
|
t['status'] = 'done'
|
||||||
t['completed_at'] = time.time()
|
t['completed_at'] = time.time()
|
||||||
@@ -286,74 +325,118 @@ def run_zip_job(job: ZipJob, task_id: str, *,
|
|||||||
f" 退回{refunded}次" if refunded else '')
|
f" 退回{refunded}次" if refunded else '')
|
||||||
t['refunded'] = refunded
|
t['refunded'] = refunded
|
||||||
t['result'] = {
|
t['result'] = {
|
||||||
'zip_path': zip_tmp_path,
|
'zip_path': run.zip_tmp_path,
|
||||||
'filename': job.zip_filename(tally.ok),
|
'filename': job.zip_filename(tally.ok),
|
||||||
'audio_files': [], # 缓存文件不归任务所有，不进入清理清单
|
'audio_files': [], # 缓存文件不归任务所有，不进入清理清单
|
||||||
}
|
}
|
||||||
job.on_delivered(items, task_id, params)
|
job.on_delivered(run.items, run.task_id, run.params)
|
||||||
api_service.logger.info(
|
api_service.logger.info(
|
||||||
f"{job.tag}完成: {job.log_label()}成功{tally.ok} 失败{len(tally.failed)} "
|
f"{job.tag}完成: {job.log_label()}成功{tally.ok} 失败{len(tally.failed)} "
|
||||||
f"退回{refunded}次 ({task_id})")
|
f"退回{refunded}次 ({run.task_id})")
|
||||||
except DownloadCanceled:
|
|
||||||
# 部分成功后取消——已入包部分照常交付（成果守恒），未完成部分按账本退费。
|
|
||||||
# ZIP 未正常完成（写坏）时不按 ok_count 交付——按全额退费收敛
|
def _finish_canceled(run: _ZipRun) -> None:
|
||||||
delivered = tally.ok if zip_valid else 0
|
"""部分成功后取消——已入包部分照常交付（成果守恒），未完成部分按账本退费。
|
||||||
refunded = settle(task_id, delivered, 'canceled')
|
|
||||||
if delivered > 0 and zip_tmp_path:
|
ZIP 未正常完成（写坏）时不按 ``ok_count`` 交付——按全额退费收敛。
|
||||||
_mark_task_terminal(task_id, 'canceled',
|
"""
|
||||||
|
job, tally = run.job, run.tally
|
||||||
|
delivered = tally.ok if run.zip_valid else 0
|
||||||
|
refunded = run.settle(run.task_id, delivered, 'canceled')
|
||||||
|
if delivered > 0 and run.zip_tmp_path:
|
||||||
|
_mark_task_terminal(run.task_id, 'canceled',
|
||||||
f"已取消（已打包 {tally.ok} {job.noun}可领取）" +
|
f"已取消（已打包 {tally.ok} {job.noun}可领取）" +
|
||||||
(f"，退回 {refunded} 次" if refunded else ''))
|
(f"，退回 {refunded} 次" if refunded else ''))
|
||||||
with DOWNLOAD_TASKS_LOCK:
|
with DOWNLOAD_TASKS_LOCK:
|
||||||
t = DOWNLOAD_TASKS.get(task_id)
|
t = DOWNLOAD_TASKS.get(run.task_id)
|
||||||
if t is not None:
|
if t is not None:
|
||||||
t['refunded'] = refunded
|
t['refunded'] = refunded
|
||||||
t['result'] = {
|
t['result'] = {
|
||||||
'zip_path': zip_tmp_path,
|
'zip_path': run.zip_tmp_path,
|
||||||
'filename': job.zip_filename(tally.ok),
|
'filename': job.zip_filename(tally.ok),
|
||||||
'audio_files': [],
|
'audio_files': [],
|
||||||
}
|
}
|
||||||
else:
|
else:
|
||||||
_discard(zip_tmp_path)
|
_discard(run.zip_tmp_path)
|
||||||
_mark_task_terminal(task_id, 'canceled',
|
_mark_task_terminal(run.task_id, 'canceled',
|
||||||
'任务已取消' + (
|
'任务已取消' + (
|
||||||
f"（已退回 {refunded} 次下载次数）" if refunded else ''))
|
f"（已退回 {refunded} 次下载次数）" if refunded else ''))
|
||||||
with DOWNLOAD_TASKS_LOCK:
|
with DOWNLOAD_TASKS_LOCK:
|
||||||
t = DOWNLOAD_TASKS.get(task_id)
|
t = DOWNLOAD_TASKS.get(run.task_id)
|
||||||
if t is not None:
|
if t is not None:
|
||||||
t['refunded'] = refunded
|
t['refunded'] = refunded
|
||||||
api_service.logger.info(
|
api_service.logger.info(
|
||||||
f"{job.tag}已取消: {task_id}（成功{tally.ok} 退回{refunded}次）")
|
f"{job.tag}已取消: {run.task_id}（成功{tally.ok} 退回{refunded}次）")
|
||||||
except DownloadException as e:
|
|
||||||
# 业务失败（清单阶段：无可下载条目 / 补扣下载次数失败 / 上游不可用）。
|
|
||||||
# 文案是我们自己拼的用户可读文本，保留原文（同播客单集下载）
|
def _finish_business_error(run: _ZipRun, exc: DownloadException) -> None:
|
||||||
delivered = tally.ok if zip_valid else 0
|
"""业务失败（清单阶段：无可下载条目 / 补扣下载次数失败 / 上游不可用）。
|
||||||
refunded = settle(task_id, delivered, 'error')
|
|
||||||
_apply_error(task_id, str(e), refunded)
|
文案是我们自己拼的用户可读文本，保留原文（同播客单集下载）。
|
||||||
_discard(zip_tmp_path)
|
"""
|
||||||
|
job, tally = run.job, run.tally
|
||||||
|
delivered = tally.ok if run.zip_valid else 0
|
||||||
|
refunded = run.settle(run.task_id, delivered, 'error')
|
||||||
|
_apply_error(run.task_id, str(exc), refunded)
|
||||||
|
_discard(run.zip_tmp_path)
|
||||||
api_service.logger.error(
|
api_service.logger.error(
|
||||||
f"{job.tag}失败: {task_id} {e}（成功{tally.ok} 退回{refunded}次）")
|
f"{job.tag}失败: {run.task_id} {exc}（成功{tally.ok} 退回{refunded}次）")
|
||||||
except Exception as e:
|
|
||||||
# 非预期异常原文 logger 先行（结算本身可能再抛，先落下 traceback），
|
|
||||||
# 用户可见文案泛化；带上退款结果再补一条，避免排障时只看到「异常」
|
def _finish_unexpected(run: _ZipRun, exc: Exception) -> None:
|
||||||
api_service.logger.error(f"{job.tag}异常: {task_id} {e}\n{traceback.format_exc()}")
|
"""非预期异常：原文 logger 先行（结算本身可能再抛，先落下 traceback），
|
||||||
delivered = tally.ok if zip_valid else 0
|
用户可见文案泛化；带上退款结果再补一条，避免排障时只看到「异常」。"""
|
||||||
refunded = settle(task_id, delivered, 'error')
|
job, tally = run.job, run.tally
|
||||||
if delivered > 0 and zip_tmp_path:
|
api_service.logger.error(f"{job.tag}异常: {run.task_id} {exc}\n{traceback.format_exc()}")
|
||||||
_apply_error(task_id, job.partial_msg(refunded), refunded)
|
delivered = tally.ok if run.zip_valid else 0
|
||||||
|
refunded = run.settle(run.task_id, delivered, 'error')
|
||||||
|
if delivered > 0 and run.zip_tmp_path:
|
||||||
|
_apply_error(run.task_id, job.partial_msg(refunded), refunded)
|
||||||
with DOWNLOAD_TASKS_LOCK:
|
with DOWNLOAD_TASKS_LOCK:
|
||||||
t = DOWNLOAD_TASKS.get(task_id)
|
t = DOWNLOAD_TASKS.get(run.task_id)
|
||||||
if t is not None:
|
if t is not None:
|
||||||
t['result'] = {
|
t['result'] = {
|
||||||
'zip_path': zip_tmp_path,
|
'zip_path': run.zip_tmp_path,
|
||||||
'filename': job.zip_filename(tally.ok),
|
'filename': job.zip_filename(tally.ok),
|
||||||
'audio_files': [],
|
'audio_files': [],
|
||||||
}
|
}
|
||||||
else:
|
else:
|
||||||
_apply_error(task_id, '打包失败，请稍后重试' + (
|
_apply_error(run.task_id, '打包失败，请稍后重试' + (
|
||||||
f"（已退回 {refunded} 次下载次数）" if refunded else ''), refunded)
|
f"（已退回 {refunded} 次下载次数）" if refunded else ''), refunded)
|
||||||
_discard(zip_tmp_path)
|
_discard(run.zip_tmp_path)
|
||||||
api_service.logger.error(
|
api_service.logger.error(
|
||||||
f"{job.tag}异常: {task_id}（成功{tally.ok} 退回{refunded}次）")
|
f"{job.tag}异常: {run.task_id}（成功{tally.ok} 退回{refunded}次）")
|
||||||
|
|
||||||
|
|
||||||
|
def run_zip_job(job: ZipJob, task_id: str, *,
|
||||||
|
settle: Callable[[str, int, str], int] = _settle_batch_billing) -> None:
|
||||||
|
"""执行一次批量打包任务（在 worker 线程内调用，自身不向外抛异常）。
|
||||||
|
|
||||||
|
``settle`` 由调用方注入，默认走 :func:`services._taskcore._settle_batch_billing`。
|
||||||
|
做成参数是为了让测试能在**模块级**替换结算函数（包装层在调用时读取模块全局，
|
||||||
|
替换即生效）。
|
||||||
|
|
||||||
|
``zip_valid`` 在这里的等价物是 ``run.zip_valid``，它**在构建前必为假**——
|
||||||
|
ZIP 构建前的业务错误（参数缺失/无条目/补扣失败）会 early raise，收尾函数
|
||||||
|
据此不按 ``ok_count`` 交付。原先它是必须手工在函数头初始化的局部变量，
|
||||||
|
漏掉就让异常处理器 NameError 二次崩线程；收进 dataclass 后这个约束由构造
|
||||||
|
函数承担。
|
||||||
|
"""
|
||||||
|
run = _ZipRun(job=job, task_id=task_id, settle=settle)
|
||||||
|
try:
|
||||||
|
_start_zip_run(run)
|
||||||
|
_build_zip_archive(run)
|
||||||
|
|
||||||
|
if run.tally.ok == 0:
|
||||||
|
_finish_all_failed(run)
|
||||||
|
return
|
||||||
|
_publish_delivery(run)
|
||||||
|
except DownloadCanceled:
|
||||||
|
_finish_canceled(run)
|
||||||
|
except DownloadException as e:
|
||||||
|
_finish_business_error(run, e)
|
||||||
|
except Exception as e:
|
||||||
|
_finish_unexpected(run, e)
|
||||||
finally:
|
finally:
|
||||||
# 回收信息预取线程池，避免线程泄漏（未启动的预取任务一并取消）
|
# 回收信息预取线程池，避免线程泄漏（未启动的预取任务一并取消）
|
||||||
job.cleanup()
|
job.cleanup()
|
||||||
|
|||||||
Reference in new issue
Block a user