From af31f0a885999c0cdf677028c75fdec886708394 Mon Sep 17 00:00:00 2001 From: yibo Date: Sat, 5 Sep 2026 16:37:21 +0800 Subject: [PATCH] h3-console: 加批量取消/删除接口 --- h3-console/server.py | 91 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++----- 1 file changed, 86 insertions(+), 5 deletions(-) diff --git a/h3-console/server.py b/h3-console/server.py index 5c2b30a..479180c 100644 --- a/h3-console/server.py +++ b/h3-console/server.py @@ -396,8 +396,10 @@ async def create_video(req: dict): return _row_view(row) -@api.post("/v1/videos/{jid}/cancel") -async def cancel_video(jid: str): +async def _cancel_job(jid: str) -> dict: + """真正干活的取消/停止逻辑。单条接口和下面的批量接口共用——批量场景需要一条失败不 + 连累其它条,所以这里只管抛 HTTPException,由调用方决定是直接透传(单条)还是 + catch 下来记进 failed 列表(批量)。""" with db() as con: row = con.execute("SELECT * FROM jobs WHERE id=?", (jid,)).fetchone() if row is None: @@ -434,6 +436,11 @@ async def cancel_video(jid: str): return _row_view(row) +@api.post("/v1/videos/{jid}/cancel") +async def cancel_video(jid: str): + return await _cancel_job(jid) + + @api.post("/v1/videos/{jid}/prioritize") async def prioritize_video(jid: str): with db() as con: @@ -588,9 +595,8 @@ async def list_videos(limit: int = 20, offset: int = 0, trash: bool = False, q: return {"jobs": [_row_view(r) for r in rows], "total": total} -@api.delete("/v1/videos/{jid}") -async def delete_video(jid: str): - """软删除:只挪出主列表,不碰 gateway/api.py 上的任务本身(可能还在跑),不删磁盘文件。""" +def _delete_job(jid: str) -> dict: + """真正干活的软删除逻辑,单条/批量共用,理由同 _cancel_job。""" with db() as con: cur = con.execute("UPDATE jobs SET deleted_at=? WHERE id=? AND deleted_at IS NULL", (time.time(), jid)) if cur.rowcount == 0: @@ -598,6 +604,12 @@ async def delete_video(jid: str): return {"id": jid, "deleted": True} +@api.delete("/v1/videos/{jid}") +async def delete_video(jid: str): + """软删除:只挪出主列表,不碰 gateway/api.py 上的任务本身(可能还在跑),不删磁盘文件。""" + return _delete_job(jid) + + @api.post("/v1/videos/{jid}/restore") async def restore_video(jid: str): with db() as con: @@ -607,6 +619,75 @@ async def restore_video(jid: str): return {"id": jid, "deleted": False} +# 2026-09-05 加:一次过载事故后要清理一批内容重复的排队任务,才发现只有按单条 id 操作的 +# cancel/delete,脚本化清理只能一个个调,慢且容易漏。下面两个批量接口给 ids 列表和"筛选 +# 条件"两种入参二选一——ids 优先且没有数量上限(调用方既然把 id 一个个列出来了,就是明确 +# 知道要动哪些);筛选条件复用跟 GET /v1/videos 一样的 status/kind/q 语义,但设了 +# _BATCH_MAX_BY_FILTER 上限,防止筛选条件写宽了误伤一大片跟预期不符的任务——命中数超过 +# 上限直接 400,让调用方缩小范围或换成显式 ids,而不是静默截断一部分。 +_BATCH_MAX_BY_FILTER = 300 + + +def _resolve_batch_targets(body: dict) -> list[str]: + ids = body.get("ids") + if ids: + return list(dict.fromkeys(ids)) # 去重但保留顺序,防止调用方列表里手滑写重了 + status, kind, q = body.get("status") or "", body.get("kind") or "", body.get("q") or "" + if not (status or kind or q): + raise HTTPException(400, "必须提供 ids 列表,或至少一个筛选条件(status/kind/q),避免误操作全部任务") + where = ["deleted_at IS NULL"] + params: list = [] + if q: + where.append("(id LIKE ? OR req_json LIKE ?)") + params += [f"%{q}%", f"%{q}%"] + if status: + where.append("status = ?") + params.append(status) + if kind: + where.append("partition = ?") + params.append(kind) + with db() as con: + total = con.execute(f"SELECT COUNT(*) FROM jobs WHERE {' AND '.join(where)}", params).fetchone()[0] + if total > _BATCH_MAX_BY_FILTER: + raise HTTPException( + 400, + f"筛选条件命中 {total} 条,超过单次批量操作上限 {_BATCH_MAX_BY_FILTER}," + "请缩小筛选范围(加 status/kind/q)或改成显式 ids 列表", + ) + rows = con.execute(f"SELECT id FROM jobs WHERE {' AND '.join(where)}", params).fetchall() + return [r["id"] for r in rows] + + +@api.post("/v1/videos/batch-cancel") +async def batch_cancel_videos(body: dict): + """批量取消/停止。body 传 {"ids": [...]} 或 {"status": "queued", "kind": "...", "q": "..."} + 筛选条件(至少给一个)。逐条按现有单条 cancel 语义处理,某一条已经是终态之类的失败不影响 + 其它条,成功/失败分别列出来。""" + ids = _resolve_batch_targets(body) + succeeded, failed = [], [] + for jid in ids: + try: + succeeded.append((await _cancel_job(jid))["id"]) + except HTTPException as e: + failed.append({"id": jid, "error": e.detail}) + return {"requested": len(ids), "succeeded": succeeded, "failed": failed} + + +@api.post("/v1/videos/batch-delete") +async def batch_delete_videos(body: dict): + """批量软删除(移入回收站),语义同单条 DELETE /v1/videos/{jid}——只是把任务从列表里挪走, + 不取消底层计算;真要连底层任务一起停掉,配合上面的 batch-cancel 一起调(谁先谁后不影响 + 结果,两个操作互不冲突)。""" + ids = _resolve_batch_targets(body) + succeeded, failed = [], [] + for jid in ids: + try: + succeeded.append(_delete_job(jid)["id"]) + except HTTPException as e: + failed.append({"id": jid, "error": e.detail}) + return {"requested": len(ids), "succeeded": succeeded, "failed": failed} + + @api.get("/v1/backends") async def backends(): resp = await client.get(f"{GATEWAY_URL}/health") -- libgit2 0.21.4