Commit af31f0a885999c0cdf677028c75fdec886708394
1 parent
25a489dc
h3-console: 加批量取消/删除接口
清理过载事故留下的重复排队任务时发现,取消/删除都只能按单条 id 调用,
脚本化清理只能一个个来。新增 POST /v1/videos/batch-cancel 和
/batch-delete,body 传 {"ids":[...]} 或 {status,kind,q} 筛选条件二选一;
筛选条件命中数超过 300 直接 400,避免条件写宽了误伤一大片。单条 cancel/
delete 的核心逻辑抽成 _cancel_job/_delete_job 供两边复用,批量场景里
一条失败(比如已经是终态)不影响其它条,成功/失败分别列出来。
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01GFw9ZBh3eVjssk6quYQ4Ki
Showing
1 changed file
with
86 additions
and
5 deletions
h3-console/server.py
| ... | ... | @@ -396,8 +396,10 @@ async def create_video(req: dict): |
| 396 | 396 | return _row_view(row) |
| 397 | 397 | |
| 398 | 398 | |
| 399 | -@api.post("/v1/videos/{jid}/cancel") | |
| 400 | -async def cancel_video(jid: str): | |
| 399 | +async def _cancel_job(jid: str) -> dict: | |
| 400 | + """真正干活的取消/停止逻辑。单条接口和下面的批量接口共用——批量场景需要一条失败不 | |
| 401 | + 连累其它条,所以这里只管抛 HTTPException,由调用方决定是直接透传(单条)还是 | |
| 402 | + catch 下来记进 failed 列表(批量)。""" | |
| 401 | 403 | with db() as con: |
| 402 | 404 | row = con.execute("SELECT * FROM jobs WHERE id=?", (jid,)).fetchone() |
| 403 | 405 | if row is None: |
| ... | ... | @@ -434,6 +436,11 @@ async def cancel_video(jid: str): |
| 434 | 436 | return _row_view(row) |
| 435 | 437 | |
| 436 | 438 | |
| 439 | +@api.post("/v1/videos/{jid}/cancel") | |
| 440 | +async def cancel_video(jid: str): | |
| 441 | + return await _cancel_job(jid) | |
| 442 | + | |
| 443 | + | |
| 437 | 444 | @api.post("/v1/videos/{jid}/prioritize") |
| 438 | 445 | async def prioritize_video(jid: str): |
| 439 | 446 | with db() as con: |
| ... | ... | @@ -588,9 +595,8 @@ async def list_videos(limit: int = 20, offset: int = 0, trash: bool = False, q: |
| 588 | 595 | return {"jobs": [_row_view(r) for r in rows], "total": total} |
| 589 | 596 | |
| 590 | 597 | |
| 591 | -@api.delete("/v1/videos/{jid}") | |
| 592 | -async def delete_video(jid: str): | |
| 593 | - """软删除:只挪出主列表,不碰 gateway/api.py 上的任务本身(可能还在跑),不删磁盘文件。""" | |
| 598 | +def _delete_job(jid: str) -> dict: | |
| 599 | + """真正干活的软删除逻辑,单条/批量共用,理由同 _cancel_job。""" | |
| 594 | 600 | with db() as con: |
| 595 | 601 | cur = con.execute("UPDATE jobs SET deleted_at=? WHERE id=? AND deleted_at IS NULL", (time.time(), jid)) |
| 596 | 602 | if cur.rowcount == 0: |
| ... | ... | @@ -598,6 +604,12 @@ async def delete_video(jid: str): |
| 598 | 604 | return {"id": jid, "deleted": True} |
| 599 | 605 | |
| 600 | 606 | |
| 607 | +@api.delete("/v1/videos/{jid}") | |
| 608 | +async def delete_video(jid: str): | |
| 609 | + """软删除:只挪出主列表,不碰 gateway/api.py 上的任务本身(可能还在跑),不删磁盘文件。""" | |
| 610 | + return _delete_job(jid) | |
| 611 | + | |
| 612 | + | |
| 601 | 613 | @api.post("/v1/videos/{jid}/restore") |
| 602 | 614 | async def restore_video(jid: str): |
| 603 | 615 | with db() as con: |
| ... | ... | @@ -607,6 +619,75 @@ async def restore_video(jid: str): |
| 607 | 619 | return {"id": jid, "deleted": False} |
| 608 | 620 | |
| 609 | 621 | |
| 622 | +# 2026-09-05 加:一次过载事故后要清理一批内容重复的排队任务,才发现只有按单条 id 操作的 | |
| 623 | +# cancel/delete,脚本化清理只能一个个调,慢且容易漏。下面两个批量接口给 ids 列表和"筛选 | |
| 624 | +# 条件"两种入参二选一——ids 优先且没有数量上限(调用方既然把 id 一个个列出来了,就是明确 | |
| 625 | +# 知道要动哪些);筛选条件复用跟 GET /v1/videos 一样的 status/kind/q 语义,但设了 | |
| 626 | +# _BATCH_MAX_BY_FILTER 上限,防止筛选条件写宽了误伤一大片跟预期不符的任务——命中数超过 | |
| 627 | +# 上限直接 400,让调用方缩小范围或换成显式 ids,而不是静默截断一部分。 | |
| 628 | +_BATCH_MAX_BY_FILTER = 300 | |
| 629 | + | |
| 630 | + | |
| 631 | +def _resolve_batch_targets(body: dict) -> list[str]: | |
| 632 | + ids = body.get("ids") | |
| 633 | + if ids: | |
| 634 | + return list(dict.fromkeys(ids)) # 去重但保留顺序,防止调用方列表里手滑写重了 | |
| 635 | + status, kind, q = body.get("status") or "", body.get("kind") or "", body.get("q") or "" | |
| 636 | + if not (status or kind or q): | |
| 637 | + raise HTTPException(400, "必须提供 ids 列表,或至少一个筛选条件(status/kind/q),避免误操作全部任务") | |
| 638 | + where = ["deleted_at IS NULL"] | |
| 639 | + params: list = [] | |
| 640 | + if q: | |
| 641 | + where.append("(id LIKE ? OR req_json LIKE ?)") | |
| 642 | + params += [f"%{q}%", f"%{q}%"] | |
| 643 | + if status: | |
| 644 | + where.append("status = ?") | |
| 645 | + params.append(status) | |
| 646 | + if kind: | |
| 647 | + where.append("partition = ?") | |
| 648 | + params.append(kind) | |
| 649 | + with db() as con: | |
| 650 | + total = con.execute(f"SELECT COUNT(*) FROM jobs WHERE {' AND '.join(where)}", params).fetchone()[0] | |
| 651 | + if total > _BATCH_MAX_BY_FILTER: | |
| 652 | + raise HTTPException( | |
| 653 | + 400, | |
| 654 | + f"筛选条件命中 {total} 条,超过单次批量操作上限 {_BATCH_MAX_BY_FILTER}," | |
| 655 | + "请缩小筛选范围(加 status/kind/q)或改成显式 ids 列表", | |
| 656 | + ) | |
| 657 | + rows = con.execute(f"SELECT id FROM jobs WHERE {' AND '.join(where)}", params).fetchall() | |
| 658 | + return [r["id"] for r in rows] | |
| 659 | + | |
| 660 | + | |
| 661 | +@api.post("/v1/videos/batch-cancel") | |
| 662 | +async def batch_cancel_videos(body: dict): | |
| 663 | + """批量取消/停止。body 传 {"ids": [...]} 或 {"status": "queued", "kind": "...", "q": "..."} | |
| 664 | + 筛选条件(至少给一个)。逐条按现有单条 cancel 语义处理,某一条已经是终态之类的失败不影响 | |
| 665 | + 其它条,成功/失败分别列出来。""" | |
| 666 | + ids = _resolve_batch_targets(body) | |
| 667 | + succeeded, failed = [], [] | |
| 668 | + for jid in ids: | |
| 669 | + try: | |
| 670 | + succeeded.append((await _cancel_job(jid))["id"]) | |
| 671 | + except HTTPException as e: | |
| 672 | + failed.append({"id": jid, "error": e.detail}) | |
| 673 | + return {"requested": len(ids), "succeeded": succeeded, "failed": failed} | |
| 674 | + | |
| 675 | + | |
| 676 | +@api.post("/v1/videos/batch-delete") | |
| 677 | +async def batch_delete_videos(body: dict): | |
| 678 | + """批量软删除(移入回收站),语义同单条 DELETE /v1/videos/{jid}——只是把任务从列表里挪走, | |
| 679 | + 不取消底层计算;真要连底层任务一起停掉,配合上面的 batch-cancel 一起调(谁先谁后不影响 | |
| 680 | + 结果,两个操作互不冲突)。""" | |
| 681 | + ids = _resolve_batch_targets(body) | |
| 682 | + succeeded, failed = [], [] | |
| 683 | + for jid in ids: | |
| 684 | + try: | |
| 685 | + succeeded.append(_delete_job(jid)["id"]) | |
| 686 | + except HTTPException as e: | |
| 687 | + failed.append({"id": jid, "error": e.detail}) | |
| 688 | + return {"requested": len(ids), "succeeded": succeeded, "failed": failed} | |
| 689 | + | |
| 690 | + | |
| 610 | 691 | @api.get("/v1/backends") |
| 611 | 692 | async def backends(): |
| 612 | 693 | resp = await client.get(f"{GATEWAY_URL}/health") | ... | ... |