/** * qBittorrent WebUI API v2 客户端。 * 迁移来源:MoviePilot app/modules/qbittorrent/qbittorrent.py(e6820a8),逐函数对照见 * docs/porting/moviepilot-downloader-and-sites.md「W1-3a 实际迁移对照」。 * 叠加:多实例、错误分类(DownloaderError)、提交幂等(本地 infohash 预查 + 收单超时按 hash 回查)。 */ import { randomBytes } from "node:crypto"; import { DownloaderError, type DownloaderErrorKind } from "./errors.ts"; import { assertSafeToDeleteData, type SafeDeleteCheck } from "./safe-delete.ts"; import { isDownloadComplete, mapQbState, type QbObservedState } from "./state.ts"; import { BencodeError, parseMagnet, parseTorrent } from "./torrent.ts"; export interface QbInstanceConfig { /** 本系统内的下载器实例 ID(downloader_instance.id) */ instanceId: string; /** 例:http://127.0.0.1:8080(不带 /api/v2) */ baseUrl: string; username?: string; password?: string; /** 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;写请求不自动重试 */ retries?: number; /** 重试退避基数,指数增长,默认 300ms */ retryBaseMs?: number; /** 提交后按 infohash/探针标签回查的次数与间隔,默认 9 次 × 3000ms(对照 MP get_torrent_id_by_tag;首查不等待) */ lookup?: { attempts: number; intervalMs: number }; /** 探针标签前缀,默认 uhd-probe- */ probeTagPrefix?: string; /** 实例级默认:顺序下载 / 首尾块优先(对照 MP sequentail / first_last_piece 配置) */ sequentialDownload?: boolean; firstLastPiecePrio?: boolean; /** 注入 fetch(测试/代理) */ 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; infohashV1: string | null; infohashV2: string | null; name: string; rawState: string; state: QbObservedState; /** 严格完成判定结果(progress>=1 && amountLeft<=0 && 终态) */ complete: boolean; progress: number; amountLeft: number; size: number; totalSize: number | null; savePath: string; contentPath: string | null; category: string; tags: string[]; addedOn: number | null; completionOn: number | null; dlspeed: number; upspeed: number; eta: number | null; ratio: number | null; seedingTimeSec: number | null; /** 已连接的做种 / 下载用户数(num_seeds / num_leechs) */ numSeeds: number; numLeechs: number; /** Tracker 报告的做种 / 下载总数(num_complete / num_incomplete;未知为 null) */ numComplete: number | null; numIncomplete: number | null; } export interface QbFile { index: number; name: string; size: number; progress: number; /** 0 不下载,1 普通,6 高,7 最高 */ priority: number; isSeed: boolean | null; } export type TorrentSource = | { kind: "magnet"; uri: string } | { kind: "url"; url: string } | { kind: "file"; bytes: Uint8Array; filename?: string }; export interface AddTorrentRequest { source: TorrentSource; savePath: string; category?: string; tags?: string[]; paused?: boolean; rename?: string; contentLayout?: "Original" | "Subfolder" | "NoSubfolder"; sequential?: boolean; firstLastPiecePrio?: boolean; } export interface AddTorrentResult { accepted: boolean; addedIds: string[]; raw: string; } export type SubmitOutcome = | { kind: "added"; hash: string; via: "confirmed" | "added_ids" | "probe_tag" | "recovered_after_timeout" } | { kind: "already_exists"; hash: string; existing: QbTorrent } /** 下载器已接收(或可能已接收),但回查窗口内未找到任务;调用方稍后按 hash/probeTag 对账,不得盲目重提 */ | { kind: "accepted_unconfirmed"; hash: string | null; probeTag: string | null } | { kind: "rejected"; reason: string; hash: string | null } /** 确认未提交成功(回查未命中且无歧义)或下载器不可达;有 hash 时用同一 hash 重试是幂等的 */ | { kind: "unreachable"; error: DownloaderError; hash: string | null }; export interface TorrentQuery { hashes?: string[]; tag?: string; /** 须同时具备的标签(对照 MP get_torrents 的 tags 子集判断);第一个走 qB 原生 tag 参数 */ tags?: string[]; category?: string; filter?: | "all" | "downloading" | "seeding" | "completed" | "paused" | "stopped" | "active" | "inactive" | "stalled" | "errored"; } export interface QbMainData { rid: number; fullUpdate: boolean; torrents: Record>; torrentsRemoved: string[]; serverState: Record | null; } /** 下载器端口(实例维度)。上层 tasks/agent 只依赖此接口。 */ export interface DownloaderClient { readonly instanceId: string; version(): Promise<{ app: string; webApi: string }>; add(req: AddTorrentRequest): Promise; submit(req: AddTorrentRequest): Promise; list(q?: TorrentQuery): Promise; /** 严格完成判定后的已完成任务(对照 get_completed_torrents) */ getCompleted(q?: Pick): Promise; getDownloading(q?: Pick): Promise; findByHash(hash: string): Promise; files(hash: string, opts?: { retries?: number; intervalMs?: number }): Promise; /** priority 0 不下载;fileIds 为空返回 false 且不发请求(对照 set_files) */ setFilePriority(hash: string, fileIds: number[], priority: 0 | 1 | 6 | 7): Promise; setForceStart(hashes: string[], enable?: boolean): Promise; recheck(hashes: string[]): Promise; setLocation(hashes: string[], location: string): Promise; setCategory(hashes: string[], category: string): Promise; changeTorrent(hash: string, opts: TorrentLimits): Promise; /** 只读下载器全局偏好(不改写,见 QA6) */ preferences(): Promise>; contentLayout(): Promise<"Original" | "Subfolder" | "NoSubfolder">; transferInfo(): Promise>; pause(hashes: string[]): Promise; resume(hashes: string[]): Promise; addTags(hashes: string[], tags: string[]): Promise; removeTags(hashes: string[], tags: string[]): Promise; deleteTagDefinitions(tags: string[]): Promise; /** 从任务移除标签并删除全局定义;hashes 为 null 时只删定义(对照 delete_torrents_tag) */ deleteTorrentsTag(hashes: string[] | null, tags: string[]): Promise; /** * deleteFiles 必须显式给出。deleteFiles=true 时每个 hash 都要在 safety 里给出检查输入, * 任一未通过 assertSafeToDeleteData 即抛 DownloaderError('unsafe_delete'),请求不会发出。 */ delete(hashes: string[], opts: DeleteOptions): Promise; mainData(rid?: number): Promise; } /** 删除选项:删数据必须附带逐个 hash 的安全检查输入(键为 hash,大小写不敏感) */ export type DeleteOptions = | { deleteFiles: false } | { deleteFiles: true; safety: Readonly> }; export interface TorrentLimits { /** KB/s,0 不限速 */ uploadLimitKBps?: number; downloadLimitKBps?: number; ratioLimit?: number; /** 分钟 */ seedingTimeLimitMin?: number; } type Body = { kind: "form"; data: Record } | { kind: "multipart"; data: FormData }; interface ReqOpts { query?: Record; body?: Body; idempotent: boolean; } const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)); function num(v: unknown): number | null { return typeof v === "number" && Number.isFinite(v) ? v : null; } /** qB 用 -1 表示未知(num_complete 等) */ function nonNeg(v: unknown): number | null { const n = num(v); return n !== null && n >= 0 ? n : null; } 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; const amountLeft = num(r.amount_left) ?? Number.POSITIVE_INFINITY; const tags = typeof r.tags === "string" ? r.tags .split(",") .map((t) => t.trim()) .filter(Boolean) : []; return { instanceId, hash: String(r.hash ?? "").toLowerCase(), infohashV1: str(r.infohash_v1), infohashV2: str(r.infohash_v2), name: String(r.name ?? ""), rawState, state: mapQbState(rawState), complete: isDownloadComplete({ progress, amountLeft, state: rawState }), progress, amountLeft, size: num(r.size) ?? 0, totalSize: num(r.total_size), savePath: String(r.save_path ?? ""), contentPath: str(r.content_path), category: String(r.category ?? ""), tags, addedOn: num(r.added_on), completionOn: (num(r.completion_on) ?? -1) > 0 ? num(r.completion_on) : null, dlspeed: num(r.dlspeed) ?? 0, upspeed: num(r.upspeed) ?? 0, eta: num(r.eta), ratio: num(r.ratio), seedingTimeSec: num(r.seeding_time), numSeeds: num(r.num_seeds) ?? 0, numLeechs: num(r.num_leechs) ?? 0, numComplete: nonNeg(r.num_complete), numIncomplete: nonNeg(r.num_incomplete), }; } export class QbClient implements DownloaderClient { readonly instanceId: string; private readonly base: string; private readonly timeoutMs: number; private readonly retries: number; private readonly retryBaseMs: number; private readonly lookup: { attempts: number; intervalMs: number }; private readonly probePrefix: string; private readonly fetchImpl: typeof fetch; private sid: string | null = null; private loginInFlight: Promise | null = null; /** 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; this.retryBaseMs = cfg.retryBaseMs ?? 300; this.lookup = cfg.lookup ?? { attempts: 9, intervalMs: 3000 }; this.probePrefix = cfg.probeTagPrefix ?? "uhd-probe-"; this.fetchImpl = cfg.fetch ?? fetch; } private err(kind: DownloaderErrorKind, message: string, endpoint: string, status?: number, cause?: unknown) { return new DownloaderError(kind, `[qb:${this.instanceId}] ${message}`, { instanceId: this.instanceId, endpoint, ...(status !== undefined ? { status } : {}), ...(cause !== undefined ? { cause } : {}), }); } private async rawFetch(endpoint: string, init: RequestInit): Promise { try { return await this.fetchImpl(`${this.base}/api/v2/${endpoint}`, { ...init, signal: AbortSignal.timeout(this.timeoutMs), }); } catch (e) { const name = (e as { name?: string })?.name; if (name === "TimeoutError" || name === "AbortError") { throw this.err("timeout", `request timed out after ${this.timeoutMs}ms`, endpoint, undefined, e); } throw this.err("network", `request failed: ${(e as Error)?.message ?? String(e)}`, endpoint, undefined, e); } } /** 登录;并发调用合并为一次。 */ async login(): Promise { if (this.authMode !== "password") return; if (!this.loginInFlight) { this.loginInFlight = this.doLogin().finally(() => { this.loginInFlight = null; }); } 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, { method: "POST", 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()).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 { const attempts = opts.idempotent ? this.retries + 1 : 1; let lastErr: DownloaderError | null = null; for (let i = 0; i < attempts; i++) { if (i > 0) await sleep(this.retryBaseMs * 2 ** (i - 1)); try { return await this.sendOnce(endpoint, opts, false); } catch (e) { if (!(e instanceof DownloaderError)) throw e; lastErr = e; const retryable = e.kind === "network" || e.kind === "timeout" || e.kind === "rate" || e.kind === "protocol"; if (!retryable) throw e; } } throw lastErr!; } private async sendOnce(endpoint: string, opts: ReqOpts, relogged: boolean): Promise { const session = this.authMode === "password"; if (session && this.sid === null) await this.login(); const headers: Record = { referer: this.base }; 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"; body = new URLSearchParams(opts.body.data); } else if (opts.body?.kind === "multipart") { body = opts.body.data; } const qs = new URLSearchParams(); for (const [k, v] of Object.entries(opts.query ?? {})) if (v !== undefined) qs.set(k, v); const path = qs.size ? `${endpoint}?${qs}` : endpoint; const res = await this.rawFetch(path, { method: body ? "POST" : "GET", headers, ...(body ? { body } : {}) }); const s = res.status; if (s === 401 || s === 403) { 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); } 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(); throw this.err("notfound", t || "not found", endpoint, s); } if (s === 429) { await res.body?.cancel(); throw this.err("rate", "rate limited", endpoint, s); } if (s === 400 || s === 409 || s === 415) { const t = await res.text(); throw this.err("rejected", t || `HTTP ${s}`, endpoint, s); } if (!res.ok) { await res.body?.cancel(); throw this.err("protocol", `unexpected HTTP ${s}`, endpoint, s); } return res; } private async json(endpoint: string, query?: Record): Promise { const res = await this.send(endpoint, { idempotent: true, ...(query ? { query } : {}) }); const text = await res.text(); try { return JSON.parse(text) as T; } catch (e) { throw this.err("protocol", "invalid JSON response", endpoint, res.status, e); } } private async post(endpoint: string, data: Record, idempotent = true): Promise { const res = await this.send(endpoint, { idempotent, body: { kind: "form", data } }); return res.text(); } async version(): Promise<{ app: string; webApi: string }> { const app = await (await this.send("app/version", { idempotent: true })).text(); const webApi = await (await this.send("app/webapiVersion", { idempotent: true })).text(); return { app: app.trim(), webApi: webApi.trim() }; } async add(req: AddTorrentRequest): Promise { const form = new FormData(); const src = req.source; if (src.kind === "file") { form.append( "torrents", new Blob([new Uint8Array(src.bytes)], { type: "application/x-bittorrent" }), src.filename ?? "upload.torrent", ); } else { form.append("urls", src.kind === "magnet" ? src.uri : src.url); } form.append("savepath", req.savePath); if (req.category !== undefined) form.append("category", req.category); if (req.tags?.length) form.append("tags", req.tags.join(",")); const paused = req.paused ? "true" : "false"; form.append("paused", paused); // <= 4.x form.append("stopped", paused); // >= 5.x if (req.rename) form.append("rename", req.rename); form.append("contentLayout", req.contentLayout ?? "Original"); form.append("autoTMM", "false"); const seq = req.sequential ?? this.cfg.sequentialDownload; const flp = req.firstLastPiecePrio ?? this.cfg.firstLastPiecePrio; if (seq !== undefined) form.append("sequentialDownload", String(seq)); if (flp !== undefined) form.append("firstLastPiecePrio", String(flp)); const res = await this.send("torrents/add", { idempotent: false, body: { kind: "multipart", data: form } }); const raw = (await res.text()).trim(); if (raw.startsWith("{")) { try { const j = JSON.parse(raw) as { success_count?: number; pending_count?: number; added_torrent_ids?: string[]; }; const ids = (j.added_torrent_ids ?? []).map((x) => x.toLowerCase()); return { accepted: ids.length > 0 || (j.success_count ?? 0) + (j.pending_count ?? 0) > 0, addedIds: ids, raw }; } catch { // 落到文本判定 } } return { accepted: raw.startsWith("Ok"), addedIds: [], raw }; } async list(q: TorrentQuery = {}): Promise { const need = [...(q.tag !== undefined ? [q.tag] : []), ...(q.tags ?? [])]; const rows = await this.json[]>("torrents/info", { filter: q.filter, category: q.category, tag: need[0], hashes: q.hashes?.length ? q.hashes.map((h) => h.toLowerCase()).join("|") : undefined, }); if (!Array.isArray(rows)) throw this.err("protocol", "torrents/info did not return an array", "torrents/info"); const list = rows.map((r) => normalizeTorrent(this.instanceId, r)); return need.length > 1 ? list.filter((t) => need.every((tag) => t.tags.includes(tag))) : list; } async getCompleted(q: Pick = {}): Promise { return (await this.list({ ...q, filter: "completed" })).filter((t) => t.complete); } async getDownloading(q: Pick = {}): Promise { return this.list({ ...q, filter: "downloading" }); } /** 按 hash 精确查询,不受管理标签限制。查不到返回 null;下载器不可达抛错(不能当作不存在)。 */ async findByHash(hash: string): Promise { const h = hash.toLowerCase(); const rows = await this.list({ hashes: [h] }); return rows.find((t) => t.hash === h || t.infohashV1 === h || t.infohashV2?.startsWith(h)) ?? null; } /** 文件列表;空列表(磁力元数据未就绪)按 retries/intervalMs 重试(对照 get_files)。 */ async files(hash: string, opts: { retries?: number; intervalMs?: number } = {}): Promise { const tries = Math.max(1, opts.retries ?? 1); let rows: Record[] = []; for (let i = 0; i < tries; i++) { if (i > 0 && opts.intervalMs) await sleep(opts.intervalMs); rows = await this.json[]>("torrents/files", { hash: hash.toLowerCase() }); if (rows.length) break; } return rows.map((r, i) => ({ index: num(r.index) ?? i, name: String(r.name ?? ""), size: num(r.size) ?? 0, progress: num(r.progress) ?? 0, priority: num(r.priority) ?? 1, isSeed: typeof r.is_seed === "boolean" ? r.is_seed : null, })); } private async pauseLike(hashes: string[], legacy: string, modern: string): Promise { if (!hashes.length) return; const data = { hashes: hashes.join("|") }; if (this.legacyPauseApi !== false) { try { await this.post(legacy, data); this.legacyPauseApi = true; return; } catch (e) { if (!(e instanceof DownloaderError && e.kind === "notfound" && this.legacyPauseApi === null)) throw e; this.legacyPauseApi = false; } } await this.post(modern, data); } pause(hashes: string[]): Promise { return this.pauseLike(hashes, "torrents/pause", "torrents/stop"); } resume(hashes: string[]): Promise { return this.pauseLike(hashes, "torrents/resume", "torrents/start"); } async addTags(hashes: string[], tags: string[]): Promise { await this.post("torrents/addTags", { hashes: hashes.join("|"), tags: tags.join(",") }); } async removeTags(hashes: string[], tags: string[]): Promise { await this.post("torrents/removeTags", { hashes: hashes.join("|"), tags: tags.join(",") }); } async deleteTagDefinitions(tags: string[]): Promise { await this.post("torrents/deleteTags", { tags: tags.join(",") }); } async deleteTorrentsTag(hashes: string[] | null, tags: string[]): Promise { if (hashes?.length) await this.removeTags(hashes, tags); await this.deleteTagDefinitions(tags); } async setFilePriority(hash: string, fileIds: number[], priority: 0 | 1 | 6 | 7): Promise { if (!hash || !fileIds.length) return false; await this.post("torrents/filePrio", { hash: hash.toLowerCase(), id: fileIds.join("|"), priority: String(priority), }); return true; } async setForceStart(hashes: string[], enable = true): Promise { if (!hashes.length) return; await this.post("torrents/setForceStart", { hashes: hashes.join("|"), value: String(enable) }); } async recheck(hashes: string[]): Promise { if (!hashes.length) return; await this.post("torrents/recheck", { hashes: hashes.join("|") }); } async setLocation(hashes: string[], location: string): Promise { if (!hashes.length || !location) return; await this.post("torrents/setLocation", { hashes: hashes.join("|"), location }); } async setCategory(hashes: string[], category: string): Promise { if (!hashes.length) return; await this.post("torrents/setCategory", { hashes: hashes.join("|"), category }); } /** 对照 change_torrent:限速单位 KB/s → B/s;只给 seedingTime 时 ratioLimit=-2(沿用全局)。 */ async changeTorrent(hash: string, o: TorrentLimits): Promise { const h = hash.toLowerCase(); if (o.uploadLimitKBps !== undefined) { await this.post("torrents/setUploadLimit", { hashes: h, limit: String(Math.trunc(o.uploadLimitKBps * 1024)) }); } if (o.downloadLimitKBps !== undefined) { await this.post("torrents/setDownloadLimit", { hashes: h, limit: String(Math.trunc(o.downloadLimitKBps * 1024)), }); } if (o.ratioLimit !== undefined || o.seedingTimeLimitMin !== undefined) { await this.post("torrents/setShareLimits", { hashes: h, ratioLimit: o.ratioLimit !== undefined ? String(Math.round(o.ratioLimit * 100) / 100) : "-2", seedingTimeLimit: String(Math.trunc(o.seedingTimeLimitMin ?? -1)), inactiveSeedingTimeLimit: "-1", }); } } async preferences(): Promise> { return this.json>("app/preferences"); } async contentLayout(): Promise<"Original" | "Subfolder" | "NoSubfolder"> { const v = (await this.preferences()).torrent_content_layout; return v === "Subfolder" || v === "NoSubfolder" ? v : "Original"; } async transferInfo(): Promise> { return this.json>("transfer/info"); } async delete(hashes: string[], opts: DeleteOptions): Promise { if (typeof opts?.deleteFiles !== "boolean") throw new TypeError("deleteFiles must be explicit"); if (!hashes.length) return; if (opts.deleteFiles) { const safety = new Map(Object.entries(opts.safety ?? {}).map(([h, c]) => [h.toLowerCase(), c])); for (const h of hashes) { const check = safety.get(h.toLowerCase()); if (!check) { throw this.err("unsafe_delete", `拒绝删除数据: ${h} 缺少安全检查输入`, "torrents/delete"); } assertSafeToDeleteData(check, this.instanceId); } } await this.post("torrents/delete", { hashes: hashes.join("|"), deleteFiles: String(opts.deleteFiles) }); } async mainData(rid = 0): Promise { const j = await this.json>("sync/maindata", { rid: String(rid) }); return { rid: num(j.rid) ?? 0, fullUpdate: j.full_update === true, torrents: (j.torrents as Record>) ?? {}, torrentsRemoved: (j.torrents_removed as string[]) ?? [], serverState: (j.server_state as Record) ?? null, }; } private async pollHash(hash: string): Promise { for (let i = 0; i < this.lookup.attempts; i++) { if (i > 0) await sleep(this.lookup.intervalMs); try { const t = await this.findByHash(hash); if (t) return t; } catch (e) { if (!(e instanceof DownloaderError) || e.kind === "auth") throw e; } } return null; } private async pollTag(tag: string): Promise { for (let i = 0; i < this.lookup.attempts; i++) { if (i > 0) await sleep(this.lookup.intervalMs); try { const rows = await this.list({ tag }); if (rows[0]) return rows[0]; } catch (e) { if (!(e instanceof DownloaderError) || e.kind === "auth") throw e; } } return null; } private async dropProbe(tag: string, hash: string | null): Promise { try { if (hash) await this.removeTags([hash], [tag]); await this.deleteTagDefinitions([tag]); } catch { // 残留探针标签由对账任务按前缀清理 } } /** * 幂等提交:本地算 infohash → 预查是否已存在 → add → 按 hash(或探针标签)确认。 * add 超时/断连时不重提,先回查(REQ §8.4)。 */ async submit(req: AddTorrentRequest): Promise { let hash: string | null = null; const src = req.source; if (src.kind === "file") { try { hash = parseTorrent(src.bytes).hash; } catch (e) { if (e instanceof BencodeError) return { kind: "rejected", reason: `invalid torrent: ${e.message}`, hash: null }; throw e; } } else if (src.kind === "magnet") { const m = parseMagnet(src.uri); if (!m) return { kind: "rejected", reason: "invalid magnet uri", hash: null }; hash = m.hash; } else { hash = parseMagnet(src.url)?.hash ?? null; } if (hash) { try { const existing = await this.findByHash(hash); if (existing) return { kind: "already_exists", hash: existing.hash, existing }; } catch (e) { if (e instanceof DownloaderError) return { kind: "unreachable", error: e, hash }; throw e; } } const probeTag = hash ? null : `${this.probePrefix}${Date.now().toString(36)}${randomBytes(5).toString("hex")}`; const tags = probeTag ? [...(req.tags ?? []), probeTag] : req.tags; let result: AddTorrentResult; try { result = await this.add({ ...req, ...(tags ? { tags } : {}) }); } catch (e) { if (!(e instanceof DownloaderError)) throw e; if (e.ambiguous) { if (hash) { const t = await this.pollHash(hash).catch(() => null); if (t) return { kind: "added", hash: t.hash, via: "recovered_after_timeout" }; return { kind: "unreachable", error: e, hash }; } const t = await this.pollTag(probeTag!).catch(() => null); if (t) { await this.dropProbe(probeTag!, t.hash); return { kind: "added", hash: t.hash, via: "recovered_after_timeout" }; } return { kind: "accepted_unconfirmed", hash: null, probeTag }; } if (e.kind === "rejected") { if (hash) { const t = await this.findByHash(hash).catch(() => null); if (t) return { kind: "already_exists", hash: t.hash, existing: t }; } if (probeTag) await this.dropProbe(probeTag, null); return { kind: "rejected", reason: e.message, hash }; } return { kind: "unreachable", error: e, hash }; } if (!result.accepted) { // 4.x:已存在/非法链接都返回 Fails.;先查是否已存在 if (hash) { const t = await this.findByHash(hash).catch(() => null); if (t) return { kind: "already_exists", hash: t.hash, existing: t }; } if (probeTag) await this.dropProbe(probeTag, null); return { kind: "rejected", reason: `downloader refused: ${result.raw.slice(0, 200)}`, hash }; } const knownHash = hash ?? result.addedIds[0] ?? null; if (knownHash) { const t = await this.pollHash(knownHash).catch(() => null); if (probeTag) await this.dropProbe(probeTag, t?.hash ?? knownHash); if (t) return { kind: "added", hash: t.hash, via: hash ? "confirmed" : "added_ids" }; return { kind: "accepted_unconfirmed", hash: knownHash, probeTag: null }; } const t = await this.pollTag(probeTag!).catch(() => null); if (t) { await this.dropProbe(probeTag!, t.hash); return { kind: "added", hash: t.hash, via: "probe_tag" }; } return { kind: "accepted_unconfirmed", hash: null, probeTag }; } } export function createQbClient(cfg: QbInstanceConfig): DownloaderClient & { login(): Promise } { return new QbClient(cfg); } /** 多实例注册表:按 instanceId 取客户端。 */ export class DownloaderRegistry { private readonly clients = new Map(); constructor(configs: QbInstanceConfig[] = []) { for (const c of configs) this.register(c); } register(cfg: QbInstanceConfig): DownloaderClient { const c = new QbClient(cfg); this.clients.set(cfg.instanceId, c); return c; } get(instanceId: string): DownloaderClient | undefined { return this.clients.get(instanceId); } remove(instanceId: string): boolean { return this.clients.delete(instanceId); } list(): DownloaderClient[] { return [...this.clients.values()]; } }