From 74f1566985b8bba03539bace7b4aed3abd50b929 Mon Sep 17 00:00:00 2001 From: root Date: Tue, 29 Sep 2026 20:42:17 +0800 Subject: [PATCH 1/4] feat(qbittorrent): login compat for v4/v5.0-5.2 and three auth modes - Login success = HTTP 200 "Ok." or 204 (5.2+); session cookie taken from Set-Cookie by name (SID / QBT_SID_ / custom) and sent back as-is. - Failure classes: "Fails." / 401 -> auth; 403 IP ban / 429 -> rate; wrong API Key (403) -> auth with hint; 5xx -> protocol. - Downloader instance auth kind: password / api_key (qB 5.2+ Bearer, no /auth/login) / none (whitelisted subnet); stored in settings.authKind, credential write-only in secret_ref, cleared when switching to none. One qbAuthConfig() shared by sync, agent capability and connection test. - Settings form: auth kind select, password / API Key fields (no echo). - testkit: mock qB 5.2 login behaviour, IP ban, bypassAuth; makeTorrent({ realPieces }) builds real piece hashes for real qB checks. - Real-container matrix (QB_REAL, skipped by default): 5.1.2 and 5.2.4, login / add / query / tags / pause-resume / delete keeping data. Co-Authored-By: Claude Opus 5.5 --- .../src/modules/acquisition-organize/ports.ts | 5 +- .../modules/agent/capabilities-download.ts | 5 +- .../modules/auth-settings/instances-routes.ts | 56 ++++- .../test/auth-settings/admin-api.test.ts | 60 +++++ .../web/app/components/settings/instances.tsx | 52 +++- docs/runbook/deploy.md | 30 ++- packages/adapters/src/qbittorrent/client.ts | 119 +++++++-- .../src/qbittorrent/qbittorrent.real.test.ts | 227 ++++++++++++++++++ .../src/qbittorrent/qbittorrent.test.ts | 100 ++++++++ packages/contracts/src/instances.ts | 13 +- packages/testkit/src/qbittorrent/bencode.ts | 68 +++++- packages/testkit/src/qbittorrent/index.ts | 1 + packages/testkit/src/qbittorrent/mock-qb.ts | 33 ++- packages/testkit/test/mock-qb.test.ts | 46 +++- 14 files changed, 756 insertions(+), 59 deletions(-) create mode 100644 packages/adapters/src/qbittorrent/qbittorrent.real.test.ts diff --git a/apps/server/src/modules/acquisition-organize/ports.ts b/apps/server/src/modules/acquisition-organize/ports.ts index eef6f99b..3b94d883 100644 --- a/apps/server/src/modules/acquisition-organize/ports.ts +++ b/apps/server/src/modules/acquisition-organize/ports.ts @@ -9,6 +9,7 @@ import { getSiteAdapter, getSiteDefinition, type ProbeResult, + qbAuthConfig, type SecretResolver, SiteError, type TorrentSource, @@ -97,12 +98,10 @@ export function createPorts( if (hit && hit.stamp === stamp) return hit.client; const settings = (inst.settings ?? {}) as Record; const secret = await resolveSecret(inst.secret_ref); - const useApiKey = settings.authKind === "api_key"; const client = createQbClient({ instanceId: inst.id, baseUrl: inst.base_url, - ...(inst.username ? { username: inst.username } : {}), - ...(secret ? (useApiKey ? { apiKey: secret } : { password: secret }) : {}), + ...qbAuthConfig({ authKind: settings.authKind, username: inst.username, secret: secret ?? null }), }); clients.set(inst.id, { stamp, client }); return client; diff --git a/apps/server/src/modules/agent/capabilities-download.ts b/apps/server/src/modules/agent/capabilities-download.ts index b63dfcde..ed278815 100644 --- a/apps/server/src/modules/agent/capabilities-download.ts +++ b/apps/server/src/modules/agent/capabilities-download.ts @@ -1,4 +1,4 @@ -import { createQbClient, type DownloaderClient } from "@umt/adapters"; +import { createQbClient, type DownloaderClient, qbAuthConfig } from "@umt/adapters"; import { CapabilityError, type CapabilityRegistry } from "@umt/agent"; import { capabilitySpec, type DownloadStatus } from "@umt/contracts"; import type { DownloaderInstanceRow } from "@umt/db"; @@ -20,8 +20,7 @@ async function defaultClient(ctx: ServerContext, inst: DownloaderInstanceRow): P return createQbClient({ instanceId: inst.id, baseUrl: inst.base_url, - ...(inst.username ? { username: inst.username } : {}), - ...(secret ? (settings.authKind === "api_key" ? { apiKey: secret } : { password: secret }) : {}), + ...qbAuthConfig({ authKind: settings.authKind, username: inst.username, secret: secret ?? null }), }); } diff --git a/apps/server/src/modules/auth-settings/instances-routes.ts b/apps/server/src/modules/auth-settings/instances-routes.ts index 9b6619e3..9edd7a8e 100644 --- a/apps/server/src/modules/auth-settings/instances-routes.ts +++ b/apps/server/src/modules/auth-settings/instances-routes.ts @@ -1,8 +1,9 @@ import { stat } from "node:fs/promises"; -import { checkRoot, DEFAULT_ROOT_SENTINEL, EmbyClient, QbClient, writeRootSentinel } from "@umt/adapters"; +import { checkRoot, DEFAULT_ROOT_SENTINEL, EmbyClient, QbClient, qbAuthConfig, writeRootSentinel } from "@umt/adapters"; import { Api, type AuditAction, + type DownloaderAuthKind, DownloaderInstance, DownloaderInstanceCreate, DownloaderInstanceUpdate, @@ -238,17 +239,36 @@ export function registerInstanceRoutes(app: ModuleApp, ctx: ServerContext) { // ── downloader_instance ── list("downloader_instance", "downloader-instances", DL.list); get("downloader_instance", "downloader-instances", DL.get); + /** 认证方式并入 settings.authKind;账密需要用户名,API Key 需要凭证(新填或已存) */ + const dlAuth = ( + authKind: DownloaderAuthKind | undefined, + next: { username?: string | null; credential?: string; settings?: Record }, + cur?: Record, + ): Record | undefined => { + const curSettings = (cur?.settings ?? {}) as Record; + if (authKind === undefined) return next.settings ? { ...curSettings, ...next.settings } : undefined; + const username = next.username !== undefined ? next.username : ((cur?.username as string | null) ?? null); + if (authKind === "password" && !username) throw new ApiException("validation_failed", "账密认证需要填写用户名"); + if (authKind === "api_key" && !next.credential && !(cur?.secret_ref && curSettings.authKind === "api_key")) + throw new ApiException("validation_failed", "API Key 认证需要填写 API Key"); + return { ...curSettings, ...(next.settings ?? {}), authKind }; + }; endpoint(app, DL.create, async ({ body, actor }) => { - const { credential, ...rest } = body; - const ref = credential ? await ctx.secrets.create({ purpose: "downloader_password", plaintext: credential }) : null; + const { credential, authKind, ...rest } = body; + const settings = dlAuth(authKind, { ...rest, ...(credential ? { credential } : {}) }); + const keep = authKind !== "none" && credential; + const ref = keep ? await ctx.secrets.create({ purpose: "downloader_password", plaintext: credential }) : null; try { - const row = await insert("downloader_instance", { ...toDb(rest, JSON_COLS), secret_ref: ref }); + const row = await insert("downloader_instance", { + ...toDb({ ...rest, ...(settings ? { settings } : {}) }, JSON_COLS), + secret_ref: ref, + }); await audit( ctx, actor, "downloader_instance.create", { type: "downloader_instance", id: row.id }, - { name: row.name }, + { name: row.name, authKind: authKind ?? null }, ); return toApi(row); } catch (e) { @@ -257,13 +277,22 @@ export function registerInstanceRoutes(app: ModuleApp, ctx: ServerContext) { } }); endpoint(app, DL.update, async ({ params, body, actor }) => { - const { credential, ...rest } = body; - const values = toDb(rest, JSON_COLS); - if (credential) { - const cur = await getRow("downloader_instance", params.id); + const { credential, authKind, ...rest } = body; + const cur = await getRow("downloader_instance", params.id); + const settings = dlAuth(authKind, { ...rest, ...(credential ? { credential } : {}) }, cur); + const values = toDb({ ...rest, ...(settings ? { settings } : {}) }, JSON_COLS); + let dropRef: string | null = null; + if (authKind === "none") { + // 无认证:不再保留密码 / Key + if (cur.secret_ref) { + values.secret_ref = null; + dropRef = cur.secret_ref; + } + } else if (credential) { values.secret_ref = await putCredential(cur.secret_ref, "downloader_password", credential); } const row = await update("downloader_instance", params.id, values); + if (dropRef) await ctx.secrets.remove(dropRef).catch(() => {}); await audit( ctx, actor, @@ -285,12 +314,15 @@ export function registerInstanceRoutes(app: ModuleApp, ctx: ServerContext) { async ({ params, actor }) => { const row = await getRow("downloader_instance", params.id); const secret = row.secret_ref ? await ctx.secrets.resolve(row.secret_ref) : undefined; - const username = (row.username as string | null) ?? undefined; - // 有用户名走 WebUI 登录(密码);没有用户名但有凭证视为 qB 5.2+ WebUI API Key + // 认证方式三选一(settings.authKind):账密 / API Key(qB 5.2+,Bearer) / 无认证;旧数据按用户名有无推断 const client = new QbClient({ instanceId: row.id, baseUrl: String(row.base_url), - ...(username ? { username, ...(secret ? { password: secret } : {}) } : secret ? { apiKey: secret } : {}), + ...qbAuthConfig({ + authKind: ((row.settings ?? {}) as Record).authKind, + username: (row.username as string | null) ?? null, + secret: secret ?? null, + }), timeoutMs: 10_000, retries: 0, }); diff --git a/apps/server/test/auth-settings/admin-api.test.ts b/apps/server/test/auth-settings/admin-api.test.ts index 5895cc5f..984dac78 100644 --- a/apps/server/test/auth-settings/admin-api.test.ts +++ b/apps/server/test/auth-settings/admin-api.test.ts @@ -188,6 +188,66 @@ describe.skipIf(!hasPg)("W2-A 管理 API(真实 PG)", () => { noLeak("qb-secret-pass"); }); + it("downloader_instance 认证方式三选一:API Key(不登录)/ 无认证(内网白名单)/ 账密;切换校验与凭证清理", async () => { + const k52 = await createMockQb({ version: "5.2.4", apiKey: "qbt_live_key" }); + const open = await createMockQb({ version: "5.2.4", bypassAuth: true }); + try { + expect( + (await admin("POST", "/downloader-instances", { name: "k0", baseUrl: k52.url, authKind: "api_key" })).status, + ).toBe(422); + expect( + (await admin("POST", "/downloader-instances", { name: "p0", baseUrl: k52.url, authKind: "password" })).status, + ).toBe(422); + const k = ( + await json( + await admin("POST", "/downloader-instances", { + name: "k52", + baseUrl: k52.url, + authKind: "api_key", + credential: "qbt_live_key", + }), + ) + ).data; + expect(k.settings.authKind).toBe("api_key"); + expect(JSON.stringify(k)).not.toContain("qbt_live_key"); + expect((await json(await admin("POST", `/downloader-instances/${k.id}/test`))).data.ok).toBe(true); + expect(k52.callsTo("auth/login")).toHaveLength(0); + await admin("PATCH", `/downloader-instances/${k.id}`, { credential: "qbt_wrong" }); + const bad = (await json(await admin("POST", `/downloader-instances/${k.id}/test`))).data; + expect(bad.ok).toBe(false); + expect(bad.message).toContain("API Key"); + + const n = ( + await json( + await admin("POST", "/downloader-instances", { name: "open", baseUrl: open.url, authKind: "none" }), + ) + ).data; + expect(n.secretRef).toBeNull(); + expect((await json(await admin("POST", `/downloader-instances/${n.id}/test`))).data.ok).toBe(true); + expect(open.callsTo("auth/login")).toHaveLength(0); + + // 切到无认证:凭证删除;无认证连需要认证的实例 → ok=false + const sw = (await json(await admin("PATCH", `/downloader-instances/${k.id}`, { authKind: "none" }))).data; + expect(sw.secretRef).toBeNull(); + await expect(h.ctx.secrets.resolve(k.secretRef)).rejects.toThrow(/不存在/); + expect((await json(await admin("POST", `/downloader-instances/${k.id}/test`))).data.ok).toBe(false); + // 再切回 API Key 必须带 Key + expect((await admin("PATCH", `/downloader-instances/${k.id}`, { authKind: "api_key" })).status).toBe(422); + const back = ( + await json( + await admin("PATCH", `/downloader-instances/${k.id}`, { authKind: "api_key", credential: "qbt_live_key" }), + ) + ).data; + expect(back.secretRef).toBeTruthy(); + expect((await json(await admin("POST", `/downloader-instances/${k.id}/test`))).data.ok).toBe(true); + for (const id of [k.id, n.id]) await admin("DELETE", `/downloader-instances/${id}`); + noLeak("qbt_live_key"); + } finally { + await k52.close(); + await open.close(); + } + }); + it("media_server_instance:无 key 时 test 报未配置;有 key 时 ping 成功", async () => { const a = (await json(await admin("POST", "/media-server-instances", { name: "emby-nokey", baseUrl: emby.url }))) .data; diff --git a/apps/web/app/components/settings/instances.tsx b/apps/web/app/components/settings/instances.tsx index d1ec79c8..aef75f5c 100644 --- a/apps/web/app/components/settings/instances.tsx +++ b/apps/web/app/components/settings/instances.tsx @@ -67,6 +67,19 @@ export function TestButton({ const credentialHint = { editHint: "留空不修改" }; +const AUTH_KINDS = [ + { value: "password", label: "用户名和密码" }, + { value: "api_key", label: "API Key" }, + { value: "none", label: "无认证（内网白名单）" }, +]; + +/** 已存认证方式;旧数据没有 settings.authKind 时按用户名 / 凭证推断(与服务端一致) */ +function authKindOf(r: DownloaderInstance): string { + const k = (r.settings as { authKind?: unknown }).authKind; + if (k === "password" || k === "api_key" || k === "none") return k; + return r.username ? "password" : r.secretRef ? "api_key" : "none"; +} + export function DownloadersSection() { return ( @@ -121,8 +134,29 @@ function DownloadersTable() { required: true, placeholder: "http://192.168.1.10:8080", }, - { name: "username", label: "用户名", kind: "text", hint: "留空时凭证按 WebUI API Key 使用" }, - { name: "credential", label: "密码 / API Key", kind: "password", ...credentialHint }, + { + name: "authKind", + label: "认证方式", + kind: "select", + options: AUTH_KINDS, + hint: "API Key 需要 qBittorrent 5.2 及以上，在 WebUI 设置里生成", + }, + { name: "username", label: "用户名", kind: "text", visible: (v) => v.authKind === "password" }, + { + name: "password", + label: "密码", + kind: "password", + visible: (v) => v.authKind === "password", + ...credentialHint, + }, + { + name: "apiKey", + label: "API Key", + kind: "password", + placeholder: "qbt_…", + visible: (v) => v.authKind === "api_key", + ...credentialHint, + }, { name: "managedTag", label: "管理标签", kind: "text", placeholder: "umt", hint: "只接管带此标签的任务" }, { name: "cacheVolumeKey", label: "缓存卷标识", kind: "text", placeholder: "cache-1" }, { name: "enabled", label: "启用", kind: "switch" }, @@ -130,8 +164,10 @@ function DownloadersTable() { initial: { name: "", baseUrl: "", + authKind: "password", username: "", - credential: "", + password: "", + apiKey: "", managedTag: "umt", cacheVolumeKey: "", enabled: true, @@ -139,8 +175,10 @@ function DownloadersTable() { fromRow: (r) => ({ name: r.name, baseUrl: r.baseUrl, + authKind: authKindOf(r), username: r.username ?? "", - credential: "", + password: "", + apiKey: "", managedTag: r.managedTag, cacheVolumeKey: r.cacheVolumeKey ?? "", enabled: r.enabled, @@ -149,8 +187,10 @@ function DownloadersTable() { name: str(v.name), kind: mode === "create" ? "qbittorrent" : undefined, baseUrl: str(v.baseUrl), - username: nullStr(v.username), - credential: optStr(v.credential), + authKind: str(v.authKind), + username: v.authKind === "password" ? nullStr(v.username) : null, + credential: + v.authKind === "password" ? optStr(v.password) : v.authKind === "api_key" ? optStr(v.apiKey) : undefined, managedTag: optStr(v.managedTag), cacheVolumeKey: nullStr(v.cacheVolumeKey), enabled: v.enabled === true, diff --git a/docs/runbook/deploy.md b/docs/runbook/deploy.md index d1e7b395..03b2c90d 100644 --- a/docs/runbook/deploy.md +++ b/docs/runbook/deploy.md @@ -246,7 +246,7 @@ services: # 配了 HTTP(S)_PROXY 时必须把 qbittorrent 加进 NO_PROXY:Deno 按主机名匹配,不会因解析到内网 IP 而直连 NO_PROXY: localhost,127.0.0.1,postgres,camofox,proxy-relay,qbittorrent,10.0.0.0/8,172.16.0.0/12,192.168.0.0/16 qbittorrent: - image: lscr.io/linuxserver/qbittorrent:5.1.2 # 5.2.x 登录返回 204 + QBT_SID_ cookie,当前适配器不兼容 + image: lscr.io/linuxserver/qbittorrent:5.2.4 # v4 / 5.0–5.2 都兼容(QB-1 实测 5.1.2、5.2.4) restart: unless-stopped environment: { PUID: "10001", PGID: "10001", TZ: Asia/Shanghai, WEBUI_PORT: "8080" } volumes: @@ -266,9 +266,35 @@ APP_VERSION=<当前版本> docker compose --env-file .env.deploy --profile tls u ``` - 本机经 8081 访问时 qB 的 Host 头校验要求端口与 `WEBUI_PORT` 一致:curl 加 `-H 'Host: localhost:8080'`,否则 401。 + 要从宿主直接用映射端口访问,就让两边端口相同(`WEBUI_PORT=18082` + `127.0.0.1:18082:18082`)。 - 用临时密码登录后 `POST /api/v2/app/setPreferences`:`web_ui_password`(随机强密码,存本机凭据文件)、`save_path=/downloads`、 `dht/pex/lsd/upnp=false`、`up_limit`/`dl_limit` 限速、`auto_tmm_enabled=false`、`torrent_content_layout=Original`。 - 控制台登记:下载器实例 `http://qbittorrent:8080`、管理标签 `umt`、`settings.savePath=/downloads`;路径映射 downloader `/downloads` ↔ `/downloads`; 自动整理要求库根为 `managed`(`register_only` 时 `no_library_root`)。 -- 模拟完成:按发布名把样片放进下载目录,用真实 piece 哈希做种子(testkit `makeTorrent` 的 pieces 是伪值,qB 校验不到 100%), +- 模拟完成:按发布名把样片放进下载目录,用真实 piece 哈希做种子(testkit `makeTorrent({ realPieces: true })` 返回 `contents`, + 写进保存目录;缺省 pieces 是伪值,qB 校验不到 100%), 带标签 `umt` 加入 qB,校验后即 100% 做种,`downloads.sync` 接管。 + +## 下载器认证与登录兼容(QB-1) + +控制台 设置 → qBittorrent 实例 的「认证方式」三选一,存 `downloader_instance.settings.authKind`,凭证只写、存 `secret_ref`、不回显: + +| 认证方式 | 凭证 | 请求 | 适用 | +| --- | --- | --- | --- | +| 用户名和密码(`password`) | 密码 | `POST /api/v2/auth/login` 取会话 cookie,之后带 cookie;会话失效(401/403)自动重登一次 | 所有版本 | +| API Key(`api_key`) | WebUI 生成的 `qbt_…` | 每个请求带 `Authorization: Bearer `,不调 `/auth/login` | qB 5.2+ | +| 无认证(`none`) | 无(切到它时删除已存凭证) | 不登录、不带凭证 | qB 开了"对本地主机 / 白名单子网跳过认证"的内网 | + +旧实例没有 `authKind`:有用户名按账密,无用户名但有凭证按 API Key,都没有按无认证(与改前行为一致)。 +API Key 在 qB 5.2+ WebUI 选项 → WebUI → API Key 处生成 / 轮换(等价 `POST /api/v2/app/rotateAPIKey`,返回 `{"apiKey":"qbt_…"}`; +轮换后旧 Key 立即失效,记得回控制台更新)。 + +登录判定(对照 qbittorrent-api `auth_log_in`): + +- 成功:HTTP 200 且体为 `Ok.`(≤5.1),或 204 空体(5.2+)。会话 cookie 名不固定(≤5.1 `SID`,5.2+ `QBT_SID_<容器内 WebUI 端口>`, + 也可能被自定义),一律取 `Set-Cookie` 里的会话 cookie 原名原值回传。 +- 失败分类:200 `Fails.` / 401 → 凭据错误(`auth`);403 "IP 已被封禁" / 429 → 限流(`rate`,默认连续 5 次失败封 1 小时, + 等封禁过期或重启 qB);API Key 错误 qB 返回 403 → `auth`;无认证被要求认证 → `auth`;5xx → `protocol`。 +- 实测矩阵(`packages/adapters/src/qbittorrent/qbittorrent.real.test.ts`,设 `QB_REAL` 才跑):5.1.2 账密 / 无认证 / 失败分类, + 5.2.4 账密 / API Key / 无认证 / 失败分类,各跑 登录 → 添加(testkit `makeTorrent({ realPieces: true })` 真实 piece 哈希, + 内容写进保存目录,qB 校验即完成)→ 查询 → 标签 → 暂停恢复 → 删除(不删数据)。 diff --git a/packages/adapters/src/qbittorrent/client.ts b/packages/adapters/src/qbittorrent/client.ts index 04fa6b5a..9b07db91 100644 --- a/packages/adapters/src/qbittorrent/client.ts +++ b/packages/adapters/src/qbittorrent/client.ts @@ -17,8 +17,13 @@ export interface QbInstanceConfig { baseUrl: string; username?: string; password?: string; - /** qB 5.2+ WebUI API Key;有则走 Bearer,不登录 */ + /** qB 5.2+ WebUI API Key(WebUI 生成,形如 qbt_…);有则走 Authorization: Bearer,不调 /auth/login */ apiKey?: string; + /** + * 认证方式(对照 MP / qbittorrent-api):password 走 /auth/login 会话 cookie;api_key 只发 Bearer 头; + * none 不认证(qB 设了"对白名单子网跳过认证"的内网)。缺省按是否给了 apiKey 推断。 + */ + authMode?: QbAuthMode; /** 单次请求超时,默认 15000 */ timeoutMs?: number; /** 幂等读请求的重试次数(不含首次),默认 2;写请求不自动重试 */ @@ -36,6 +41,30 @@ export interface QbInstanceConfig { fetch?: typeof fetch; } +export type QbAuthMode = "password" | "api_key" | "none"; + +/** + * 实例配置 → 认证参数(服务端各处构造客户端共用)。 + * authKind 显式给出时按它;缺省兼容旧数据:有用户名 → 账密,无用户名但有凭证 → API Key,否则无认证。 + */ +export function qbAuthConfig(i: { + authKind?: unknown; + username?: string | null; + secret?: string | null; +}): Pick { + const kind: QbAuthMode = + i.authKind === "password" || i.authKind === "api_key" || i.authKind === "none" + ? i.authKind + : i.username + ? "password" + : i.secret + ? "api_key" + : "none"; + if (kind === "api_key") return { authMode: kind, ...(i.secret ? { apiKey: i.secret } : {}) }; + if (kind === "none") return { authMode: kind }; + return { authMode: kind, username: i.username ?? "", ...(i.secret ? { password: i.secret } : {}) }; +} + export interface QbTorrent { instanceId: string; hash: string; @@ -201,6 +230,23 @@ function str(v: unknown): string | null { return typeof v === "string" && v !== "" ? v : null; } +/** + * 从登录响应的 Set-Cookie 中取会话 cookie,返回可直接放进 Cookie 头的 "name=value"。 + * 优先 SID / QBT_SID_ / 名字含 SID 的;否则取第一个有值的 cookie。 + */ +export function pickSessionCookie(setCookies: readonly string[]): string | null { + const pairs = setCookies + .map((c) => /^\s*([^=;\s]+)=([^;]*)/.exec(c)) + .filter((m): m is RegExpExecArray => m !== null && m[2] !== "") + .map((m) => ({ name: m[1]!, value: m[2]! })); + const hit = + pairs.find((p) => p.name === "SID") ?? + pairs.find((p) => /^QBT_SID(_\d+)?$/i.test(p.name)) ?? + pairs.find((p) => /sid/i.test(p.name)) ?? + pairs[0]; + return hit ? `${hit.name}=${hit.value}` : null; +} + export function normalizeTorrent(instanceId: string, r: Record): QbTorrent { const rawState = typeof r.state === "string" ? r.state : ""; const progress = num(r.progress) ?? 0; @@ -253,8 +299,11 @@ export class QbClient implements DownloaderClient { /** qB 5.x 把 pause/resume 改名为 stop/start;探测到后缓存 */ private legacyPauseApi: boolean | null = null; + private readonly authMode: QbAuthMode; + constructor(private readonly cfg: QbInstanceConfig) { this.instanceId = cfg.instanceId; + this.authMode = cfg.authMode ?? (cfg.apiKey ? "api_key" : "password"); this.base = cfg.baseUrl.replace(/\/+$/, ""); this.timeoutMs = cfg.timeoutMs ?? 15_000; this.retries = cfg.retries ?? 2; @@ -290,7 +339,7 @@ export class QbClient implements DownloaderClient { /** 登录;并发调用合并为一次。 */ async login(): Promise { - if (this.cfg.apiKey) return; + if (this.authMode !== "password") return; if (!this.loginInFlight) { this.loginInFlight = this.doLogin().finally(() => { this.loginInFlight = null; @@ -299,6 +348,13 @@ export class QbClient implements DownloaderClient { return this.loginInFlight; } + /** + * 登录判定(对照 qbittorrent-api auth_log_in,覆盖 v4 / v5.0–5.2): + * - 成功:HTTP 200 且体为 "Ok."(≤5.1),或 204 空体(5.2+); + * - 会话 cookie 名不固定:≤5.1 为 SID,5.2+ 为 QBT_SID_(容器内端口,不一定等于映射端口), + * 也可能被自定义 —— 一律取 Set-Cookie 里的会话 cookie 原名原值回传; + * - 失败:200 "Fails." / 401 → 凭据错误(auth);403 被封禁 / 429 → 限流(rate);5xx → protocol。 + */ private async doLogin(): Promise { const endpoint = "auth/login"; const res = await this.rawFetch(endpoint, { @@ -306,18 +362,30 @@ export class QbClient implements DownloaderClient { headers: { "content-type": "application/x-www-form-urlencoded", referer: this.base }, body: new URLSearchParams({ username: this.cfg.username ?? "", password: this.cfg.password ?? "" }), }); - const text = await res.text(); - if (res.status === 403) throw this.err("auth", "login forbidden (IP banned after failed attempts?)", endpoint, 403); - if (res.status === 429) throw this.err("rate", "login rate limited", endpoint, 429); - if (!res.ok) - throw this.err(res.status >= 500 ? "protocol" : "auth", `login HTTP ${res.status}`, endpoint, res.status); - if (!text.trim().startsWith("Ok")) throw this.err("auth", "login rejected (bad credentials)", endpoint, res.status); - const cookie = res.headers - .getSetCookie() - .map((c) => /^SID=([^;]*)/.exec(c)?.[1]) - .find((v) => v !== undefined); - // 关闭鉴权或白名单 IP 时 qB 可能不下发 cookie,仍视为成功 - this.sid = cookie ?? ""; + const text = (await res.text()).trim(); + const s = res.status; + if (s === 204 || (s === 200 && /^ok\.?$/i.test(text))) { + // 关闭鉴权或白名单 IP 时 qB 可能不下发 cookie,仍视为成功 + this.sid = pickSessionCookie(res.headers.getSetCookie()) ?? ""; + return; + } + if (s === 200 && /^fails\.?$/i.test(text)) + throw this.err("auth", "login rejected: bad username or password", endpoint, s); + if (s === 429) throw this.err("rate", "login rate limited", endpoint, s); + if (s === 403) { + if (/ban/i.test(text)) + throw this.err("rate", "login blocked: IP banned after too many failed attempts", endpoint, s); + throw this.err("auth", `login forbidden (HTTP 403${text ? `: ${text.slice(0, 80)}` : ""})`, endpoint, s); + } + if (s === 401) + throw this.err( + "auth", + "login rejected (HTTP 401): bad username or password; for a port-mapped container also check that WEBUI_PORT equals the published port (qB rejects mismatched Host)", + endpoint, + s, + ); + if (s >= 500) throw this.err("protocol", `login HTTP ${s}`, endpoint, s); + throw this.err("protocol", `unexpected login response: HTTP ${s} ${text.slice(0, 80)}`, endpoint, s); } private async send(endpoint: string, opts: ReqOpts): Promise { @@ -338,10 +406,13 @@ export class QbClient implements DownloaderClient { } private async sendOnce(endpoint: string, opts: ReqOpts, relogged: boolean): Promise { - if (!this.cfg.apiKey && this.sid === null) await this.login(); + const session = this.authMode === "password"; + if (session && this.sid === null) await this.login(); const headers: Record = { referer: this.base }; - if (this.cfg.apiKey) headers.authorization = `Bearer ${this.cfg.apiKey}`; - else if (this.sid) headers.cookie = `SID=${this.sid}`; + if (this.authMode === "api_key") { + if (!this.cfg.apiKey) throw this.err("auth", "API Key auth selected but no API Key configured", endpoint); + headers.authorization = `Bearer ${this.cfg.apiKey}`; + } else if (session && this.sid) headers.cookie = this.sid; let body: RequestInit["body"] | undefined; if (opts.body?.kind === "form") { headers["content-type"] = "application/x-www-form-urlencoded"; @@ -355,13 +426,21 @@ export class QbClient implements DownloaderClient { const res = await this.rawFetch(path, { method: body ? "POST" : "GET", headers, ...(body ? { body } : {}) }); const s = res.status; if (s === 401 || s === 403) { - await res.body?.cancel(); - if (!this.cfg.apiKey && !relogged) { + const t = (await res.text().catch(() => "")).trim(); + if (s === 403 && /ban/i.test(t)) + throw this.err("rate", "IP banned after too many failed authentication attempts", endpoint, s); + if (session && !relogged) { this.sid = null; await this.login(); return this.sendOnce(endpoint, opts, true); } - throw this.err("auth", `unauthorized (HTTP ${s})`, endpoint, s); + const why = + this.authMode === "api_key" + ? "API Key rejected (check the key; API Key needs qBittorrent 5.2+)" + : this.authMode === "none" + ? "WebUI requires authentication (use username/password or API Key, or whitelist this subnet in qB)" + : "session rejected after re-login"; + throw this.err("auth", `unauthorized (HTTP ${s}): ${why}`, endpoint, s); } if (s === 404) { const t = await res.text(); diff --git a/packages/adapters/src/qbittorrent/qbittorrent.real.test.ts b/packages/adapters/src/qbittorrent/qbittorrent.real.test.ts new file mode 100644 index 00000000..bc12af54 --- /dev/null +++ b/packages/adapters/src/qbittorrent/qbittorrent.real.test.ts @@ -0,0 +1,227 @@ +/** + * 真 qBittorrent 兼容性测试(默认跳过)。设置 QB_REAL 后对每个实例、每种认证方式跑一遍 + * 登录 / 添加 / 查询 / 标签 / 暂停恢复 / 删除(不删数据),以及失败分类(错密码 → auth,IP 封禁 → rate)。 + * + * QB_REAL='[{"label":"5.2.4","url":"http://127.0.0.1:18082","hostDir":"/tmp/qb52/downloads","qbDir":"/downloads", + * "username":"admin","password":"adminadmin","apiKey":"qbt_…"}]' + * hostDir 是 qB 保存目录在本机的路径(测试先把真实内容写进去,qB 校验通过即完成)。 + * 容器做了端口映射时 WEBUI_PORT 必须等于映射端口,否则 qB 按 Host 头拒绝(一律 401)。 + */ +import { existsSync } from "node:fs"; +import { mkdir, rm, writeFile } from "node:fs/promises"; +import { dirname, join } from "node:path"; +import { makeTorrent } from "@umt/testkit"; +import { describe, expect, it } from "vitest"; +import { createQbClient, type DownloaderClient, DownloaderError, qbAuthConfig } from "./index.ts"; + +interface Target { + label: string; + url: string; + hostDir: string; + qbDir: string; + username: string; + password: string; + apiKey?: string | null; +} + +const targets: Target[] = (() => { + try { + return JSON.parse(process.env.QB_REAL ?? "[]") as Target[]; + } catch { + return []; + } +})(); + +const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)); +const fast = { retries: 1, retryBaseMs: 200, lookup: { attempts: 10, intervalMs: 500 }, timeoutMs: 10_000 }; + +async function waitFor(fn: () => Promise, ms = 30_000): Promise { + const end = Date.now() + ms; + for (;;) { + const v = await fn(); + if (v) return v; + if (Date.now() > end) throw new Error("waitFor timeout"); + await sleep(300); + } +} + +async function roundTrip(qb: DownloaderClient, t: Target, tag: string) { + const v = await qb.version(); + expect(v.app).toMatch(new RegExp(`^v?${t.label.replace(/\./g, "\\.")}`)); + const dir = `umt-real-${tag}-${Date.now().toString(36)}`; + const made = makeTorrent({ + name: `Real.Compat.${tag}.2024.1080p`, + pieceLength: 16384, + files: [ + { path: "Real.Compat.2024.1080p.mkv", length: 150_000 }, + { path: "Real.Compat.2024.1080p.chs.srt", length: 4_000 }, + ], + realPieces: true, + }); + for (const c of made.contents!) { + const p = join(t.hostDir, dir, c.path); + await mkdir(dirname(p), { recursive: true }); + await writeFile(p, c.bytes); + } + try { + const out = await qb.submit({ + source: { kind: "file", bytes: made.bytes, filename: "real.torrent" }, + savePath: `${t.qbDir}/${dir}`, + tags: ["umt-real"], + }); + expect(out.kind).toBe("added"); + // qB 对已有数据做校验:内容正确 → 完成(必要时手动 recheck) + let rechecked = false; + const done = await waitFor(async () => { + const x = await qb.findByHash(made.hash); + if (x && !x.complete && !rechecked && /^(stopped|paused)/.test(x.rawState)) { + rechecked = true; + await qb.recheck([made.hash]); + } + return x?.complete ? x : null; + }); + expect(done.progress).toBe(1); + expect((await qb.files(made.hash)).map((f) => f.size).sort((a, b) => a - b)).toEqual([4_000, 150_000]); + + await qb.addTags([made.hash], ["umt-real-extra"]); + expect((await qb.list({ tags: ["umt-real", "umt-real-extra"] })).map((x) => x.hash)).toContain(made.hash); + await qb.removeTags([made.hash], ["umt-real-extra"]); + await qb.deleteTagDefinitions(["umt-real-extra"]); + expect((await qb.findByHash(made.hash))?.tags).toEqual(["umt-real"]); + + await qb.pause([made.hash]); + await waitFor(async () => (await qb.findByHash(made.hash))?.state === "completed_paused"); + await qb.resume([made.hash]); + await waitFor(async () => (await qb.findByHash(made.hash))?.state !== "completed_paused"); + + await qb.delete([made.hash], { deleteFiles: false }); + await waitFor(async () => (await qb.findByHash(made.hash)) === null); + // 不删数据 + for (const c of made.contents!) expect(existsSync(join(t.hostDir, dir, c.path))).toBe(true); + } finally { + await qb.delete([made.hash], { deleteFiles: false }).catch(() => {}); + await rm(join(t.hostDir, dir), { recursive: true, force: true }).catch(() => {}); + } +} + +/** 测试夹具:直接调 WebUI 读写偏好(客户端本身对偏好只读) */ +async function sessionCookie(t: Target): Promise { + const login = await fetch(`${t.url}/api/v2/auth/login`, { + method: "POST", + body: new URLSearchParams({ username: t.username, password: t.password }), + }); + await login.body?.cancel(); + return login.headers.getSetCookie()[0]!.split(";")[0]!; +} +async function getPrefs(t: Target, cookie: string): Promise> { + return (await (await fetch(`${t.url}/api/v2/app/preferences`, { headers: { cookie } })).json()) as Record< + string, + unknown + >; +} +async function setPrefs(t: Target, cookie: string, json: Record): Promise { + const r = await fetch(`${t.url}/api/v2/app/setPreferences`, { + method: "POST", + headers: { cookie, "content-type": "application/x-www-form-urlencoded" }, + body: new URLSearchParams({ json: JSON.stringify(json) }), + }); + await r.body?.cancel(); + expect(r.ok).toBe(true); +} + +describe.skipIf(targets.length === 0)("真 qBittorrent 兼容性", () => { + for (const t of targets) { + it(`${t.label} 账密:登录 / 添加 / 查询 / 标签 / 暂停恢复 / 删除(不删数据)`, async () => { + const qb = createQbClient({ + instanceId: `real-${t.label}`, + baseUrl: t.url, + ...qbAuthConfig({ authKind: "password", username: t.username, secret: t.password }), + ...fast, + }); + await roundTrip(qb, t, "pw"); + }, 90_000); + + it.skipIf(!t.apiKey)( + `${t.label} API Key:不登录,全流程`, + async () => { + const qb = createQbClient({ + instanceId: `real-${t.label}-key`, + baseUrl: t.url, + ...qbAuthConfig({ authKind: "api_key", secret: t.apiKey ?? null }), + ...fast, + }); + await roundTrip(qb, t, "key"); + const bad = createQbClient({ + instanceId: "bad-key", + baseUrl: t.url, + ...qbAuthConfig({ authKind: "api_key", secret: "qbt_definitely_wrong" }), + ...fast, + }); + const e = await bad.version().catch((x) => x); + expect(e).toBeInstanceOf(DownloaderError); + expect(e.kind).toBe("auth"); + }, + 90_000, + ); + + it(`${t.label} 无认证(白名单子网跳过认证):不登录也能查询与添加`, async () => { + const qb = createQbClient({ + instanceId: `real-${t.label}-pw`, + baseUrl: t.url, + ...qbAuthConfig({ authKind: "password", username: t.username, secret: t.password }), + ...fast, + }); + await qb.version(); + const cookie = await sessionCookie(t); + const prefs = await getPrefs(t, cookie); + await setPrefs(t, cookie, { + bypass_auth_subnet_whitelist_enabled: true, + bypass_auth_subnet_whitelist: "0.0.0.0/0", + }); + try { + const none = createQbClient({ + instanceId: "none-ok", + baseUrl: t.url, + ...qbAuthConfig({ authKind: "none" }), + ...fast, + }); + await roundTrip(none, t, "none"); + } finally { + await setPrefs(t, cookie, { + bypass_auth_subnet_whitelist_enabled: prefs.bypass_auth_subnet_whitelist_enabled, + bypass_auth_subnet_whitelist: prefs.bypass_auth_subnet_whitelist, + }); + } + }, 90_000); + + it(`${t.label} 失败分类:错密码 → auth;连续失败被封禁 → rate;无认证 → auth`, async () => { + // 临时把封禁阈值调小(测试结束恢复),避免把本机 IP 封一小时 + const cookie = await sessionCookie(t); + const prefs = await getPrefs(t, cookie); + await setPrefs(t, cookie, { web_ui_max_auth_fail_count: 2, web_ui_ban_duration: 3 }); + try { + const none = createQbClient({ instanceId: "none", baseUrl: t.url, authMode: "none", ...fast }); + expect((await none.version().catch((x) => x)).kind).toBe("auth"); + const bad = createQbClient({ + instanceId: "bad", + baseUrl: t.url, + username: t.username, + password: "wrong-password", + ...fast, + retries: 0, + }); + expect((await bad.version().catch((x) => x)).kind).toBe("auth"); + expect((await bad.version().catch((x) => x)).kind).toBe("auth"); + const banned = await bad.version().catch((x) => x); + expect(banned.kind).toBe("rate"); + expect(banned.detail.status).toBe(403); + } finally { + await sleep(4_000); + await setPrefs(t, cookie, { + web_ui_max_auth_fail_count: prefs.web_ui_max_auth_fail_count, + web_ui_ban_duration: prefs.web_ui_ban_duration, + }); + } + }, 60_000); + } +}); diff --git a/packages/adapters/src/qbittorrent/qbittorrent.test.ts b/packages/adapters/src/qbittorrent/qbittorrent.test.ts index fafa94dc..68ffc011 100644 --- a/packages/adapters/src/qbittorrent/qbittorrent.test.ts +++ b/packages/adapters/src/qbittorrent/qbittorrent.test.ts @@ -8,6 +8,8 @@ import { mapQbState, parseMagnet, parseTorrent, + pickSessionCookie, + qbAuthConfig, } from "./index.ts"; import { fromDownloaderPath, toDownloaderPath } from "./paths.ts"; @@ -94,6 +96,104 @@ describe("QbClient 鉴权与错误分类", () => { expect(e.kind).toBe("auth"); }); + it("5.2:登录 204 空体 + QBT_SID_ 视为成功,会话 cookie 原名回传;过期 401 重登", async () => { + mock = await createMockQb({ version: "5.2.4", webUiPort: 18082 }); + const qb = createQbClient({ + instanceId: "qb-52", + baseUrl: mock.url, + username: "admin", + password: "adminadmin", + ...fast, + }); + expect((await qb.version()).webApi).toBe("2.11.4"); + await qb.list(); + expect(mock.callsTo("auth/login")).toHaveLength(1); + mock.expireSessions(); + await expect(qb.list()).resolves.toEqual([]); + expect(mock.callsTo("auth/login")).toHaveLength(2); + }); + + it("失败分类:Fails./401 → auth;IP 封禁 403 / 429 → rate", async () => { + for (const version of ["4.6.7", "5.2.4"] as const) { + const m = await createMockQb({ version, banAfterFailures: 2 }); + const bad = createQbClient({ + instanceId: "qb-b", + baseUrl: m.url, + username: "admin", + password: "x", + ...fast, + retries: 0, + }); + expect((await bad.version().catch((e) => e)).kind).toBe("auth"); + expect((await bad.version().catch((e) => e)).kind).toBe("auth"); + const banned = await bad.version().catch((e) => e); + expect(banned.kind).toBe("rate"); + expect(banned.detail.status).toBe(403); + await m.close(); + } + mock = await createMockQb(); + mock.fault("auth/login", { status: 429 }); + const qb = createQbClient({ + instanceId: "qb-r", + baseUrl: mock.url, + username: "admin", + password: "adminadmin", + ...fast, + retries: 0, + }); + expect((await qb.version().catch((e) => e)).kind).toBe("rate"); + }); + + it("认证方式三选一:API Key 不登录、错 Key → auth 不重登;none 不登录,需要认证时 → auth", async () => { + mock = await createMockQb({ version: "5.2.4", apiKey: "qbt_good" }); + const bad = createQbClient({ + instanceId: "k", + baseUrl: mock.url, + ...qbAuthConfig({ authKind: "api_key", secret: "qbt_bad" }), + ...fast, + }); + const e = await bad.version().catch((x) => x); + expect(e.kind).toBe("auth"); + expect(e.message).toContain("API Key"); + const none = createQbClient({ instanceId: "n", baseUrl: mock.url, ...qbAuthConfig({ authKind: "none" }), ...fast }); + expect((await none.version().catch((x) => x)).kind).toBe("auth"); + expect(mock.callsTo("auth/login")).toHaveLength(0); + await mock.close(); + mock = await createMockQb({ version: "5.2.4", bypassAuth: true }); + const open = createQbClient({ + instanceId: "n2", + baseUrl: mock.url, + ...qbAuthConfig({ authKind: "none" }), + ...fast, + }); + expect((await open.version()).webApi).toBe("2.11.4"); + expect(mock.callsTo("auth/login")).toHaveLength(0); + }); + + it("qbAuthConfig:显式 authKind 优先;缺省按旧数据推断", () => { + expect(qbAuthConfig({ authKind: "api_key", username: "admin", secret: "k" })).toEqual({ + authMode: "api_key", + apiKey: "k", + }); + expect(qbAuthConfig({ authKind: "none", username: "admin", secret: "p" })).toEqual({ authMode: "none" }); + expect(qbAuthConfig({ username: "admin", secret: "p" })).toEqual({ + authMode: "password", + username: "admin", + password: "p", + }); + expect(qbAuthConfig({ username: null, secret: "k" })).toEqual({ authMode: "api_key", apiKey: "k" }); + expect(qbAuthConfig({})).toEqual({ authMode: "none" }); + }); + + it("pickSessionCookie:SID / QBT_SID_ / 自定义名都能取到", () => { + expect(pickSessionCookie(["SID=abc; HttpOnly; path=/"])).toBe("SID=abc"); + expect(pickSessionCookie(["foo=1; path=/", "QBT_SID_8080=xyz; HttpOnly"])).toBe("QBT_SID_8080=xyz"); + expect(pickSessionCookie(["MY_QB_SID=q; path=/"])).toBe("MY_QB_SID=q"); + expect(pickSessionCookie(["session=zz"])).toBe("session=zz"); + expect(pickSessionCookie(["SID=; Max-Age=0"])).toBeNull(); + expect(pickSessionCookie([])).toBeNull(); + }); + it("会话过期 403 → 自动重登一次后成功", async () => { const { mock, qb } = await setup(); await qb.list(); diff --git a/packages/contracts/src/instances.ts b/packages/contracts/src/instances.ts index 881ea1a8..27dfd035 100644 --- a/packages/contracts/src/instances.ts +++ b/packages/contracts/src/instances.ts @@ -50,6 +50,13 @@ const Credential = z.string().min(1).max(4096); export const DownloaderKind = z.enum(["qbittorrent"]); export type DownloaderKind = z.infer; +/** + * 下载器认证方式(存 settings.authKind):password 账密登录(凭证=密码);api_key qB 5.2+ WebUI 生成的 API Key + * (凭证=Key,走 Authorization: Bearer,不登录);none 无认证(qB 对内网白名单跳过认证)。 + */ +export const DownloaderAuthKind = z.enum(["password", "api_key", "none"]); +export type DownloaderAuthKind = z.infer; + export const DownloaderInstance = EntityBase.extend({ name: z.string().min(1), kind: DownloaderKind, @@ -78,7 +85,11 @@ export const DownloaderInstanceCreate = DownloaderInstance.pick({ settings: true, }) .partial({ kind: true, username: true, managedTag: true, cacheVolumeKey: true, enabled: true, settings: true }) - .extend({ credential: Credential.optional() }); + .extend({ + credential: Credential.optional(), + /** 认证方式,写入 settings.authKind;缺省按"有用户名 → 账密,否则有凭证 → API Key,否则无认证"推断 */ + authKind: DownloaderAuthKind.optional(), + }); export type DownloaderInstanceCreate = z.infer; export const DownloaderInstanceUpdate = DownloaderInstanceCreate.partial(); export type DownloaderInstanceUpdate = z.infer; diff --git a/packages/testkit/src/qbittorrent/bencode.ts b/packages/testkit/src/qbittorrent/bencode.ts index 1dd6a127..3102f17a 100644 --- a/packages/testkit/src/qbittorrent/bencode.ts +++ b/packages/testkit/src/qbittorrent/bencode.ts @@ -65,8 +65,14 @@ export interface MakeTorrentInput { pieceLength?: number; private?: boolean; announce?: string; - /** 改变伪 pieces 内容,用于构造同名同大小但 infohash 不同的种子 */ + /** 改变伪 pieces 内容,用于构造同名同大小但 infohash 不同的种子(realPieces 时也参与确定性内容生成) */ pieceSeed?: number; + /** + * 生成真实 piece hash(可被真 qB 校验通过):true 时按 pieceSeed + 文件路径生成确定性内容, + * 也可传函数自定义每个文件的内容(长度须等于 length)。内容随返回值 contents 给出,调用方写盘后 qB 重新校验即完成。 + * 默认 false:pieces 为伪值,不可真正下载(旧行为)。 + */ + realPieces?: boolean | ((path: string, length: number) => Uint8Array); } export interface MadeTorrent { @@ -75,9 +81,28 @@ export interface MadeTorrent { name: string; totalSize: number; files: { path: string; length: number }[]; + /** realPieces 时:每个文件的内容(path 同 files[].path,含种子根目录) */ + contents?: { path: string; bytes: Uint8Array }[]; } -/** 构造一个合法的 v1 .torrent(pieces 为伪值,不可真正下载)。 */ +/** 确定性伪随机内容(xorshift32),同 seed + path 得到同样字节 */ +export function deterministicContent(seedKey: string, length: number): Uint8Array { + let x = 2166136261; + for (let i = 0; i < seedKey.length; i++) x = Math.imul(x ^ seedKey.charCodeAt(i), 16777619) >>> 0; + if (x === 0) x = 1; + const out = new Uint8Array(length); + for (let i = 0; i < length; i++) { + x ^= x << 13; + x >>>= 0; + x ^= x >>> 17; + x ^= x << 5; + x >>>= 0; + out[i] = x & 0xff; + } + return out; +} + +/** 构造一个合法的 v1 .torrent;realPieces 时 pieces 为按内容计算的真实 SHA-1,否则为伪值。 */ export function makeTorrent(input: MakeTorrentInput): MadeTorrent { const pieceLength = input.pieceLength ?? 4 * 1024 * 1024; const info: { [k: string]: BVal } = { name: input.name, "piece length": pieceLength }; @@ -90,13 +115,46 @@ export function makeTorrent(input: MakeTorrentInput): MadeTorrent { files = [{ path: input.name, length: info.length as number }]; } const total = files.reduce((a, f) => a + f.length, 0); - const pieces = Math.max(1, Math.ceil(total / pieceLength)); - info.pieces = Buffer.alloc(20 * Math.min(pieces, 4), input.pieceSeed ?? 7); + let contents: MadeTorrent["contents"]; + if (input.realPieces) { + const gen = + typeof input.realPieces === "function" + ? input.realPieces + : (path: string, length: number) => deterministicContent(`${input.pieceSeed ?? 7}:${path}`, length); + contents = files.map((f) => { + const bytes = gen(f.path, f.length); + if (bytes.length !== f.length) throw new Error(`realPieces: ${f.path} 内容长度 ${bytes.length} ≠ ${f.length}`); + return { path: f.path, bytes }; + }); + // v1:所有文件按顺序拼成一个字节流,按 piece length 切片逐片 SHA-1 + const hashes: Buffer[] = []; + let h = createHash("sha1"); + let filled = 0; + for (const c of contents) { + let off = 0; + while (off < c.bytes.length) { + const n = Math.min(pieceLength - filled, c.bytes.length - off); + h.update(c.bytes.subarray(off, off + n)); + off += n; + filled += n; + if (filled === pieceLength) { + hashes.push(h.digest()); + h = createHash("sha1"); + filled = 0; + } + } + } + if (filled > 0 || hashes.length === 0) hashes.push(h.digest()); + info.pieces = Buffer.concat(hashes); + } else { + const pieces = Math.max(1, Math.ceil(total / pieceLength)); + info.pieces = Buffer.alloc(20 * Math.min(pieces, 4), input.pieceSeed ?? 7); + } if (input.private) info.private = 1; const root: { [k: string]: BVal } = { announce: input.announce ?? "https://tracker.example.invalid/announce", info }; const bytes = bencode(root); const hash = createHash("sha1").update(bencode(info)).digest("hex"); - return { bytes, hash, name: input.name, totalSize: total, files }; + return { bytes, hash, name: input.name, totalSize: total, files, ...(contents ? { contents } : {}) }; } export function makeMagnet(hash: string, name?: string): string { diff --git a/packages/testkit/src/qbittorrent/index.ts b/packages/testkit/src/qbittorrent/index.ts index d845d499..657617ea 100644 --- a/packages/testkit/src/qbittorrent/index.ts +++ b/packages/testkit/src/qbittorrent/index.ts @@ -1,4 +1,5 @@ export { + deterministicContent, inspectTorrent as inspectTorrentFixture, type MadeTorrent, type MakeTorrentInput, diff --git a/packages/testkit/src/qbittorrent/mock-qb.ts b/packages/testkit/src/qbittorrent/mock-qb.ts index 055ca26f..0afd0fa2 100644 --- a/packages/testkit/src/qbittorrent/mock-qb.ts +++ b/packages/testkit/src/qbittorrent/mock-qb.ts @@ -70,7 +70,17 @@ export interface MockQbFault { } export interface MockQbOptions { - version?: "4.6.7" | "5.1.0"; + /** + * 登录行为随版本:≤5.1 成功 200 "Ok." + SID、失败 200 "Fails."; + * 5.2.x 成功 204 空体 + QBT_SID_、失败 401 "Unauthorized"。 + */ + version?: "4.6.7" | "5.1.0" | "5.2.4"; + /** 模拟 qB"对白名单子网跳过认证":不带会话/Key 的请求也放行 */ + bypassAuth?: boolean; + /** 连续登录失败达到该次数后封禁(403 "Your IP address has been banned…"),默认不封禁 */ + banAfterFailures?: number; + /** 5.2.x 会话 cookie 名里的端口(真实 qB 用容器内 WebUI 端口),默认 8080 */ + webUiPort?: number; username?: string; password?: string; apiKey?: string; @@ -136,6 +146,9 @@ export async function createMockQb(opts: MockQbOptions = {}): Promise { const faults = new Map(); const calls: MockQbCall[] = []; let rid = 0; + const v52 = version.startsWith("5.2"); + const cookieName = v52 ? `QBT_SID_${opts.webUiPort ?? 8080}` : "SID"; + let loginFailures = 0; const pausedState = (done: boolean) => (v5 ? (done ? "stoppedUP" : "stoppedDL") : done ? "pausedUP" : "pausedDL"); @@ -557,17 +570,25 @@ export async function createMockQb(opts: MockQbOptions = {}): Promise { } if (endpoint === "auth/login") { + if (opts.banAfterFailures !== undefined && loginFailures >= opts.banAfterFailures) + return send(403, "Your IP address has been banned after too many failed authentication attempts."); if (form.username === user && form.password === pass) { + loginFailures = 0; const sid = randomBytes(16).toString("hex"); sessions.add(sid); - return send(200, "Ok.", { "set-cookie": `SID=${sid}; HttpOnly; path=/` }); + return v52 + ? send(204, "", { "set-cookie": `${cookieName}=${sid}; HttpOnly; SameSite=Lax; path=/` }) + : send(200, "Ok.", { "set-cookie": `SID=${sid}; HttpOnly; path=/` }); } - return send(200, "Fails."); + loginFailures++; + return v52 ? send(401, "Unauthorized") : send(200, "Fails."); } const bearer = /^Bearer (.+)$/.exec(req.headers.authorization ?? "")?.[1]; - const sid = parseCookies(req.headers.cookie).SID; - const authed = (opts.apiKey && bearer === opts.apiKey) || (sid !== undefined && sessions.has(sid)); - if (!authed) return send(403, "Forbidden"); + const sid = parseCookies(req.headers.cookie)[cookieName]; + const authed = + opts.bypassAuth === true || (opts.apiKey && bearer === opts.apiKey) || (sid !== undefined && sessions.has(sid)); + // 真 qB 5.2:错误的 Bearer Key → 403;没有会话 → 401 + if (!authed) return v52 && !bearer ? send(401, "Unauthorized") : send(403, "Forbidden"); const h = handlers[endpoint]; if (!h) return send(404, "Not Found"); diff --git a/packages/testkit/test/mock-qb.test.ts b/packages/testkit/test/mock-qb.test.ts index 4b68314e..8c74eab2 100644 --- a/packages/testkit/test/mock-qb.test.ts +++ b/packages/testkit/test/mock-qb.test.ts @@ -3,7 +3,8 @@ import { mkdtemp, readFile, rm } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { afterEach, describe, expect, it } from "vitest"; -import { createMockQb, type MockQb } from "../src/index.ts"; +import { createMockQb, type MockQb, makeTorrent } from "../src/index.ts"; +import { bdecodeLoose } from "../src/qbittorrent/bencode.ts"; let mock: MockQb | null = null; let dir: string | null = null; @@ -88,3 +89,46 @@ describe("mock qB 落盘(B8)", () => { expect(readdirSync(dir)).toEqual([]); }); }); + +describe("makeTorrent realPieces", () => { + it("按内容计算真实 piece hash:跨文件切片,片数正确,内容确定", async () => { + const { createHash } = await import("node:crypto"); + const t = makeTorrent({ + name: "Real.Pieces", + pieceLength: 16384, + files: [ + { path: "a.mkv", length: 20000 }, + { path: "Subs/a.ass", length: 30000 }, + ], + realPieces: true, + }); + expect(t.contents?.map((c) => [c.path, c.bytes.length])).toEqual([ + ["Real.Pieces/a.mkv", 20000], + ["Real.Pieces/Subs/a.ass", 30000], + ]); + const all = Buffer.concat(t.contents!.map((c) => Buffer.from(c.bytes))); + const want = Buffer.concat( + [0, 1, 2, 3].map((i) => + createHash("sha1") + .update(all.subarray(i * 16384, (i + 1) * 16384)) + .digest(), + ), + ); + const info = (bdecodeLoose(t.bytes) as { info: { pieces: Buffer } }).info; + expect(info.pieces.equals(want)).toBe(true); + const again = makeTorrent({ + name: "Real.Pieces", + pieceLength: 16384, + files: [ + { path: "a.mkv", length: 20000 }, + { path: "Subs/a.ass", length: 30000 }, + ], + realPieces: true, + }); + expect(again.hash).toBe(t.hash); + expect( + makeTorrent({ name: "x", length: 10, realPieces: () => new Uint8Array(10) }).contents?.[0]?.bytes, + ).toHaveLength(10); + expect(() => makeTorrent({ name: "x", length: 10, realPieces: () => new Uint8Array(3) })).toThrow(); + }); +}); From 05d1f76b9f8c70c7f425a0102ca8402d071d31de Mon Sep 17 00:00:00 2001 From: root Date: Tue, 29 Sep 2026 20:54:10 +0800 Subject: [PATCH 2/4] feat(library): download history lists adopted tasks and organize records MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Migration 0053: download_task.media_entity_id (FK, set null on delete), backfilled from plan slot / source release / latest organize record. - Submitted and adopted tasks carry the entity; entity merge follows it. - GET /library/{id}: downloads also matched by dt.media_entity_id (adopted takeover tasks), plus organizationRecords (incl. manual path organizes); library list / inventory counts use it too. - Web detail: "接管" marker and a list of records without a download task. Co-Authored-By: Claude Opus 5.5 --- .../__tests__/vertical.test.ts | 15 ++++- .../acquisition-organize/routes/library.ts | 55 +++++++++++++++++-- .../services/downloads.ts | 10 +++- .../service/identity-admin.ts | 5 ++ apps/web/app/routes/console/library.$id.tsx | 32 ++++++++++- packages/contracts/src/execution.ts | 2 + packages/contracts/src/organization.ts | 21 +++++++ .../migrations/0053_download_task_entity.ts | 29 ++++++++++ packages/db/src/types.ts | 1 + 9 files changed, 159 insertions(+), 11 deletions(-) create mode 100644 packages/db/migrations/0053_download_task_entity.ts diff --git a/apps/server/src/modules/acquisition-organize/__tests__/vertical.test.ts b/apps/server/src/modules/acquisition-organize/__tests__/vertical.test.ts index 5f76aaeb..a3606902 100644 --- a/apps/server/src/modules/acquisition-organize/__tests__/vertical.test.ts +++ b/apps/server/src/modules/acquisition-organize/__tests__/vertical.test.ts @@ -314,7 +314,12 @@ describe.skipIf(!hasPg)("acquisition-organize 纵切(真实 PG + mock qB + 临 .selectAll() .where("id", "=", sub.downloadTaskId!) .executeTakeFirstOrThrow(); - expect(row).toMatchObject({ origin: "adopted", managed: true, acquisition_plan_id: ev.plans[0] }); + expect(row).toMatchObject({ + origin: "adopted", + managed: true, + acquisition_plan_id: ev.plans[0], + media_entity_id: e, + }); expect(qb.callsTo("torrents/add").length).toBe(1); }, 30_000); @@ -460,6 +465,14 @@ describe.skipIf(!hasPg)("acquisition-organize 纵切(真实 PG + mock qB + 临 ); expect(det.body.data.candidates.some((c: { decision: string }) => c.decision === "reject")).toBe(true); expect(det.body.data.downloads[0].organizationRecordIds.length).toBeGreaterThan(0); + // QB-1:下载任务直接记作品;整理记录按作品列出 + const dune = det.body.data.downloads.find((x: { downloadTask: { name: string } }) => + x.downloadTask.name.startsWith("Dune.Part.Two"), + ); + expect(dune.downloadTask.mediaEntityId).toBe(movieId); + expect(det.body.data.organizationRecords.map((r: { id: string }) => r.id)).toEqual( + expect.arrayContaining(dune.organizationRecordIds), + ); const inv = await h.api("GET", `/inventory?entityIds=${movieId}`); expect(inv.body.data.items[0].versions[0]).toMatchObject({ resolution: "1080p", available: true }); const ms = await h.api("GET", `/inventory/media-servers?mediaEntityId=${movieId}`); diff --git a/apps/server/src/modules/acquisition-organize/routes/library.ts b/apps/server/src/modules/acquisition-organize/routes/library.ts index 7950c4fc..cb4b13ad 100644 --- a/apps/server/src/modules/acquisition-organize/routes/library.ts +++ b/apps/server/src/modules/acquisition-organize/routes/library.ts @@ -147,7 +147,7 @@ export function registerLibraryRoutes(app: ModuleApp, d: AoDeps) { .select( sql`(select count(*) from download_task dt left join acquisition_plan p on p.id = dt.acquisition_plan_id left join acquisition_slot s on s.id = p.slot_id left join source_release sr on sr.id = dt.source_release_id - where (s.media_entity_id = e.id or sr.media_entity_id = e.id) and dt.removed_at is null + where (dt.media_entity_id = e.id or s.media_entity_id = e.id or sr.media_entity_id = e.id) and dt.removed_at is null and dt.status in ('queued','submitting','downloading','stalled','paused'))::int`.as("active_downloads"), ) .select(sql`e.metadata #>> '{posterPath,value}'`.as("poster_path")) @@ -205,7 +205,7 @@ export function registerLibraryRoutes(app: ModuleApp, d: AoDeps) { if (query.filter === "downloading") q = q.where( sql`exists (select 1 from download_task dt left join acquisition_plan p on p.id = dt.acquisition_plan_id left join acquisition_slot s on s.id = p.slot_id - where s.media_entity_id = e.id and dt.removed_at is null and dt.status in ('queued','submitting','downloading','stalled','paused'))`, + where (dt.media_entity_id = e.id or s.media_entity_id = e.id) and dt.removed_at is null and dt.status in ('queued','submitting','downloading','stalled','paused'))`, ); const p = await paginate(q, query, "e."); const items: LibraryItem[] = p.rows.map((r) => ({ @@ -288,10 +288,36 @@ export function registerLibraryRoutes(app: ModuleApp, d: AoDeps) { .leftJoin("acquisition_slot as s", "s.id", "p.slot_id") .leftJoin("source_release as sr", "sr.id", "dt.source_release_id") .selectAll("dt") - .where((eb) => eb.or([eb("s.media_entity_id", "=", id), eb("sr.media_entity_id", "=", id)])) + // 接管任务(无计划 / 站点资源)按识别 / 指定 / 整理回填的 dt.media_entity_id 归属 + .where((eb) => + eb.or([ + eb("dt.media_entity_id", "=", id), + eb("s.media_entity_id", "=", id), + eb("sr.media_entity_id", "=", id), + ]), + ) .orderBy("dt.created_at", "desc") .limit(100) .execute(); + const entityRecs = await db() + .selectFrom("organization_record") + .select([ + "id", + "status", + "action", + "trigger", + "download_task_id", + "source_path", + "target_path", + "file_succeeded", + "file_total", + "created_at", + "finished_at", + ]) + .where("media_entity_id", "=", id) + .orderBy("created_at", "desc") + .limit(100) + .execute(); const planIds = dts.map((t) => t.acquisition_plan_id).filter((x): x is string => !!x); const plans = planIds.length ? await db().selectFrom("acquisition_plan").selectAll().where("id", "in", planIds).execute() @@ -339,6 +365,19 @@ export function registerLibraryRoutes(app: ModuleApp, d: AoDeps) { lastOrganizationStatus: mine[mine.length - 1]?.status ?? null, }; }), + organizationRecords: entityRecs.map((r) => ({ + id: r.id, + status: r.status, + action: r.action, + trigger: r.trigger, + downloadTaskId: r.download_task_id, + sourcePath: r.source_path, + targetPath: r.target_path, + fileSucceeded: r.file_succeeded, + fileTotal: r.file_total, + createdAt: new Date(r.created_at).toISOString(), + finishedAt: r.finished_at ? new Date(r.finished_at).toISOString() : null, + })), }; return detail; }, @@ -392,8 +431,14 @@ export function registerLibraryRoutes(app: ModuleApp, d: AoDeps) { .leftJoin("acquisition_slot as s", "s.id", "p.slot_id") .leftJoin("source_release as sr", "sr.id", "dt.source_release_id") .select(["dt.id", "dt.name", "dt.status", "dt.progress"]) - .select(sql`coalesce(s.media_entity_id, sr.media_entity_id)`.as("entity_id")) - .where((eb) => eb.or([eb("s.media_entity_id", "in", ids), eb("sr.media_entity_id", "in", ids)])) + .select(sql`coalesce(dt.media_entity_id, s.media_entity_id, sr.media_entity_id)`.as("entity_id")) + .where((eb) => + eb.or([ + eb("dt.media_entity_id", "in", ids), + eb("s.media_entity_id", "in", ids), + eb("sr.media_entity_id", "in", ids), + ]), + ) .where("dt.removed_at", "is", null) .where("dt.status", "in", [...ACTIVE_DL]) .execute(), diff --git a/apps/server/src/modules/acquisition-organize/services/downloads.ts b/apps/server/src/modules/acquisition-organize/services/downloads.ts index 608a47fd..803ade94 100644 --- a/apps/server/src/modules/acquisition-organize/services/downloads.ts +++ b/apps/server/src/modules/acquisition-organize/services/downloads.ts @@ -313,15 +313,18 @@ export async function submitDownload(d: AoDeps, planId: string, opts: { runId?: } const r = res as Extract; const ins = await sql<{ id: string; inserted: boolean }>` - insert into download_task (instance_id, torrent_hash, infohash_v1, infohash_v2, acquisition_plan_id, source_release_id, name, + insert into download_task (instance_id, torrent_hash, infohash_v1, infohash_v2, acquisition_plan_id, source_release_id, + media_entity_id, name, save_path, content_path, category, tags, managed, origin, status, size_bytes, next_check_at) values (${topo.inst.id}, ${r.hash}, ${r.hash.length === 40 ? r.hash : null}, ${r.hash.length === 64 ? r.hash : null}, - ${plan.id}, ${rel.id}, ${r.name}, ${r.savePath}, ${r.contentPath}, ${topo.category}, + ${plan.id}, ${rel.id}, (select media_entity_id from acquisition_slot where id = ${plan.slot_id}), + ${r.name}, ${r.savePath}, ${r.contentPath}, ${topo.category}, ${r.tags}, true, ${r.kind === "already_exists" ? "adopted" : "submitted"}, ${r.kind === "accepted_unconfirmed" ? "submitting" : "queued"}, ${rel.size_bytes}, now()) on conflict (instance_id, torrent_hash) do update set acquisition_plan_id = coalesce(download_task.acquisition_plan_id, excluded.acquisition_plan_id), source_release_id = coalesce(download_task.source_release_id, excluded.source_release_id), + media_entity_id = coalesce(download_task.media_entity_id, excluded.media_entity_id), managed = true, tags = excluded.tags, updated_at = now() returning id, (xmax = 0) as inserted`.execute(trx); dtId = ins.rows[0]!.id; @@ -787,7 +790,7 @@ async function adopt(d: AoDeps, topo: InstanceTopology, t: QbTorrent, now: Date, const db = d.ctx.db; const rel = await db .selectFrom("source_release") - .select("id") + .select(["id", "media_entity_id"]) .where((eb) => eb.or([eb("infohash_v1", "=", t.hash), eb("infohash_v2", "=", t.hash)])) .executeTakeFirst(); const obs = mapQbState( @@ -804,6 +807,7 @@ async function adopt(d: AoDeps, topo: InstanceTopology, t: QbTorrent, now: Date, infohash_v1: t.infohashV1, infohash_v2: t.infohashV2, source_release_id: rel?.id ?? null, + media_entity_id: rel?.media_entity_id ?? null, name: t.name, save_path: t.savePath, content_path: t.contentPath, diff --git a/apps/server/src/modules/sources-identity/service/identity-admin.ts b/apps/server/src/modules/sources-identity/service/identity-admin.ts index 76ffec59..b357424c 100644 --- a/apps/server/src/modules/sources-identity/service/identity-admin.ts +++ b/apps/server/src/modules/sources-identity/service/identity-admin.ts @@ -440,6 +440,11 @@ export async function mergeEntities( .set({ resolved_media_entity_id: intoResolved }) .where("resolved_media_entity_id", "=", fromId) .execute(); + await trx + .updateTable("download_task") + .set({ media_entity_id: intoResolved }) + .where("media_entity_id", "=", fromId) + .execute(); await trx .updateTable("media_entity") .set({ status: "merged", merged_into_id: intoResolved }) diff --git a/apps/web/app/routes/console/library.$id.tsx b/apps/web/app/routes/console/library.$id.tsx index 42b00361..3939615d 100644 --- a/apps/web/app/routes/console/library.$id.tsx +++ b/apps/web/app/routes/console/library.$id.tsx @@ -815,7 +815,9 @@ function DownloadsPanel({ detail }: { detail: LibraryDetail }) {

{d.plan ? `${d.plan.purpose === "upgrade" ? "升级" : d.plan.purpose === "fill" ? "补缺" : "换源"}计划 · ` - : ""} + : d.downloadTask.origin === "adopted" + ? "接管 · " + : ""} {formatDateTime(d.downloadTask.createdAt)}

@@ -872,6 +874,9 @@ function DownloadsPanel({ detail }: { detail: LibraryDetail }) { ), }, ]; + // 没有下载任务的整理记录(文件管理按路径手动整理等),下载任务的记录已在上表 + const listed = new Set(detail.downloads.map((d) => d.downloadTask.id)); + const manual = (detail.organizationRecords ?? []).filter((r) => !r.downloadTaskId || !listed.has(r.downloadTaskId)); return ( d.downloadTask.id} - empty={{ title: "还没有下载记录" }} + empty={{ title: manual.length ? "没有下载任务，只有手动整理记录" : "还没有下载记录" }} /> + {manual.length ? ( +
+

其他整理记录

+
    + {manual.map((r) => ( +
  • + + + {r.targetPath ?? r.sourcePath ?? r.id} + + + {r.trigger === "manual" ? "手动" : r.trigger === "agent" ? "Agent" : "自动"} · {r.fileSucceeded}/ + {r.fileTotal} 个文件 · {formatDateTime(r.createdAt)} + +
    • + ))} +
+
+ ) : null} ); } diff --git a/packages/contracts/src/execution.ts b/packages/contracts/src/execution.ts index 8bf5d8df..06b808e6 100644 --- a/packages/contracts/src/execution.ts +++ b/packages/contracts/src/execution.ts @@ -189,6 +189,8 @@ export const DownloadTask = EntityBase.extend({ infohashV2: z.string().nullable(), acquisitionPlanId: Id.nullable(), sourceReleaseId: Id.nullable(), + /** 关联作品:计划 / 站点资源 / 接管任务按文件名识别 / 手动指定 / 手动整理后回填(QB-1) */ + mediaEntityId: Id.nullable().optional(), name: z.string(), savePath: z.string().nullable(), contentPath: z.string().nullable(), diff --git a/packages/contracts/src/organization.ts b/packages/contracts/src/organization.ts index 4b1e043b..b658e2c7 100644 --- a/packages/contracts/src/organization.ts +++ b/packages/contracts/src/organization.ts @@ -374,6 +374,27 @@ export const LibraryDetail = z.object({ slots: z.array(AcquisitionSlot), candidates: z.array(CandidateEvaluation), downloads: z.array(LibraryDownloadView), + /** + * 该作品的整理记录(最近 100 条,新→旧):含没有下载任务的手动整理(文件管理按路径整理)。 + * 下载历史里已按下载任务列出的记录也在这里,前端按 downloadTaskId 去重展示。 + */ + organizationRecords: z + .array( + z.object({ + id: Id, + status: OrganizationStatus, + action: OrganizationAction, + trigger: OrganizationTrigger, + downloadTaskId: Id.nullable(), + sourcePath: z.string().nullable(), + targetPath: z.string().nullable(), + fileSucceeded: NonNegInt, + fileTotal: NonNegInt, + createdAt: Timestamp, + finishedAt: Timestamp.nullable(), + }), + ) + .optional(), }); export type LibraryDetail = z.infer; diff --git a/packages/db/migrations/0053_download_task_entity.ts b/packages/db/migrations/0053_download_task_entity.ts new file mode 100644 index 00000000..a891dfcc --- /dev/null +++ b/packages/db/migrations/0053_download_task_entity.ts @@ -0,0 +1,29 @@ +import type { Kysely } from "kysely"; +import { execDdl } from "../src/migration-helpers.ts"; + +// QB-1:下载任务直接记关联作品(接管任务按文件名识别 / 手动指定 / 手动整理后回填),作品详情"下载历史"据此列出 +// 没有计划与站点资源的接管任务。已有行按 计划 Slot → 站点资源 → 最近一次整理记录 回填。 + +// biome-ignore lint/suspicious/noExplicitAny: 迁移不依赖应用表类型 +export async function up(db: Kysely): Promise { + await execDdl( + db, + `alter table download_task add column if not exists media_entity_id uuid references media_entity (id) on delete set null; + create index if not exists download_task_media_entity_idx on download_task (media_entity_id) where media_entity_id is not null; + update download_task dt set media_entity_id = coalesce( + (select s.media_entity_id from acquisition_plan p join acquisition_slot s on s.id = p.slot_id where p.id = dt.acquisition_plan_id), + (select sr.media_entity_id from source_release sr where sr.id = dt.source_release_id), + (select r.media_entity_id from organization_record r + where r.download_task_id = dt.id and r.media_entity_id is not null order by r.created_at desc limit 1)) + where dt.media_entity_id is null;`, + ); +} + +// biome-ignore lint/suspicious/noExplicitAny: 同上 +export async function down(db: Kysely): Promise { + await execDdl( + db, + `drop index if exists download_task_media_entity_idx; + alter table download_task drop column if exists media_entity_id;`, + ); +} diff --git a/packages/db/src/types.ts b/packages/db/src/types.ts index e244423f..a0e6b14d 100644 --- a/packages/db/src/types.ts +++ b/packages/db/src/types.ts @@ -701,6 +701,7 @@ export interface DownloadTaskTable { last_error: Json | null; created_at: GeneratedTimestamptz; updated_at: GeneratedTimestamptz; + media_entity_id: string | null; } export interface DownloaderInstanceTable { From e7de7e2e69e36359e01565a4b3b93478eb192e34 Mon Sep 17 00:00:00 2001 From: root Date: Tue, 29 Sep 2026 20:54:19 +0800 Subject: [PATCH 3/4] feat(organize): identify takeover downloads by file name and close organize decisions Takeover identify (item 2): - Adopted tasks without source release / plan: on completion planImport parses main video path (file + parent + grandparent dirs, like MP MetaInfoPath) merged with the torrent name, then runs ID-2 matching via identifyParsed (local recall + TMDB; trigger "takeover" may query TMDB). auto -> bind/create entity, backfill download_task.media_entity_id and organize right away; no pool entry, no identity decision. - Failure / ambiguity: organize_identity decision with evidence and options: candidates, select entity (local / TMDB), provisional entity, dismiss. Decision closure (item 3): - decision-hooks registry: the decisions resolve route runs the owning module's executor before closing; errors keep the decision open. - organize_identity: backfill entity, freeze manual plan, enqueue run. - organize_blocked with target_exists: replace (verify + trash old) / keep both (rename " (n)" with sidecars) / skip. - Successful organize closes both organize_identity and organize_blocked of the same download task (resolution manual_organize), also for manual path organizes matched to download tasks via path mappings. Co-Authored-By: Claude Opus 5.5 --- .../acquisition-organize/__tests__/harness.ts | 10 +- .../__tests__/takeover.test.ts | 283 +++++++++++++ .../src/modules/acquisition-organize/index.ts | 4 + .../src/modules/acquisition-organize/ports.ts | 44 ++ .../acquisition-organize/services/common.ts | 6 +- .../acquisition-organize/services/organize.ts | 86 +++- .../acquisition-organize/services/takeover.ts | 390 ++++++++++++++++++ apps/server/src/modules/decision-hooks.ts | 38 ++ .../sources-identity/routes/decisions.ts | 21 +- .../sources-identity/service/resolve.ts | 187 ++++++++- .../sources-identity/lazy-identity.test.ts | 44 +- docs/runbook/files.md | 12 + docs/runbook/identity.md | 19 + 13 files changed, 1127 insertions(+), 17 deletions(-) create mode 100644 apps/server/src/modules/acquisition-organize/__tests__/takeover.test.ts create mode 100644 apps/server/src/modules/acquisition-organize/services/takeover.ts create mode 100644 apps/server/src/modules/decision-hooks.ts diff --git a/apps/server/src/modules/acquisition-organize/__tests__/harness.ts b/apps/server/src/modules/acquisition-organize/__tests__/harness.ts index 9e689fed..c04b34fc 100644 --- a/apps/server/src/modules/acquisition-organize/__tests__/harness.ts +++ b/apps/server/src/modules/acquisition-organize/__tests__/harness.ts @@ -14,6 +14,7 @@ import { hashApiKey } from "../../../http/auth.ts"; import { EventBus } from "../../../http/sse.ts"; import { createLogger } from "../../../logger.ts"; import { createCapabilityRegistry } from "../../agent/capability-registry.ts"; +import { createSourcesIdentityModule } from "../../sources-identity/index.ts"; import type { ServerContext } from "../../types.ts"; import { type AoOverrides, createAcquisitionOrganizeModule } from "../index.ts"; import type { AoDeps } from "../services/common.ts"; @@ -129,7 +130,12 @@ export interface Harness { close(): Promise; } -export async function createHarness(name: string, overrides: AoOverrides = {}): Promise { +export async function createHarness( + name: string, + overrides: AoOverrides = {}, + /** withSources:同时装配 sources-identity 模块(待处理中心路由等) */ + opts: { withSources?: boolean } = {}, +): Promise { const schema = `ao_test_${name}_${process.pid.toString(36)}${Math.random().toString(36).slice(2, 7)}`; const admin = new pg.Client({ connectionString: DB_URL }); await admin.connect(); @@ -207,7 +213,7 @@ export async function createHarness(name: string, overrides: AoOverrides = {}): role: "all", version: "test", db, - modules: [module], + modules: opts.withSources ? [module, createSourcesIdentityModule()] : [module], ctx, apiKeyStore: { async findByHash(hash) { diff --git a/apps/server/src/modules/acquisition-organize/__tests__/takeover.test.ts b/apps/server/src/modules/acquisition-organize/__tests__/takeover.test.ts new file mode 100644 index 00000000..3a5f280f --- /dev/null +++ b/apps/server/src/modules/acquisition-organize/__tests__/takeover.test.ts @@ -0,0 +1,283 @@ +import { existsSync, readdirSync, statSync } from "node:fs"; +import { join } from "node:path"; +import { createMockQb, createTempTree, type MockQb, type TempTree } from "@umt/testkit"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { addAliasIfAbsent } from "../../sources-identity/service/entities.ts"; +import { syncDownloads } from "../services/downloads.ts"; +import { executeRecord, planImport } from "../services/organize.ts"; +import { organizeDecisionHook, pickMainVideo, takeoverParsed } from "../services/takeover.ts"; +import { createHarness, type Harness, pgAvailable, seedBase, seedEntity } from "./harness.ts"; + +// QB-1:接管任务(用户手动加进 qB、贴管理标签,没有站点资源)完成后按 种子名 + 主视频文件名 识别 → 直接整理; +// 识别失败建带选项的事项(候选 / 选择作品 / 临时作品 / 忽略),处理后执行并闭环;目标已存在 → 替换 / 保留两者 / 跳过; +// 手动按路径整理也关闭同一下载任务的整理事项;作品详情下载历史列出接管任务与整理记录。 + +const hasPg = await pgAvailable(); +const SIZE = 300_000; + +function walkFiles(dir: string): string[] { + const out: string[] = []; + for (const n of readdirSync(dir)) { + const p = join(dir, n); + if (statSync(p).isDirectory()) out.push(...walkFiles(p)); + else out.push(p); + } + return out; +} + +describe("接管识别:文件名解析", () => { + it("主视频取要下载的最大视频,跳过 sample;文件名 + 目录 + 种子名合并", () => { + expect( + pickMainVideo([ + { name: "Movie.2021/Sample/sample.mkv", size: 9e9 }, + { name: "Movie.2021/Movie.2021.1080p.mkv", size: 5e9 }, + { name: "Movie.2021/Extra.mkv", size: 6e9, priority: 0 }, + { name: "Movie.2021/Movie.2021.nfo", size: 1e3 }, + ]), + ).toBe("Movie.2021/Movie.2021.1080p.mkv"); + expect(pickMainVideo([{ name: "a.nfo", size: 1 }])).toBeNull(); + // 文件名只有集号:标题 / 季从目录名(=种子根目录)补 + const p = takeoverParsed("The.Bear.S02.1080p.WEB-DL", "The.Bear.S02.1080p.WEB-DL/E03.mkv"); + expect(p.title?.toLowerCase()).toContain("bear"); + expect(p.season).toBe(2); + expect(p.episodes).toEqual({ start: 3, end: 3 }); + // 单文件种子:种子名即文件名 + const m = takeoverParsed("Oppenheimer.2023.2160p.UHD.BluRay.mkv", "Oppenheimer.2023.2160p.UHD.BluRay.mkv"); + expect(m.year).toBe(2023); + // 文件名解析不出标题:用种子名 + expect(takeoverParsed("Heat.1995.1080p.BluRay", "Heat.1995.1080p.BluRay/0001.mkv").year).toBe(1995); + }); +}); + +describe.skipIf(!hasPg)("QB-1 接管任务识别与事项闭环(真实 PG + mock qB + 临时目录)", () => { + let h: Harness; + let qb: MockQb; + let lib: TempTree; + let cache: TempTree; + let n = 0; + const actor = { type: "user", id: "tester" }; + + beforeAll(async () => { + qb = await createMockQb({ addVisibleAfterPolls: 0 }); + lib = await createTempTree({}, { sentinel: true, prefix: "umt-lib-" }); + cache = await createTempTree({}, { prefix: "umt-cache-" }); + h = await createHarness("takeover", {}, { withSources: true }); + await seedBase(h.db, { libraryPath: lib.root, cachePath: cache.root, qbUrl: qb.url }); + await h.setSettings("retention", { seedRetentionDays: 0 }); + }, 60_000); + + afterAll(async () => { + await h?.close(); + await qb?.close(); + await lib?.cleanup(); + await cache?.cleanup(); + }); + + /** 用户在 qB 里手动加的任务(贴管理标签)→ 同步接管 → 完成落盘 */ + async function adoptCompleted(name: string, file: string, size = SIZE) { + n++; + const hash = n.toString(16).padStart(40, "a"); + qb.seed({ hash, name, tags: ["umt"], save_path: cache.root }); + await qb.complete(hash, { files: [{ path: `${name}/${file}`, size }] }); + const sync = await syncDownloads(h.d); + expect(sync.adopted).toBeGreaterThanOrEqual(1); + const dt = await h.db + .selectFrom("download_task") + .selectAll() + .where("torrent_hash", "=", hash) + .executeTakeFirstOrThrow(); + expect(dt).toMatchObject({ origin: "adopted", source_release_id: null, status: "completed" }); + return dt; + } + + const decisionOf = (key: string) => + h.db + .selectFrom("decision") + .selectAll() + .where("dedupe_key", "=", key) + .orderBy("created_at", "desc") + .executeTakeFirst(); + + let movieId: string; + let firstDtId: string; + + it("识别成功(本地别名召回):不建事项,直接整理入库并回填下载任务的作品", async () => { + movieId = await seedEntity(h.db, { kind: "movie", title: "Quiet Harbor", year: 2021 }); + await addAliasIfAbsent(h.db, movieId, "Quiet Harbor", "original", "user"); + const dt = await adoptCompleted( + "Quiet.Harbor.2021.1080p.BluRay.x264-GRP", + "Quiet.Harbor.2021.1080p.BluRay.x264-GRP.mkv", + ); + firstDtId = dt.id; + const plan = await planImport(h.d, { downloadTaskId: dt.id, trigger: "auto", actorType: "system" }); + expect(plan.status).toBe("running"); + const done = await executeRecord(h.d, plan.recordId!); + expect(done.status).toBe("succeeded"); + const row = await h.db.selectFrom("download_task").selectAll().where("id", "=", dt.id).executeTakeFirstOrThrow(); + expect(row).toMatchObject({ media_entity_id: movieId, status: "imported" }); + expect(await decisionOf(`organize_identity:${dt.id}`)).toBeUndefined(); + }); + + let unknownDtId: string; + + it("识别失败:事项带 选择作品(本地 / TMDB)/ 临时作品 / 忽略;选临时作品 → 建临时作品并整理,事项闭环", async () => { + const dt = await adoptCompleted("Zzqx.Vorlanth.2019.1080p.WEB-DL", "Zzqx.Vorlanth.2019.1080p.WEB-DL.mkv"); + unknownDtId = dt.id; + const plan = await planImport(h.d, { downloadTaskId: dt.id, trigger: "auto", actorType: "system" }); + expect(plan).toMatchObject({ status: "failed", reason: "identity_unknown" }); + const dec = (await decisionOf(`organize_identity:${dt.id}`))!; + expect(dec).toBeTruthy(); + const keys = (dec.options as { key: string }[]).map((o) => o.key); + expect(keys).toEqual(expect.arrayContaining(["select_entity", "select_tmdb", "provisional", "dismiss"])); + expect(dec.evidence).toMatchObject({ downloadTaskId: dt.id, outcome: "no_match" }); + expect((dec.evidence as { parsed: { year: number } }).parsed.year).toBe(2019); + + const hook = organizeDecisionHook(h.d); + expect(hook.match(dec)).toBe(true); + // 选择作品(本地)但没给作品 → 错误,事项保持打开 + await expect(hook.execute(dec, { optionKey: "select_entity" }, actor)).rejects.toThrow(/mediaEntityId/); + const out = await hook.execute(dec, { optionKey: "provisional" }, actor); + expect(out).toMatchObject({ kind: "organize", status: "running" }); + const ent = await h.db + .selectFrom("media_entity") + .selectAll() + .where("id", "=", out.mediaEntityId as string) + .executeTakeFirstOrThrow(); + expect(ent).toMatchObject({ status: "provisional", year: 2019 }); + expect(h.engine.enqueued.some((e) => (e.input as { recordId?: string }).recordId === out.recordId)).toBe(true); + const done = await executeRecord(h.d, out.recordId as string); + expect(done.status).toBe("succeeded"); + const closed = (await decisionOf(`organize_identity:${dt.id}`))!; + expect(closed.status).toBe("resolved"); + expect(closed.resolution).toMatchObject({ key: "manual_organize", organizationRecordId: out.recordId }); + const row = await h.db.selectFrom("download_task").selectAll().where("id", "=", dt.id).executeTakeFirstOrThrow(); + expect(row.media_entity_id).toBe(ent.id); + }); + + it("待处理中心 API:选候选 / 本地作品 → 执行并关闭;缺作品 → 400 且事项保持打开", async () => { + const target = await seedEntity(h.db, { kind: "movie", title: "Amber Coast", year: 2017 }); + const dt = await adoptCompleted("Qwzz.Nomatch.2017.1080p.WEB", "Qwzz.Nomatch.2017.1080p.WEB.mkv"); + await planImport(h.d, { downloadTaskId: dt.id, trigger: "auto", actorType: "system" }); + const dec = (await decisionOf(`organize_identity:${dt.id}`))!; + const bad = await h.api("POST", `/decisions/${dec.id}/resolve`, { optionKey: "select_entity" }); + expect(bad.status).toBe(400); + expect((await decisionOf(`organize_identity:${dt.id}`))?.status).toBe("open"); + const ok = await h.api("POST", `/decisions/${dec.id}/resolve`, { + optionKey: "select_entity", + identity: { mediaEntityId: target }, + }); + expect(ok.status).toBe(200); + expect(ok.body.data.status).toBe("resolved"); + expect(ok.body.data.resolution.execution).toMatchObject({ kind: "organize", mediaEntityId: target }); + const row = await h.db.selectFrom("download_task").selectAll().where("id", "=", dt.id).executeTakeFirstOrThrow(); + expect(row.media_entity_id).toBe(target); + const recId = ok.body.data.resolution.execution.recordId as string; + expect((await executeRecord(h.d, recId)).status).toBe("succeeded"); + }); + + it("识别失败后在文件管理按路径手动整理:同一下载任务的整理事项关闭,回填作品", async () => { + const other = await seedEntity(h.db, { kind: "movie", title: "Glass Orchard", year: 2018 }); + const dt = await adoptCompleted("Xqqz.Unmatched.2018.720p.HDTV", "Xqqz.Unmatched.2018.720p.HDTV.mkv"); + await planImport(h.d, { downloadTaskId: dt.id, trigger: "auto", actorType: "system" }); + expect((await decisionOf(`organize_identity:${dt.id}`))?.status).toBe("open"); + const src = join(cache.root, "Xqqz.Unmatched.2018.720p.HDTV"); + const plan = await planImport(h.d, { + sourcePaths: [src], + mediaEntityId: other, + trigger: "manual", + actorType: "user", + actorId: "tester", + }); + expect(plan.status).toBe("running"); + expect((await executeRecord(h.d, plan.recordId!)).status).toBe("succeeded"); + const closed = (await decisionOf(`organize_identity:${dt.id}`))!; + expect(closed).toMatchObject({ status: "resolved" }); + expect(closed.resolution).toMatchObject({ key: "manual_organize" }); + const row = await h.db.selectFrom("download_task").selectAll().where("id", "=", dt.id).executeTakeFirstOrThrow(); + expect(row.media_entity_id).toBe(other); + }); + + it("目标已存在:事项给 替换 / 保留两者 / 跳过;保留两者 → 新文件加序号入库,旧文件不动,事项闭环", async () => { + // 同一作品的另一份(不同内容、同名)→ 目标文件已存在 + const dt = await adoptCompleted( + "Quiet.Harbor.2021.1080p.BluRay.x264-GRP", + "Quiet.Harbor.2021.1080p.BluRay.x264-GRP.mkv", + SIZE + 1000, + ); + // 同名种子在 qB 里是新 hash;接管识别仍命中同一作品 + const plan = await planImport(h.d, { downloadTaskId: dt.id, trigger: "auto", actorType: "system" }); + expect(plan.status).toBe("failed"); + expect(plan.reason).toContain("target_exists"); + const dec = (await decisionOf(`organize_blocked:${dt.id}`))!; + expect((dec.options as { key: string }[]).map((o) => o.key)).toEqual(["replace", "keep_both", "skip"]); + const before = walkFiles(lib.root).filter((p) => p.endsWith(".mkv")); + expect(before.filter((p) => p.includes("Quiet"))).toHaveLength(1); + const hook = organizeDecisionHook(h.d); + const out = await hook.execute(dec, { optionKey: "keep_both" }, actor); + expect(out).toMatchObject({ kind: "organize", choice: "keep_both", status: "running" }); + expect((await executeRecord(h.d, out.recordId as string)).status).toBe("succeeded"); + const after = walkFiles(lib.root).filter((p) => p.endsWith(".mkv")); + expect(after).toHaveLength(before.length + 1); + const renamed = after.find((p) => p.includes(" (2).")); + expect(renamed && statSync(renamed).size).toBe(SIZE + 1000); + for (const p of before) expect(existsSync(p)).toBe(true); + expect((await decisionOf(`organize_blocked:${dt.id}`))?.status).toBe("resolved"); + }); + + it("目标已存在 → 跳过:其余照常,已存在的不整理;替换:核验后替换", async () => { + const dt = await adoptCompleted( + "Quiet.Harbor.2021.1080p.BluRay.x264-GRP", + "Quiet.Harbor.2021.1080p.BluRay.x264-GRP.mkv", + SIZE + 2000, + ); + await planImport(h.d, { downloadTaskId: dt.id, trigger: "auto", actorType: "system" }); + const dec = (await decisionOf(`organize_blocked:${dt.id}`))!; + const hook = organizeDecisionHook(h.d); + const skip = await hook.execute(dec, { optionKey: "skip" }, actor); + const rec = await h.db + .selectFrom("organization_record") + .selectAll() + .where("id", "=", skip.recordId as string) + .executeTakeFirstOrThrow(); + expect(rec.file_skipped).toBeGreaterThanOrEqual(1); + + const dt2 = await adoptCompleted( + "Quiet.Harbor.2021.1080p.BluRay.x264-GRP", + "Quiet.Harbor.2021.1080p.BluRay.x264-GRP.mkv", + SIZE + 3000, + ); + await planImport(h.d, { downloadTaskId: dt2.id, trigger: "auto", actorType: "system" }); + const dec2 = (await decisionOf(`organize_blocked:${dt2.id}`))!; + const rep = await hook.execute(dec2, { optionKey: "replace" }, actor); + expect(rep.status).toBe("running"); + expect((await executeRecord(h.d, rep.recordId as string)).status).toBe("succeeded"); + // 被替换的旧文件进库根回收站(.umt-trash),不在库内 + const inLib = walkFiles(lib.root).filter( + (p) => p.endsWith(".mkv") && p.includes("Quiet") && !p.includes("/.umt-trash/"), + ); + const sizes = inLib.map((p) => statSync(p).size); + expect(sizes).toContain(SIZE + 3000); + expect(sizes).not.toContain(SIZE); + expect(walkFiles(lib.root).some((p) => p.includes("/.umt-trash/") && statSync(p).size === SIZE)).toBe(true); + }); + + it("作品详情下载历史:列出接管的下载任务与整理记录", async () => { + const r = await h.api("GET", `/library/${movieId}`); + expect(r.status).toBe(200); + const ids = r.body.data.downloads.map((d: { downloadTask: { id: string } }) => d.downloadTask.id); + expect(ids).toContain(firstDtId); + const first = r.body.data.downloads.find((d: { downloadTask: { id: string } }) => d.downloadTask.id === firstDtId); + expect(first.downloadTask).toMatchObject({ origin: "adopted", mediaEntityId: movieId }); + expect(first.lastOrganizationStatus).toBe("succeeded"); + expect(r.body.data.organizationRecords.length).toBeGreaterThanOrEqual(3); + const prov = await h.db + .selectFrom("download_task") + .select("media_entity_id") + .where("id", "=", unknownDtId) + .executeTakeFirstOrThrow(); + const r2 = await h.api("GET", `/library/${prov.media_entity_id}`); + expect(r2.body.data.downloads.map((d: { downloadTask: { id: string } }) => d.downloadTask.id)).toContain( + unknownDtId, + ); + }); +}); diff --git a/apps/server/src/modules/acquisition-organize/index.ts b/apps/server/src/modules/acquisition-organize/index.ts index 7ec51daf..8313eb33 100644 --- a/apps/server/src/modules/acquisition-organize/index.ts +++ b/apps/server/src/modules/acquisition-organize/index.ts @@ -1,5 +1,6 @@ import { type EventPayload, EventType } from "@umt/contracts"; import { sql } from "@umt/db"; +import { registerDecisionHook } from "../decision-hooks.ts"; import { registerScheduledJob } from "../schedules/registry.ts"; import type { ServerContext, ServerModule } from "../types.ts"; import { registerAoCapabilities } from "./capabilities.ts"; @@ -15,6 +16,7 @@ import { registerRuleRoutes } from "./routes/rules.ts"; import { type AoDeps, automationBlock, enqueueSweep, enqueueTracked, timeSlot } from "./services/common.ts"; import { SWEEP_SLOT_MS, sweepReservedPlans, syncDownloads } from "./services/downloads.ts"; import { sweepDueSlots, sweepMetadataRefresh, sweepSubscriptionSearch } from "./services/subscriptions.ts"; +import { organizeDecisionHook } from "./services/takeover.ts"; import { loadVolumes, usedPercent } from "./services/volumes.ts"; import { errMessage } from "./util.ts"; @@ -66,6 +68,8 @@ export function createAcquisitionOrganizeModule(overrides: AoOverrides = {}): Se }, registerRoutes(app, ctx) { const d = lazy(ctx); + // 待处理"处理":接管识别失败 / 目标已存在 两类整理事项在关闭前由本模块执行所选处理 + registerDecisionHook(ctx, organizeDecisionHook(d)); registerRuleRoutes(app, d); registerDownloadRoutes(app, d); registerOrganizationRoutes(app, d); diff --git a/apps/server/src/modules/acquisition-organize/ports.ts b/apps/server/src/modules/acquisition-organize/ports.ts index 3b94d883..b878e087 100644 --- a/apps/server/src/modules/acquisition-organize/ports.ts +++ b/apps/server/src/modules/acquisition-organize/ports.ts @@ -15,7 +15,16 @@ import { type TorrentSource, type VerifyMethod, } from "@umt/adapters"; +import type { ActorType } from "@umt/contracts"; import type { DownloaderInstanceRow, MediaServerInstanceRow, SiteRow, SourceReleaseRow } from "@umt/db"; +import type { ParsedRelease } from "@umt/domain"; +import { SourcesRuntime } from "../sources-identity/runtime.ts"; +import { + createEntityFromTmdbCandidate, + enqueueNewEntityRefresh, +} from "../sources-identity/service/entity-from-candidate.ts"; +import { metadataLanguages } from "../sources-identity/service/metadata-rich.ts"; +import { identifyParsed, type NameIdentifyResult } from "../sources-identity/service/resolve.ts"; import type { ServerContext } from "../types.ts"; // 外部依赖端口:默认实现接真实 qB / 站点 / 文件系统 / Emby;测试注入 fake(mock qB、临时目录、固定磁盘用量)。 @@ -58,9 +67,21 @@ export interface AoPorts { diskUsage(path: string): Promise; probe(path: string): Promise; mediaServer(inst: MediaServerInstanceRow): Promise; + /** + * QB-1 接管任务按名称识别作品(sources-identity 的 ID-2 打分;trigger=takeover 允许查 TMDB)。 + * auto 时已绑定 / 新建作品;其余返回候选。 + */ + identify(input: { parsed: ParsedRelease; evidence?: Record }): Promise; + /** 按 TMDB id 找或建本地作品(待处理"按 TMDB 指定"),返回作品 id */ + entityFromTmdb( + ref: { type: "movie" | "tv"; id: string }, + actor: { type: string; id: string | null }, + ): Promise; options: AoOptions; } +export type { NameIdentifyResult }; + export const DEFAULT_OPTIONS: AoOptions = { backgroundLoops: true, noProgressMs: 6 * 3600_000, @@ -90,8 +111,31 @@ export function createPorts( } }; + let sources: SourcesRuntime | null = null; + const rt = () => { + sources ??= new SourcesRuntime(ctx); + return sources; + }; + const ports: AoPorts = { now: () => new Date(), + async identify({ parsed, evidence }) { + const r = await identifyParsed(rt(), parsed, { trigger: "takeover", ...(evidence ? { evidence } : {}) }); + if (r.created && r.mediaEntityId) await enqueueNewEntityRefresh(ctx, r.mediaEntityId); + return r; + }, + async entityFromTmdb(ref, actor) { + const type: ActorType = (["user", "agent", "system", "api_key"] as const).find((t) => t === actor.type) ?? "user"; + const r = await createEntityFromTmdbCandidate( + ctx.db, + () => rt().tmdb(), + { kind: ref.type, externalId: ref.id }, + { type, id: actor.id }, + { languages: await metadataLanguages(ctx) }, + ); + if (r.created) await enqueueNewEntityRefresh(ctx, r.mediaEntityId); + return r.mediaEntityId; + }, async downloader(inst) { const stamp = `${inst.updated_at.toISOString()}|${inst.base_url}`; const hit = clients.get(inst.id); diff --git a/apps/server/src/modules/acquisition-organize/services/common.ts b/apps/server/src/modules/acquisition-organize/services/common.ts index 3363c003..a676f45a 100644 --- a/apps/server/src/modules/acquisition-organize/services/common.ts +++ b/apps/server/src/modules/acquisition-organize/services/common.ts @@ -26,7 +26,7 @@ export interface DecisionInput { /** 同一 dedupeKey 同时只有一个未关闭条目;重复写入只累加 attempts、刷新证据 */ dedupeKey: string; evidence?: Record; - options?: { key: string; label: string; description?: string }[]; + options?: { key: string; label: string; description?: string; input?: Record }[]; taskId?: string | null; } @@ -38,6 +38,10 @@ export async function upsertDecision(d: AoDeps, exec: Exec, input: DecisionInput ${jsonb(input.evidence)}::jsonb, ${JSON.stringify(input.options ?? [])}::jsonb, ${input.taskId ?? null}) on conflict (dedupe_key) where status in ('open', 'investigating', 'awaiting_user') do update set attempts = decision.attempts + 1, evidence = excluded.evidence, summary = excluded.summary, + subject_type = coalesce(excluded.subject_type, decision.subject_type), + subject_id = coalesce(excluded.subject_id, decision.subject_id), + -- 重新判定时选项随最新证据刷新(如接管识别的候选);没给选项的保留原有(如 Agent 补的"重新校验") + options = case when jsonb_array_length(excluded.options) > 0 then excluded.options else decision.options end, updated_at = now() returning id, (xmax = 0) as inserted`.execute(exec); const row = r.rows[0]!; diff --git a/apps/server/src/modules/acquisition-organize/services/organize.ts b/apps/server/src/modules/acquisition-organize/services/organize.ts index 0f8483df..e2b776bc 100644 --- a/apps/server/src/modules/acquisition-organize/services/organize.ts +++ b/apps/server/src/modules/acquisition-organize/services/organize.ts @@ -46,6 +46,16 @@ import { PathGuardError } from "./organize-guard.ts"; import { type FrozenTransfer, finalizeTransfer, planTransfer, type TransferSpec } from "./organize-transfer.ts"; import { hnrSatisfied } from "./retention.ts"; import { slotComplete as sharedSlotComplete, subscriptionLibraryRoot } from "./subscriptions.ts"; +import { + CONFLICT_OPTIONS, + downloadTasksUnder, + identifyTakeover, + identityOptions, + keepBothTargets, + type QbFileLike, + type TakeoverIdentify, + takeoverSummary, +} from "./takeover.ts"; // 入库:冻结整理计划(organization_record + 文件明细)→ 逐文件可恢复执行(操作账本 fencing)→ // 资产/位置/路径绑定同事务登记 → Slot 推进 → Move 后移除 qB 任务记录(不删数据)。 @@ -70,6 +80,8 @@ export interface ImportRequest { onExisting?: ExistingPolicy | undefined; /** 已获替换授权(审批/手动确认) */ grantReplace?: boolean | undefined; + /** QB-1 目标已存在时"保留两者":新文件(连同附属文件)改名加序号放在旁边,已有文件不动 */ + keepBoth?: boolean | undefined; /** * W5-A2 逐文件覆写(文件管理整理向导的人工确认,优先于文件名解析):键为源文件或目录的绝对路径 * (目录对其下文件生效,最长前缀优先)。season/episodes 决定剧集季目录与命名以及集单元覆盖; @@ -312,6 +324,7 @@ export async function planImport(d: AoDeps, req: ImportRequest): Promise ({ name: f.path, size: f.size })); } if (list.length === 0) return fail("file_list_unavailable"); + qbFiles = list; sourceRoot = dt.content_path ? hostPathOf(topo, dt.content_path) : null; for (const f of list) { if (f.priority === 0) continue; @@ -389,16 +403,41 @@ export async function planImport(d: AoDeps, req: ImportRequest): Promise c.kind === "hnr_unsatisfied" || c.kind === "hnr_unknown"); + const existsConflict = blocking.some((c) => c.kind === "target_exists"); await upsertDecision(d, trx, { kind: hnrConflict ? "approval_required" : "path_conflict", title: hnrConflict ? `做种义务未满足，已阻止移动：${entity.title}` : `整理冲突：${entity.title}`, @@ -718,13 +766,15 @@ export async function planImport(d: AoDeps, req: ImportRequest): Promise f.source_path)].filter((x): x is string => !!x)), + ]; + for (const t of await downloadTasksUnder(trx, srcs)) { + await closeOrganizeDecisions(d, trx, t.id, rec); + await trx + .updateTable("download_task") + .set({ media_entity_id: frozen.entityId }) + .where("id", "=", t.id) + .where("media_entity_id", "is", null) + .execute(); + } } // 手动整理指定的 Slot(文件管理整理向导):核验通过后以新资产满足该 Slot(获取中的 Slot 不动,由其计划推进) if (!dt && frozen.slotId && okAny) { @@ -1654,7 +1726,7 @@ export async function finalizeRecord(d: AoDeps, recordId: string): Promise best.size) best = f; + } + return best?.name ?? null; +} + +/** + * 种子名 + 主视频文件名 → 解析结果。主视频路径(qB 文件名含种子根目录)按 文件 → 上级 → 上上级 目录合并; + * 种子名作为再上一级兜底(补标题 / 年份 / 季);文件名解析不出标题时直接用种子名。 + */ +export function takeoverParsed(torrentName: string, mainFile: string | null): ParsedRelease { + const byName = parseRelease(torrentName, { isFile: fileKindOf(torrentName) !== undefined }); + if (!mainFile) return byName; + const byPath = parseReleasePath(mainFile); + if (!byPath.title || byPath.auxiliaryOnly) return byName; + // 多文件种子:根目录就是种子名,parseReleasePath 已合并过;单文件种子文件名≈种子名,再合并一次无害 + return mergeParsed(byPath, byName, "parent_dir"); +} + +export interface TakeoverIdentify { + entityId: string | null; + mainFile: string | null; + parsed: ParsedRelease; + result: NameIdentifyResult; +} + +/** 识别接管任务;识别到作品时回填 download_task.media_entity_id */ +export async function identifyTakeover( + d: AoDeps, + dt: DownloadTaskRow, + files: readonly QbFileLike[], +): Promise { + const mainFile = pickMainVideo(files); + const parsed = takeoverParsed(dt.name, mainFile); + const result = await d.ports.identify({ + parsed, + evidence: { downloadTaskId: dt.id, torrentName: dt.name, mainFile }, + }); + const entityId = result.outcome === "auto" ? result.mediaEntityId : null; + if (entityId) + await d.ctx.db + .updateTable("download_task") + .set({ media_entity_id: entityId }) + .where("id", "=", dt.id) + .where("media_entity_id", "is", null) + .execute(); + return { entityId, mainFile, parsed, result }; +} + +export interface DecisionOptionInput { + key: string; + label: string; + description?: string; + input?: Record; +} + +const OUTCOME_TEXT: Record = { + auto: "已识别", + proposed: "识别到候选，但自动化未开启，需要确认", + ambiguous: "有多个相近的作品，无法自动确定", + no_match: "本地与 TMDB 都没有匹配的作品", + lookup_failed: "查询 TMDB 失败", + unparseable: "文件名解析不出标题", +}; + +export function takeoverSummary(t: TakeoverIdentify | null): string { + if (!t) return "下载内容未关联作品。识别或指定作品后再整理"; + const what = [t.parsed.title, t.parsed.year].filter(Boolean).join(" "); + return `按文件名识别${what ? `「${what}」` : ""}：${OUTCOME_TEXT[t.result.outcome]}。选择作品、作为临时作品入库，或忽略`; +} + +/** 接管识别失败 / 歧义的事项选项:候选(最多 3 个)、选择作品(本地 / TMDB)、作为临时作品入库、忽略 */ +export function identityOptions(t: TakeoverIdentify | null, fallbackTitle: string): DecisionOptionInput[] { + const out: DecisionOptionInput[] = []; + for (const [i, c] of (t?.result.candidates ?? []).slice(0, 3).entries()) { + const target = c.localEntityId + ? { mediaEntityId: c.localEntityId } + : c.provider === "tmdb" && (c.entityType === "movie" || c.entityType === "tv") + ? { tmdb: { type: c.entityType, id: c.externalId } } + : null; + if (!target) continue; + out.push({ + key: `candidate_${i}`, + label: `选择：${c.title}${c.year ? `（${c.year}）` : ""}`, + description: c.localEntityId ? "本地已有作品" : `TMDB ${c.entityType}/${c.externalId}，本地没有时自动建档`, + input: { candidate: target }, + }); + } + out.push( + { + key: "select_entity", + label: "选择作品（本地）", + description: "在本地作品里搜索并指定", + input: { requires: "mediaEntityId" }, + }, + { + key: "select_tmdb", + label: "选择作品（TMDB）", + description: "按 TMDB ID 指定，本地没有时自动建档", + input: { requires: "tmdb" }, + }, + ); + const title = t?.parsed.title ?? fallbackTitle; + if (title) { + const tvLike = + t?.parsed.mediaTypeHint === "tv" || t?.parsed.season !== undefined || t?.parsed.episodes !== undefined; + out.push({ + key: "provisional", + label: "作为临时作品入库", + description: "按文件名建本地临时作品后整理，之后可合并或补外部身份", + input: { provisional: { kind: tvLike ? "series" : "movie", title, year: t?.parsed.year ?? null } }, + }); + } + out.push({ key: "dismiss", label: "忽略" }); + return out; +} + +/** 目标已存在(target_exists)冲突的事项选项 */ +export const CONFLICT_OPTIONS: DecisionOptionInput[] = [ + { key: "replace", label: "替换", description: "新文件核验通过后替换已有文件，旧文件进回收站" }, + { key: "keep_both", label: "保留两者", description: "新文件改名（加序号）后放在旁边，已有文件不动" }, + { key: "skip", label: "跳过", description: "已存在的文件不整理，其余文件照常入库" }, +]; + +// ── 保留两者:把目标已存在的主视频(连同附属文件)改名为 "<名称> (n)" ── + +function withSuffix(p: string, stem: string, n: number): string { + const dir = dirname(p); + const name = basename(p); + if (name.startsWith(stem)) return join(dir, `${stem} (${n})${name.slice(stem.length)}`); + const ext = extname(name); + return join(dir, `${name.slice(0, name.length - ext.length)} (${n})${ext}`); +} + +export async function keepBothTargets(plan: OrganizePlan): Promise { + const held = plan.conflicts.filter((c) => c.kind === "target_exists" && c.blocking && c.src); + if (held.length === 0) return; + const mains = new Set(); + for (const c of held) { + const f = plan.files.find((x) => x.src === c.src); + if (f) mains.add(f.role === "main" ? f.src : (f.mainSrc ?? f.src)); + } + const taken = new Set(plan.files.map((f) => f.dst)); + const resolved = new Set(); + for (const m of mains) { + const group = plan.files.filter((f) => f.src === m || f.mainSrc === m); + const main = group.find((f) => f.src === m) ?? group[0]!; + const ext = extname(main.dst); + const stem = basename(main.dst).slice(0, basename(main.dst).length - ext.length); + for (let n = 2; n < 100; n++) { + const alts = group.map((f) => withSuffix(f.dst, stem, n)); + let free = true; + for (const a of alts) { + if (taken.has(a) || (await statSafe(a)).kind !== "missing") { + free = false; + break; + } + } + if (!free) continue; + group.forEach((f, i) => { + taken.add(alts[i]!); + f.dst = alts[i]!; + if (f.action === "hold") f.action = "transfer"; + resolved.add(f.src); + }); + break; + } + } + plan.conflicts = plan.conflicts.filter((c) => !(c.kind === "target_exists" && c.src && resolved.has(c.src))); + for (const pc of plan.preconditions) + if (pc.kind === "target_absent") pc.paths = plan.files.filter((f) => f.action === "transfer").map((f) => f.dst); + plan.requiresApproval = plan.conflicts.some((c) => c.blocking); +} + +// ── 手动整理(待处理里选了处理方式):冻结计划 + 入队执行 ── + +export async function startManualImport( + d: AoDeps, + req: Omit, + actor: DecisionHookActor, + via: string, +): Promise<{ recordId: string | null; status: string; reason: string | null; taskId: string }> { + const db = d.ctx.db; + const idem = `${via}:${req.downloadTaskId ?? "x"}:${randomUUID()}`; + const by = actor.id ? `${actor.type}:${actor.id}` : actor.type; + const taskId = await ensureTaskRow(db, { + kind: "verifyAndImport", + idempotencyKey: idem, + input: { ...req, via }, + subjectType: req.downloadTaskId ? "download_task" : "organize_request", + subjectId: req.downloadTaskId ?? null, + triggeredBy: by, + }); + const actorType = actor.type === "api_key" || actor.type === "agent" || actor.type === "system" ? actor.type : "user"; + const plan = await planImport(d, { + ...req, + trigger: "manual", + actorType, + actorId: by, + taskId, + }); + if (plan.status === "running" || plan.status === "queued") { + const { runId } = await d.ctx.engine.enqueue( + "verifyAndImport", + { recordId: plan.recordId, downloadTaskId: req.downloadTaskId, taskId }, + { idempotencyKey: `verifyAndImport:${idem}`, traceId: taskId }, + ); + await db.updateTable("task").set({ engine_run_id: runId }).where("id", "=", taskId).execute(); + } else { + await markTask(db, taskId, "failed", { error: plan.reason ?? plan.status }); + } + return { recordId: plan.recordId, status: plan.status, reason: plan.reason, taskId }; +} + +// ── 待处理执行钩子:organize_identity / organize_blocked ── + +class HookError extends Error { + constructor( + readonly code: "bad_request" | "conflict" | "not_found", + message: string, + ) { + super(message); + } +} + +export { HookError as OrganizeDecisionError }; + +async function createProvisional( + exec: Exec, + p: { kind: "movie" | "series"; title: string; year: number | null }, +): Promise { + const e = await exec + .insertInto("media_entity") + .values({ + kind: p.kind, + title: p.title, + original_title: null, + year: p.year, + status: "provisional", + metadata: jsonb({}), + }) + .returning(["id"]) + .executeTakeFirstOrThrow(); + await addAliasIfAbsent(exec, e.id, p.title, "other", "user"); + return e.id; +} + +async function chosenEntity( + d: AoDeps, + body: DecisionResolveBody, + opt: DecisionOptionInput | undefined, + actor: DecisionHookActor, +): Promise { + const input = (opt?.input ?? {}) as { + requires?: string; + candidate?: { mediaEntityId?: string; tmdb?: { type: "movie" | "tv"; id: string } }; + provisional?: { kind: "movie" | "series"; title: string; year: number | null }; + }; + const pick = + input.requires === "mediaEntityId" + ? { mediaEntityId: body.identity?.mediaEntityId } + : input.requires === "tmdb" + ? { tmdb: body.identity?.tmdb } + : (input.candidate ?? null); + if (input.provisional) return createProvisional(d.ctx.db, input.provisional); + if (pick?.mediaEntityId) { + const e = await d.ctx.db + .selectFrom("media_entity") + .select(["id", "status", "merged_into_id"]) + .where("id", "=", pick.mediaEntityId) + .executeTakeFirst(); + if (!e) throw new HookError("not_found", "作品不存在"); + return e.status === "merged" && e.merged_into_id ? e.merged_into_id : e.id; + } + if (pick?.tmdb) return d.ports.entityFromTmdb(pick.tmdb, actor); + throw new HookError("bad_request", input.requires ? `需要 identity.${input.requires}` : "该选项没有指定作品"); +} + +export function organizeDecisionHook(d: AoDeps): DecisionHook { + return { + match: (row) => row.dedupe_key.startsWith("organize_identity:") || row.dedupe_key.startsWith("organize_blocked:"), + async execute(row, body, actor) { + const dismissing = body.dismiss === true || body.optionKey === "dismiss"; + if (dismissing) return { kind: "none", reason: "dismissed" }; + const options = (row.options ?? []) as DecisionOptionInput[]; + const opt = options.find((o) => o.key === body.optionKey); + const ev = (row.evidence ?? {}) as { downloadTaskId?: string | null }; + const dtId = ev.downloadTaskId ?? row.dedupe_key.split(":")[1] ?? null; + const dt = dtId + ? await d.ctx.db.selectFrom("download_task").selectAll().where("id", "=", dtId).executeTakeFirst() + : undefined; + if (row.dedupe_key.startsWith("organize_identity:")) { + if (!dt) throw new HookError("not_found", "关联的下载任务不存在"); + const entityId = await chosenEntity(d, body, opt, actor); + await d.ctx.db + .updateTable("download_task") + .set({ media_entity_id: entityId }) + .where("id", "=", dt.id) + .execute(); + const r = await startManualImport(d, { downloadTaskId: dt.id, mediaEntityId: entityId }, actor, "decision"); + return { kind: "organize", mediaEntityId: entityId, ...r }; + } + // organize_blocked:目标已存在 → 替换 / 保留两者 / 跳过 + const key = body.optionKey; + if (key !== "replace" && key !== "keep_both" && key !== "skip") return { kind: "none", reason: `option_${key}` }; + if (!dt) throw new HookError("not_found", "关联的下载任务不存在"); + const rec = await d.ctx.db + .selectFrom("organization_record") + .select(["id", "media_entity_id", "target_library_root_id"]) + .where("download_task_id", "=", dt.id) + .orderBy("created_at", "desc") + .executeTakeFirst(); + const r = await startManualImport( + d, + { + downloadTaskId: dt.id, + ...(rec?.media_entity_id ? { mediaEntityId: rec.media_entity_id } : {}), + ...(rec?.target_library_root_id ? { targetLibraryRootId: rec.target_library_root_id } : {}), + // 操作方式不沿用原记录:仍按策略 / 设置决定(Move 的授权与做种义务照常把关) + ...(key === "replace" + ? { onExisting: "replace_with_verify" as const, grantReplace: true } + : key === "skip" + ? { onExisting: "skip" as const } + : { keepBoth: true }), + }, + actor, + `decision-${key}`, + ); + if (r.status !== "running" && r.status !== "queued") + throw new HookError("conflict", `重新整理仍未能执行：${r.reason ?? r.status}`); + return { kind: "organize", choice: key, ...r }; + }, + }; +} + +/** 已登记下载任务里,内容路径(本机视图)与这些源路径重叠的(手动按路径整理时找回对应下载任务) */ +export async function downloadTasksUnder(exec: Exec, paths: readonly string[]): Promise { + if (paths.length === 0) return []; + const topos = await loadInstances(exec, { enabledOnly: false }); + const rows = await exec + .selectFrom("download_task") + .selectAll() + .where("removed_at", "is", null) + .where("content_path", "is not", null) + .execute(); + const out: DownloadTaskRow[] = []; + for (const r of rows) { + const topo = topos.find((t) => t.inst.id === r.instance_id); + const host = topo && r.content_path ? hostPathOf(topo, r.content_path) : null; + if (!host) continue; + const root = host.replace(/\/+$/, ""); + if (paths.some((p) => p === root || p.startsWith(`${root}/`) || root.startsWith(`${p.replace(/\/+$/, "")}/`))) + out.push(r); + } + return out; +} diff --git a/apps/server/src/modules/decision-hooks.ts b/apps/server/src/modules/decision-hooks.ts new file mode 100644 index 00000000..abab4e05 --- /dev/null +++ b/apps/server/src/modules/decision-hooks.ts @@ -0,0 +1,38 @@ +import type { DecisionRow } from "@umt/db"; +import type { ServerContext } from "./types.ts"; + +// 待处理"处理"端点(sources-identity routes/decisions.ts)的执行钩子:其它模块按 dedupe_key 认领自己的事项, +// 在事项关闭前同步执行所选处理(失败抛错 → 事项保持打开,错误回给用户)。按 ctx 隔离(测试里多个装配互不影响)。 + +export interface DecisionResolveBody { + optionKey?: string | undefined; + dismiss?: boolean | undefined; + note?: string | undefined; + identity?: + | { mediaEntityId?: string | undefined; tmdb?: { type: "movie" | "tv"; id: string } | undefined } + | undefined; +} + +export interface DecisionHookActor { + type: string; + id: string | null; +} + +export interface DecisionHook { + /** 认领该事项(通常按 dedupe_key 前缀) */ + match(d: DecisionRow): boolean; + /** 执行所选处理;返回写进 resolution.execution 的摘要 */ + execute(d: DecisionRow, body: DecisionResolveBody, actor: DecisionHookActor): Promise>; +} + +const hooks = new WeakMap(); + +export function registerDecisionHook(ctx: ServerContext, hook: DecisionHook): void { + const list = hooks.get(ctx) ?? []; + list.push(hook); + hooks.set(ctx, list); +} + +export function decisionHookFor(ctx: ServerContext, d: DecisionRow): DecisionHook | undefined { + return hooks.get(ctx)?.find((h) => h.match(d)); +} diff --git a/apps/server/src/modules/sources-identity/routes/decisions.ts b/apps/server/src/modules/sources-identity/routes/decisions.ts index 273ba903..91538f88 100644 --- a/apps/server/src/modules/sources-identity/routes/decisions.ts +++ b/apps/server/src/modules/sources-identity/routes/decisions.ts @@ -3,6 +3,7 @@ import { type Decision, DecisionKind, DecisionStatus, Id, type PoolCandidate } f import { sql } from "@umt/db"; import { z } from "zod"; import { ApiException } from "../../../http/errors.ts"; +import { decisionHookFor } from "../../decision-hooks.ts"; import type { ModuleApp } from "../../types.ts"; import { D, DecisionMinSeverity } from "../defs.ts"; import { route } from "../http.ts"; @@ -223,6 +224,19 @@ export function decisionRoutes(app: ModuleApp, rt: () => SourcesRuntime) { { idempotencyKey: `refreshSiteAccount:decision:${d.id}:${Date.now()}`, traceId: randomUUID() }, ); } + // 其它模块认领的事项(如整理:接管识别 / 目标已存在):关闭前同步执行,失败则事项保持打开 + let execution: Record | null = null; + const hook = decisionHookFor(rt().ctx, d); + if (hook) { + try { + execution = await hook.execute(d, body, { type: actor.type, id: actor.id ?? null }); + } catch (e) { + const code = (e as { code?: string }).code; + if (code === "bad_request" || code === "conflict" || code === "not_found") + throw new ApiException(code, (e as Error).message); + throw e; + } + } // 上面的执行路径已按 dedupeKey 关闭;其余类型在这里记录处理结果 const row = await db() .transaction() @@ -239,7 +253,12 @@ export function decisionRoutes(app: ModuleApp, rt: () => SourcesRuntime) { .updateTable("decision") .set({ status, - resolution: jsonb({ optionKey: body.optionKey ?? null, note: body.note ?? null, by }), + resolution: jsonb({ + optionKey: body.optionKey ?? null, + note: body.note ?? null, + by, + ...(execution ? { execution } : {}), + }), resolved_by: by, resolved_at: new Date(), }) diff --git a/apps/server/src/modules/sources-identity/service/resolve.ts b/apps/server/src/modules/sources-identity/service/resolve.ts index d68377e0..0f4ea625 100644 --- a/apps/server/src/modules/sources-identity/service/resolve.ts +++ b/apps/server/src/modules/sources-identity/service/resolve.ts @@ -65,7 +65,11 @@ export interface ResolveInput { trigger?: ResolveTrigger | undefined; } -export type ResolveTrigger = "observed" | "review" | "manual" | "search"; +/** + * takeover = 接管的下载任务(用户手动加进 qB、没有站点资源)完成后按种子名 / 主视频文件名识别(QB-1,对照 MP transfer + * 无下载历史时按文件名 MetaInfoPath 识别):视同用户动作,允许查 TMDB。 + */ +export type ResolveTrigger = "observed" | "review" | "manual" | "search" | "takeover"; export interface ResolveResult { outcome: @@ -118,10 +122,10 @@ function poolCandidate(r: RankedCandidate): PoolCandidate { async function localCandidates( exec: Exec, parsed: ParsedRelease, - release: SourceReleaseRow, + release: SourceReleaseRow | null, ): Promise { const ids = new Set(); - const explicit = [...parsed.explicitIds, ...siteExplicitIds(release.raw)]; + const explicit = [...parsed.explicitIds, ...(release ? siteExplicitIds(release.raw) : [])]; for (const x of explicit) { const row = await exec .selectFrom("external_identity") @@ -408,7 +412,14 @@ export async function resolveRelease(rt: SourcesRuntime, input: ResolveInput): P } /** 为什么要查 TMDB;null = 不查(惰性识别) */ -export type LookupReason = "tracked_alias" | "manual" | "search" | "previously_looked_up" | "eager" | "new_title_rule"; +export type LookupReason = + | "tracked_alias" + | "manual" + | "search" + | "takeover" + | "previously_looked_up" + | "eager" + | "new_title_rule"; async function lookupReason( rt: SourcesRuntime, @@ -419,6 +430,7 @@ async function lookupReason( ): Promise { if (trigger === "manual") return "manual"; if (trigger === "search") return "search"; + if (trigger === "takeover") return "takeover"; if (tracked) return "tracked_alias"; // 池条目此前已查过 TMDB(旧行为 / 手动 / 规则触发):复查延续,仍受识别缓存与有界退避约束 if (trigger === "review" && poolReason && poolReason !== "unmatched") return "previously_looked_up"; @@ -820,3 +832,170 @@ async function toPool( } export { recordAttempt }; + +// ── 按名称识别(没有站点资源行):接管的下载任务完成后用 ── + +export interface NameIdentifyCandidate { + provider: string; + entityType: string; + externalId: string; + title: string; + year: number | null; + localEntityId: string | null; + level: string; +} + +export interface NameIdentifyResult { + outcome: "auto" | "proposed" | "ambiguous" | "no_match" | "lookup_failed" | "unparseable"; + mediaEntityId: string | null; + contentUnitIds: string[]; + reason: string; + /** 候选(歧义 / 待确认时给用户挑;最多 5 个) */ + candidates: NameIdentifyCandidate[]; + /** 本次是否新建了作品 */ + created: boolean; +} + +/** + * 按已解析的名称识别作品(ID-2 同一套打分:本地召回 + TMDB 候选 → resolveIdentity)。不写待识别池、不开待处理, + * 由调用方决定失败后怎么办。auto 且自动化开启时绑定 / 新建作品并确保季集单元;其余只返回候选。 + */ +export async function identifyParsed( + rt: SourcesRuntime, + parsed: ParsedRelease, + opts: { trigger?: ResolveTrigger; evidence?: Record } = {}, +): Promise { + const { ctx } = rt; + const none = { mediaEntityId: null, contentUnitIds: [] as string[], candidates: [], created: false }; + if (!parsed.title || parsed.auxiliaryOnly) return { outcome: "unparseable", reason: "unparseable_title", ...none }; + const trigger = opts.trigger ?? "takeover"; + const local = await localCandidates(ctx.db, parsed, null); + const lookup = await lookupReason(rt, parsed, trigger, local.length > 0, null); + let remote: MetadataCandidate[] = []; + let lookupError: string | null = null; + let recognize: { cacheHit: boolean; meta: RecognizeMeta | null } = { cacheHit: false, meta: null }; + if (lookup) { + try { + const r = await tmdbCandidates(rt, parsed, true); + remote = r.remote; + recognize = r; + } catch (e) { + if (!(e instanceof TmdbError)) throw e; + lookupError = `${e.kind}: ${e.message}`; + } + } + const resolution = resolveIdentity(parsed, [...local, ...remote.map(tmdbToCandidate)]); + if (recognize.meta && !recognize.cacheHit && !lookupError) { + const top = resolution.status === "auto" ? resolution.best?.candidate : undefined; + const hit = + top?.provider === "tmdb" + ? remote.find((m) => m.externalId === top.externalId && m.entityType === top.entityType) + : undefined; + const info = hit ?? (resolution.status === "no_match" ? null : undefined); + await saveRecognizeCache(rt, recognize.meta, info); + } + const candidates = resolution.ranked.slice(0, 5).map((r) => { + const c = poolCandidate(r); + return { + provider: c.provider, + entityType: c.entityType, + externalId: c.externalId, + title: c.title, + year: c.year, + localEntityId: c.localEntityId, + level: String(c.level), + }; + }); + const automation = await ctx.settings.get("automation"); + const autoAccept = automation.enabled && !automation.readOnlyReconcile; + if (resolution.status === "auto" && resolution.best && autoAccept) { + const best = resolution.best; + const level: EvidenceLevel = toContractEvidenceLevel(best.level) ?? "needs_review"; + let detail = + best.candidate.provider === "tmdb" + ? remote.find((m) => m.externalId === best.candidate.externalId && m.entityType === best.candidate.entityType) + : undefined; + if (detail && !detail.detailed) { + const existing = await entityByExternal(ctx.db, { + provider: "tmdb", + entityType: detail.entityType, + externalId: detail.externalId, + }); + if (!existing) { + try { + detail = await (await rt.tmdb())!.details(detail.entityType, detail.externalId); + } catch (e) { + if (!(e instanceof TmdbError)) throw e; + } + } + } + for (let attempt = 0; ; attempt++) { + try { + const bound = await ctx.db.transaction().execute(async (trx) => { + let entityId: string | null = null; + let created = false; + if (best.candidate.localEntityId) entityId = await resolveEntityId(trx, best.candidate.localEntityId); + if (!entityId && best.candidate.provider === "tmdb") { + entityId = await entityByExternal(trx, { + provider: "tmdb", + entityType: best.candidate.entityType, + externalId: best.candidate.externalId, + }); + if (!entityId && detail) { + const r = await createEntityFromTmdb(trx, detail, { + level, + evidence: { + ...(opts.evidence ?? {}), + trigger, + rules: best.evidence, + titleMatch: best.titleMatch, + yearMatch: best.yearMatch, + }, + actor: SYSTEM, + variety: looksLikeVariety(parsed, null), + }); + entityId = r.entityId; + created = r.created; + } + } + if (!entityId) return null; + const entity = await trx + .selectFrom("media_entity") + .select(["id", "kind"]) + .where("id", "=", entityId) + .executeTakeFirstOrThrow(); + const units = await ensureUnits(trx, entityId, entity.kind, parsed); + return { entityId, created, unitIds: units.unitIds }; + }); + if (!bound) break; + return { + outcome: "auto", + mediaEntityId: bound.entityId, + contentUnitIds: bound.unitIds, + reason: resolution.reason, + candidates, + created: bound.created, + }; + } catch (e) { + if (e instanceof IdentityConflictError && attempt < 2) continue; + throw e; + } + } + } + const outcome: NameIdentifyResult["outcome"] = + resolution.status === "auto" + ? "proposed" + : resolution.status === "ambiguous" + ? "ambiguous" + : lookupError + ? "lookup_failed" + : "no_match"; + return { + outcome, + mediaEntityId: null, + contentUnitIds: [], + reason: lookupError ?? (resolution.status === "auto" && !autoAccept ? "automation_off" : resolution.reason), + candidates, + created: false, + }; +} diff --git a/apps/server/test/sources-identity/lazy-identity.test.ts b/apps/server/test/sources-identity/lazy-identity.test.ts index 3941c8c4..3d279281 100644 --- a/apps/server/test/sources-identity/lazy-identity.test.ts +++ b/apps/server/test/sources-identity/lazy-identity.test.ts @@ -1,9 +1,9 @@ import { sql } from "@umt/db"; -import { normalizeTitle } from "@umt/domain"; +import { normalizeTitle, parseReleasePath } from "@umt/domain"; import { createMockTmdb, type MockTmdb, TMDB_FIXTURES, type TmdbFixture } from "@umt/testkit"; import { afterAll, beforeAll, describe, expect, it } from "vitest"; import { down, up } from "../../../../packages/db/migrations/0047_identity_pool.ts"; -import { resolveRelease } from "../../src/modules/sources-identity/service/resolve.ts"; +import { identifyParsed, resolveRelease } from "../../src/modules/sources-identity/service/resolve.ts"; import { createHarness, type Harness, json, pgAvailable, seedSite } from "./harness.ts"; // DEC-1 惰性识别(对照 MP:RSS / 站点资源只在匹配订阅或用户搜索时识别,识别失败不告警)与待识别池治理。 @@ -37,6 +37,8 @@ describe.skipIf(!pg)("DEC-1 惰性识别 + 待识别池", () => { twin(112), twin(211, "Nightfall", "夜幕", "1995-03-01"), twin(212, "Nightfall", "夜幕", "1995-03-01"), + twin(311, "Harborlight", "港灯", "2003-05-01"), + twin(312, "Harborlight", "港灯", "2003-05-01"), ], }); h = await createHarness("lazyid", { tmdbUrl: tmdb.url }); @@ -271,6 +273,44 @@ describe.skipIf(!pg)("DEC-1 惰性识别 + 待识别池", () => { .executeTakeFirstOrThrow(); expect(p).toMatchObject({ status: "archived", next_review_at: null }); }); + + it("QB-1 接管任务按文件名识别(trigger=takeover):不需要本地别名也查 TMDB,auto 建档;歧义只返回候选,不进池不开待处理", async () => { + const calls = tmdbCalls(); + const parsed = parseReleasePath( + "Dune.Part.Two.2024.2160p.UHD.BluRay.x265-GRP/Dune.Part.Two.2024.2160p.UHD.BluRay.x265-GRP.mkv", + ); + const r = await identifyParsed(h.rt, parsed, { trigger: "takeover" }); + expect(tmdbCalls()).toBeGreaterThan(calls); + expect(r.outcome).toBe("auto"); + expect(r.mediaEntityId).toBeTruthy(); + const ent = await h.db + .selectFrom("media_entity") + .selectAll() + .where("id", "=", r.mediaEntityId!) + .executeTakeFirstOrThrow(); + expect(ent.year).toBe(2024); + // 同名两部 → 歧义:只给候选 + const pool = await h.db + .selectFrom("identity_candidate") + .select((eb) => eb.fn.countAll().as("n")) + .executeTakeFirstOrThrow(); + const amb = await identifyParsed(h.rt, parseReleasePath("Harborlight.2003.1080p.BluRay.mkv"), { + trigger: "takeover", + }); + expect(amb.outcome).toBe("ambiguous"); + expect(amb.candidates.map((c) => c.externalId).sort()).toEqual(["311", "312"]); + expect(amb.mediaEntityId).toBeNull(); + const pool2 = await h.db + .selectFrom("identity_candidate") + .select((eb) => eb.fn.countAll().as("n")) + .executeTakeFirstOrThrow(); + expect(pool2.n).toBe(pool.n); + // 默认 observed 触发:不相关资源不查 TMDB(惰性识别不变) + const before = tmdbCalls(); + const lazy = await identifyParsed(h.rt, parseReleasePath("Zqxv.Kxlmn.2011.720p.mkv"), { trigger: "observed" }); + expect(lazy.outcome).toBe("no_match"); + expect(tmdbCalls()).toBe(before); + }); }); describe.skipIf(!pg)("迁移 0047:存量 identity_unresolved 清理", () => { diff --git a/docs/runbook/files.md b/docs/runbook/files.md index bb788d17..793c0a13 100644 --- a/docs/runbook/files.md +++ b/docs/runbook/files.md @@ -79,3 +79,15 @@ - NFO/图片投影(scrape 的 `writeNfo/images`)、硬链接、蓝光原盘整目录操作仍未做。 - 搬运 move 在终态提交前删源侧空目录;若进程恰在两者之间退出,重跑 finalize 幂等,但极端情况下可能留下空目录(无数据风险)。 - 路径守卫未下沉到 `packages/adapters/src/fs/guard.ts`(本模块自带按根解析的守卫,整理仍用 `sourceGuard`)。 + +## 整理事项闭环(QB-1) + +- 整理成功(自动重试或手动,有资产入库)后,同一下载任务的 `organize_identity:`(无法确定作品)与 `organize_blocked:` + (整理冲突 / 做种义务)事项一并关闭,`resolution.key = manual_organize`(自动重试为 `organize_retry`),记整理记录 id。 +- 文件管理按路径手动整理(没有 `downloadTaskId`):按下载器路径映射找内容路径与源路径重叠的下载任务,同样关闭其整理事项, + 并把作品回填到 `download_task.media_entity_id`(作品详情「下载历史」据此列出)。 +- 目标已存在(`target_exists`)的整理冲突事项选项为「替换 / 保留两者 / 跳过」,处理时按原记录的作品与库根重新冻结计划并执行(操作方式仍按策略 / 设置,Move 的做种义务照常把关): + 替换 = `onExisting=replace_with_verify` + 授权(新文件核验后旧文件进库根回收站);保留两者 = 新主视频连同附属文件改名 + `<名称> (n)`(n 从 2 起取第一个不存在的);跳过 = `onExisting=skip`,其余文件照常入库。仍有其它阻塞冲突 → 409,事项保持打开。 +- 作品详情(`GET /library/{id}`):`downloads` 按 计划 Slot / 站点资源 / `download_task.media_entity_id` 归属(接管任务按识别结果、 + 手动指定或整理后回填);`organizationRecords` 列该作品最近 100 条整理记录(含没有下载任务的手动整理)。 diff --git a/docs/runbook/identity.md b/docs/runbook/identity.md index 235769db..7558d3d6 100644 --- a/docs/runbook/identity.md +++ b/docs/runbook/identity.md @@ -16,6 +16,7 @@ |---|---| | `manual` | 任务输入 `trigger: "manual"`:池里点「重新识别」、Agent / 能力 `identity.resolve` | | `search` | 任务输入 `trigger: "search"`:搜索结果 | + | `takeover` | 接管的下载任务按文件名识别(`identifyParsed(…, { trigger: "takeover" })`,见下文「接管任务按文件名识别」) | | `tracked_alias` | 本地召回有候选(标题命中订阅 / 本地作品的别名) | | `previously_looked_up` | 周期复查,且这条池条目以前查过 TMDB(旧数据、手动、规则触发) | | `eager` | 设置 `identity.eagerTmdbLookup = true` | @@ -38,6 +39,24 @@ 搜索模块想让搜索结果直接识别,入队 `resolveObservation` 时传 `trigger: "search"`(目前搜索结果经 `release.observed` 进来, 按 observed 处理;命中订阅别名的照样会查)。 +## 接管任务按文件名识别(QB-1) + +用户手动加进 qB、贴了管理标签的任务(`download_task.origin = adopted`,没有 `source_release` / 计划)下载完成后,`verifyAndImport` +的 `planImport` 先按 MP transfer 的做法识别作品(`acquisition-organize/services/takeover.ts`): + +1. 主视频 = 要下载(priority≠0)的视频文件里最大的一个(跳过 sample);`parseReleasePath(主视频路径)` 合并 文件 → 上级 → 上上级 + 目录名(qB 文件名含种子根目录,相当于 MP `MetaInfoPath`),再用种子名补缺(`mergeParsed`);文件名解析不出标题时直接用种子名。 +2. `identifyParsed`(`service/resolve.ts`):同一套 ID-2 打分(本地召回 + TMDB 候选 → `resolveIdentity`),`trigger: "takeover"` 视为 + 允许查 TMDB;不写待识别池、不开身份事项。auto 且自动化开启 → 关联 / 新建作品(新建的入队一次 full 刷新),回填 + `download_task.media_entity_id`,直接整理。 +3. 失败 / 歧义 / 自动化关闭:整理记录 `identity_unknown`,事项 `organize_identity:<下载任务 id>`(`identity_ambiguous`), + 证据带种子名、主视频、解析出的标题 / 年份 / 季集、识别结果与候选。选项:候选(最多 3 个,本地作品或 TMDB)、 + 「选择作品（本地）」(`requires: mediaEntityId`)、「选择作品（TMDB）」(`requires: tmdb`,本地没有时建档)、 + 「作为临时作品入库」(建 `provisional` 作品 + 别名)、「忽略」。 +4. 处理(`POST /decisions/{id}/resolve`):整理模块注册的执行钩子(`modules/decision-hooks.ts`)在关闭前同步执行 —— 回填 + `download_task.media_entity_id` → 冻结手动整理计划 → 入队执行;执行结果记在 `resolution.execution`。缺输入 / 作品不存在 + → 400 / 404,事项保持打开。 + ## 待处理只收相关的 | 结果 | 开事项? | From 50c8d534f8a0584c8f3eab30e25b0978124f5ee7 Mon Sep 17 00:00:00 2001 From: root Date: Tue, 29 Sep 2026 20:54:37 +0800 Subject: [PATCH 4/4] chore(deploy): NO_PROXY defaults for downloaders; explain library root modes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - compose: app gets a default NO_PROXY (UMT_NO_PROXY) listing compose service names incl. qbittorrent / qbittorrent-nox / qb / transmission / deluge / aria2, so a proxy override never swallows downloader traffic. - runbook: proxy and test-qB override templates use the same list. - Settings: one-line hint on library root "管理方式" (managed vs register_only). Co-Authored-By: Claude Opus 5.5 --- apps/web/app/components/settings/instances.tsx | 2 ++ docker-compose.yml | 4 ++++ docs/runbook/deploy.md | 7 +++++-- 3 files changed, 11 insertions(+), 2 deletions(-) diff --git a/apps/web/app/components/settings/instances.tsx b/apps/web/app/components/settings/instances.tsx index aef75f5c..2abdac95 100644 --- a/apps/web/app/components/settings/instances.tsx +++ b/apps/web/app/components/settings/instances.tsx @@ -347,6 +347,8 @@ export function LibraryRootsTable() { { value: "register_only", label: "只登记" }, { value: "managed", label: "自动维护（补齐、升级）" }, ], + hint: "自动维护：自动整理可把新下载放进来，文件管理可在其中改名、移动、删除；只登记：只扫描登记已有文件，不自动写入也不可改动。", + wide: true, }, { name: "enabled", label: "启用", kind: "switch" }, { name: "notes", label: "备注", kind: "textarea" }, diff --git a/docker-compose.yml b/docker-compose.yml index d5d0acc7..588e1bca 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -188,6 +188,10 @@ services: BROWSER_ENABLED: ${BROWSER_ENABLED:-true} # 原站视图桌面模式的上游(桌面镜像的 Selkies);缺省由 BROWSER_PROVIDER_URL 推出(端口换 3000) BROWSER_DESKTOP_URL: ${BROWSER_DESKTOP_URL:-} + # 出站代理(HTTP_PROXY/HTTPS_PROXY)在 override 里给(见 docs/runbook/deploy.md「容器经宿主代理」)。 + # Deno 按主机名匹配 NO_PROXY、不会因解析到内网 IP 而直连:同 compose 网络的服务名(postgres、camofox、 + # 下载器 qbittorrent 等)必须列在这里,否则连下载器也会绕到代理上失败。自有下载器服务名不同就改 UMT_NO_PROXY。 + NO_PROXY: ${UMT_NO_PROXY:-localhost,127.0.0.1,::1,postgres,camofox,proxy-relay,qbittorrent,qbittorrent-nox,qb,transmission,deluge,aria2,10.0.0.0/8,172.16.0.0/12,192.168.0.0/16} ports: - "${UMT_BIND:-0.0.0.0}:${UMT_PORT:-4380}:4380" volumes: diff --git a/docs/runbook/deploy.md b/docs/runbook/deploy.md index 03b2c90d..bbae529b 100644 --- a/docs/runbook/deploy.md +++ b/docs/runbook/deploy.md @@ -228,11 +228,14 @@ services: environment: HTTP_PROXY: http://172.17.0.1:7891 HTTPS_PROXY: http://172.17.0.1:7891 - NO_PROXY: localhost,127.0.0.1,postgres,camofox,proxy-relay,10.0.0.0/8,172.16.0.0/12,192.168.0.0/16 + # 同网络的服务名都要列上(含下载器:qbittorrent / transmission / deluge / aria2 等),与 compose 缺省 UMT_NO_PROXY 一致 + NO_PROXY: localhost,127.0.0.1,::1,postgres,camofox,proxy-relay,qbittorrent,qbittorrent-nox,qb,transmission,deluge,aria2,10.0.0.0/8,172.16.0.0/12,192.168.0.0/16 camofox: environment: { HTTP_PROXY: http://172.17.0.1:7891, HTTPS_PROXY: http://172.17.0.1:7891, NO_PROXY: localhost,127.0.0.1,app,10.0.0.0/8,172.16.0.0/12 } ``` +compose 已给 app 设了缺省 `NO_PROXY`(`UMT_NO_PROXY` 可改),override 里重写 `NO_PROXY` 时别漏掉下载器的服务名。 + 防火墙只放行 docker 网段:`ufw allow from 172.16.0.0/12 to 172.17.0.1 port 7891 proto tcp`。compose 默认网络已固定在 `UMT_SUBNET`(缺省 172.30.0.0/24,与本机其他网络冲突时在 `.env.deploy` 改成 172.16.0.0/12 内的空闲网段,如 172.16.80.0/24),否则 compose 自动分配的 192.168.x 网段不在放行范围内。Deno 的 fetch 会读 `HTTP(S)_PROXY`/`NO_PROXY`;TMDB、境外 Pi 上游、camofox 访问境外站都经此代理,站内(postgres/camofox)不走代理。 ## 测试用 qB(override 示例) @@ -244,7 +247,7 @@ services: app: environment: # 配了 HTTP(S)_PROXY 时必须把 qbittorrent 加进 NO_PROXY:Deno 按主机名匹配,不会因解析到内网 IP 而直连 - NO_PROXY: localhost,127.0.0.1,postgres,camofox,proxy-relay,qbittorrent,10.0.0.0/8,172.16.0.0/12,192.168.0.0/16 + NO_PROXY: localhost,127.0.0.1,::1,postgres,camofox,proxy-relay,qbittorrent,qbittorrent-nox,qb,transmission,deluge,aria2,10.0.0.0/8,172.16.0.0/12,192.168.0.0/16 qbittorrent: image: lscr.io/linuxserver/qbittorrent:5.2.4 # v4 / 5.0–5.2 都兼容(QB-1 实测 5.1.2、5.2.4) restart: unless-stopped