Merge branch 'qb1'

Co-Authored-By: Claude Fable 5.1 <[email protected]>
This commit is contained in:
rootandClaude Fable 5.1 committed 2026-09-29 21:05:25 +08:00
commit 8e47d8715b
36 files changed
+2053 -89

No files matched your search

@@ -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<void>;
}
export async function createHarness(name: string, overrides: AoOverrides = {}): Promise<Harness> {
export async function createHarness(
name: string,
overrides: AoOverrides = {},
/** withSources:同时装配 sources-identity 模块(待处理中心路由等) */
opts: { withSources?: boolean } = {},
): Promise<Harness> {
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) {
@@ -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,
);
});
});
@@ -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}`);
@@ -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);
@@ -9,12 +9,22 @@ import {
getSiteAdapter,
getSiteDefinition,
type ProbeResult,
qbAuthConfig,
type SecretResolver,
SiteError,
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、临时目录、固定磁盘用量)。
@@ -57,9 +67,21 @@ export interface AoPorts {
diskUsage(path: string): Promise<DiskUsage>;
probe(path: string): Promise<ProbeResult>;
mediaServer(inst: MediaServerInstanceRow): Promise<EmbyClient>;
/**
* QB-1 接管任务按名称识别作品(sources-identity 的 ID-2 打分;trigger=takeover 允许查 TMDB)。
* auto 时已绑定 / 新建作品;其余返回候选。
*/
identify(input: { parsed: ParsedRelease; evidence?: Record<string, unknown> }): Promise<NameIdentifyResult>;
/** 按 TMDB id 找或建本地作品(待处理"按 TMDB 指定"),返回作品 id */
entityFromTmdb(
ref: { type: "movie" | "tv"; id: string },
actor: { type: string; id: string | null },
): Promise<string>;
options: AoOptions;
}
export type { NameIdentifyResult };
export const DEFAULT_OPTIONS: AoOptions = {
backgroundLoops: true,
noProgressMs: 6 * 3600_000,
@@ -89,20 +111,41 @@ 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);
if (hit && hit.stamp === stamp) return hit.client;
const settings = (inst.settings ?? {}) as Record<string, unknown>;
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;
@@ -149,7 +149,7 @@ export function registerLibraryRoutes(app: ModuleApp, d: AoDeps) {
.select(
sql<number>`(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<string | null>`e.metadata #>> '{posterPath,value}'`.as("poster_path"))
@@ -207,7 +207,7 @@ export function registerLibraryRoutes(app: ModuleApp, d: AoDeps) {
if (query.filter === "downloading")
q = q.where(
sql<boolean>`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) => ({
@@ -290,10 +290,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()
@@ -341,6 +367,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;
},
@@ -394,8 +433,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<string>`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<string>`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(),
@@ -26,7 +26,7 @@ export interface DecisionInput {
/** 同一 dedupeKey 同时只有一个未关闭条目;重复写入只累加 attempts、刷新证据 */
dedupeKey: string;
evidence?: Record<string, unknown>;
options?: { key: string; label: string; description?: string }[];
options?: { key: string; label: string; description?: string; input?: Record<string, unknown> }[];
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]!;
@@ -313,15 +313,18 @@ export async function submitDownload(d: AoDeps, planId: string, opts: { runId?:
}
const r = res as Extract<SubmitResult, { hash: string }>;
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,
@@ -47,6 +47,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 任务记录(不删数据)。
@@ -71,6 +81,8 @@ export interface ImportRequest {
onExisting?: ExistingPolicy | undefined;
/** 已获替换授权(审批/手动确认) */
grantReplace?: boolean | undefined;
/** QB-1 目标已存在时"保留两者":新文件(连同附属文件)改名加序号放在旁边,已有文件不动 */
keepBoth?: boolean | undefined;
/**
* W5-A2 逐文件覆写(文件管理整理向导的人工确认,优先于文件名解析):键为源文件或目录的绝对路径
* (目录对其下文件生效,最长前缀优先)。season/episodes 决定剧集季目录与命名以及集单元覆盖;
@@ -313,6 +325,7 @@ export async function planImport(d: AoDeps, req: ImportRequest): Promise<PlanRes
const files: SourceFile[] = [];
let sourceRoot: string | null = null;
const topo = dt ? (await loadInstances(db, { enabledOnly: false, ids: [dt.instance_id] }))[0] : undefined;
let qbFiles: QbFileLike[] = [];
if (dt) {
if (!topo || !dt.save_path) return fail("source_path_unknown");
const base = hostPathOf(topo, dt.save_path);
@@ -330,6 +343,7 @@ export async function planImport(d: AoDeps, req: ImportRequest): Promise<PlanRes
list = ((dt.file_list ?? []) as { path: string; size: number }[]).map((f) => ({ 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;
@@ -390,16 +404,41 @@ export async function planImport(d: AoDeps, req: ImportRequest): Promise<PlanRes
? await db.selectFrom("source_release").selectAll().where("id", "=", dt.source_release_id).executeTakeFirst()
: undefined;
entityId ??= rel?.media_entity_id ?? null;
entityId ??= dt?.media_entity_id ?? null;
// QB-1 接管任务(没有站点资源 / 计划,用户手动加进 qB 的):按种子名 + 主视频文件名识别,成功直接整理
let takeover: TakeoverIdentify | null = null;
if (!entityId && dt && !dt.source_release_id && !dt.acquisition_plan_id && !dry) {
takeover = await identifyTakeover(d, dt, qbFiles);
entityId = takeover.entityId;
}
if (!entityId) {
const rid = await fail("identity_unknown");
const rid = await fail("identity_unknown", takeover ? { outcome: takeover.result.outcome } : {});
await decide_({
kind: "identity_ambiguous",
title: `无法确定作品:${dt?.name ?? req.sourcePaths?.[0] ?? "未知来源"}`,
summary: "下载内容未关联作品。识别或指定作品后再整理",
summary: takeoverSummary(takeover),
subjectType: "organization_record",
subjectId: rid.recordId,
dedupeKey: `organize_identity:${dt?.id ?? rid.recordId ?? "x"}`,
evidence: { downloadTaskId: dt?.id ?? null },
evidence: {
downloadTaskId: dt?.id ?? null,
...(takeover
? {
torrentName: dt?.name ?? null,
mainFile: takeover.mainFile,
parsed: {
title: takeover.parsed.title ?? null,
year: takeover.parsed.year ?? null,
season: takeover.parsed.season ?? null,
episodes: takeover.parsed.episodes ?? null,
},
outcome: takeover.result.outcome,
reason: takeover.result.reason,
candidates: takeover.result.candidates,
}
: {}),
},
...(dt ? { options: identityOptions(takeover, dt.name) } : {}),
});
return rid;
}
@@ -613,6 +652,7 @@ export async function planImport(d: AoDeps, req: ImportRequest): Promise<PlanRes
}
const plan = planOrganize({ ...baseInput, files, existingTargets: existing });
if (keepNames) keepOriginalNames(plan);
if (req.keepBoth) await keepBothTargets(plan);
const frozen: FrozenPlan = {
version: 1,
organize: plan,
@@ -706,9 +746,17 @@ export async function planImport(d: AoDeps, req: ImportRequest): Promise<PlanRes
.where("id", "=", dt.id)
.where("status", "in", ["completed", "importing"])
.execute();
// 下载历史按作品列出:整理计划定下的作品回填到下载任务(已有的不覆盖)
await trx
.updateTable("download_task")
.set({ media_entity_id: entity.id })
.where("id", "=", dt.id)
.where("media_entity_id", "is", null)
.execute();
}
if (blocking.length) {
const hnrConflict = blocking.find((c) => 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}`,
@@ -719,13 +767,15 @@ export async function planImport(d: AoDeps, req: ImportRequest): Promise<PlanRes
subjectType: "organization_record",
subjectId: rec.id,
dedupeKey: `organize_blocked:${dt?.id ?? rec.id}`,
evidence: { conflicts: blocking, action },
evidence: { conflicts: blocking, action, downloadTaskId: dt?.id ?? null },
options: hnrConflict
? [
{ key: "wait", label: "做种满足后再移动" },
{ key: "copy", label: "改用复制" },
]
: [{ key: "review", label: "指定处理方式" }],
: existsConflict
? CONFLICT_OPTIONS
: [{ key: "review", label: "指定处理方式" }],
});
if (slotId) await applySlotEvent(d, trx, slotId, { type: "verify_failed" }, { reason: reason ?? "blocked" });
}
@@ -1608,6 +1658,28 @@ export async function finalizeRecord(d: AoDeps, recordId: string): Promise<Execu
}
// 本下载任务先前被拦的整理事项:已有资产入库即视为处理完毕
if (okAny) await closeOrganizeDecisions(d, trx, dt.id, rec);
if (okAny)
await trx
.updateTable("download_task")
.set({ media_entity_id: frozen.entityId })
.where("id", "=", dt.id)
.where("media_entity_id", "is", null)
.execute();
}
// 手动按路径整理了某个下载任务的内容(文件管理整理向导):关闭该任务的整理事项,回填作品
if (!dt && okAny) {
const srcs = [
...new Set([frozen.sourceRoot, ...files.map((f) => 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) {
@@ -1658,7 +1730,7 @@ export async function finalizeRecord(d: AoDeps, recordId: string): Promise<Execu
return { recordId: result.recordId, status: result.status, succeeded: result.succeeded, failed: result.failed };
}
/** 关闭下载任务的"整理冲突 / 做种义务阻止移动"事项(resolution.key=manual_organize,记整理记录 id) */
/** 关闭下载任务的"无法确定作品 / 整理冲突 / 做种义务阻止移动"事项(resolution.key=manual_organize,记整理记录 id) */
async function closeOrganizeDecisions(d: AoDeps, trx: Exec, downloadTaskId: string, rec: OrganizationRecordRow) {
const by = rec.actor_id ? `${rec.actor_type}:${rec.actor_id}` : rec.actor_type;
const rows = await trx
@@ -1675,7 +1747,7 @@ async function closeOrganizeDecisions(d: AoDeps, trx: Exec, downloadTaskId: stri
organizationRecordId: rec.id,
}),
})
.where("dedupe_key", "=", `organize_blocked:${downloadTaskId}`)
.where("dedupe_key", "in", [`organize_blocked:${downloadTaskId}`, `organize_identity:${downloadTaskId}`])
.where("status", "in", ["open", "investigating", "awaiting_user"])
.returning("id")
.execute();
@@ -0,0 +1,390 @@
import { randomUUID } from "node:crypto";
import { basename, dirname, extname, join } from "node:path";
import { statSafe } from "@umt/adapters";
import type { DownloadTaskRow } from "@umt/db";
import {
fileKindOf,
mergeParsed,
type OrganizePlan,
type ParsedRelease,
parseRelease,
parseReleasePath,
} from "@umt/domain";
import type { DecisionHook, DecisionHookActor, DecisionResolveBody } from "../../decision-hooks.ts";
import { addAliasIfAbsent } from "../../sources-identity/service/entities.ts";
import type { NameIdentifyResult } from "../ports.ts";
import { jsonb } from "../util.ts";
import { type AoDeps, type Exec, ensureTaskRow, hostPathOf, loadInstances, markTask } from "./common.ts";
import { type ImportRequest, planImport } from "./organize.ts";
// QB-1 接管任务(用户手动加进 qB、贴了管理标签,没有站点资源 / 计划)完成后的作品识别与事项闭环。
// 识别对照 MP transfer:没有下载历史时按主视频文件名(合并上级 / 上上级目录名,即 MetaInfoPath)+ 种子名识别,
// 走 release-parser → ID-2 打分(本地召回 + TMDB,trigger=takeover 允许查 TMDB);识别成功直接整理,失败 / 歧义才建事项。
export interface QbFileLike {
name: string;
size: number;
priority?: number | undefined;
}
/** 主视频:要下载(priority≠0)的视频文件里最大的一个;没有视频返回 null */
export function pickMainVideo(files: readonly QbFileLike[]): string | null {
let best: QbFileLike | null = null;
for (const f of files) {
if (f.priority === 0 || fileKindOf(f.name) !== "video") continue;
if (/(^|[/._ -])sample([/._ -]|$)/i.test(f.name)) continue;
if (!best || f.size > 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<TakeoverIdentify> {
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<string, unknown>;
}
const OUTCOME_TEXT: Record<NameIdentifyResult["outcome"], string> = {
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<void> {
const held = plan.conflicts.filter((c) => c.kind === "target_exists" && c.blocking && c.src);
if (held.length === 0) return;
const mains = new Set<string>();
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<string>();
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<ImportRequest, "trigger" | "actorType" | "actorId" | "taskId">,
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<string> {
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<string> {
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<DownloadTaskRow[]> {
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;
}
@@ -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 }),
});
}
@@ -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<string, unknown> },
cur?: Record<string, unknown>,
): Record<string, unknown> | undefined => {
const curSettings = (cur?.settings ?? {}) as Record<string, unknown>;
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<string, unknown>).authKind,
username: (row.username as string | null) ?? null,
secret: secret ?? null,
}),
timeoutMs: 10_000,
retries: 0,
});
+38
View File
@@ -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<Record<string, unknown>>;
}
const hooks = new WeakMap<ServerContext, DecisionHook[]>();
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));
}
@@ -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<string, unknown> | 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(),
})
@@ -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 })
@@ -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<DomainCandidate[]> {
const ids = new Set<string>();
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<LookupReason | null> {
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<string, unknown> } = {},
): Promise<NameIdentifyResult> {
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,
};
}
@@ -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;
@@ -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 清理", () => {
+48 -6
View File
@@ -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 (
<SettingsBlock section="downloaders" anchor="instances" title="qBittorrent 实例" plain>
@@ -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,
@@ -307,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" },
+30 -2
View File
@@ -845,7 +845,9 @@ function DownloadsPanel({ detail }: { detail: LibraryDetail }) {
<p className="text-meta text-muted-foreground">
{d.plan
? `${d.plan.purpose === "upgrade" ? "升级" : d.plan.purpose === "fill" ? "补缺" : "换源"}计划 · `
: ""}
: d.downloadTask.origin === "adopted"
? "接管 · "
: ""}
{formatDateTime(d.downloadTask.createdAt)}
</p>
</div>
@@ -902,6 +904,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 (
<Panel title="下载历史与校验结果" bodyClassName="p-0 sm:p-0">
<DataTable
@@ -909,8 +914,31 @@ function DownloadsPanel({ detail }: { detail: LibraryDetail }) {
columns={columns}
rows={detail.downloads}
rowKey={(d) => d.downloadTask.id}
empty={{ title: "还没有下载记录" }}
empty={{ title: manual.length ? "没有下载任务,只有手动整理记录" : "还没有下载记录" }}
/>
{manual.length ? (
<div className="border-t border-border/60 px-4 py-3 sm:px-5">
<p className="mb-2 text-meta text-muted-foreground">其他整理记录</p>
<ul className="space-y-1.5">
{manual.map((r) => (
<li key={r.id} className="flex min-w-0 flex-wrap items-center gap-2">
<Tone v={ORG_STATUS[r.status]} />
<Link
to={consolePath("organize", r.id)}
className="min-w-0 max-w-full truncate font-mono text-caption text-foreground hover:underline sm:max-w-md"
title={r.sourcePath ?? r.id}
>
{r.targetPath ?? r.sourcePath ?? r.id}
</Link>
<span className="text-meta text-muted-foreground tabular-nums">
{r.trigger === "manual" ? "手动" : r.trigger === "agent" ? "Agent" : "自动"} · {r.fileSucceeded}/
{r.fileTotal} 个文件 · {formatDateTime(r.createdAt)}
</span>
</li>
))}
</ul>
</div>
) : null}
</Panel>
);
}
+4
View File
@@ -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:
+33 -4
View File
@@ -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,9 +247,9 @@ 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.1.2 # 5.2.x 登录返回 204 + QBT_SID_<port> 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 +269,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 <key>`,不调 `/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 校验即完成)→ 查询 → 标签 → 暂停恢复 → 删除(不删数据)。
+12
View File
@@ -79,3 +79,15 @@
- NFO/图片投影(scrape 的 `writeNfo/images`)、硬链接、蓝光原盘整目录操作仍未做。
- 搬运 move 在终态提交前删源侧空目录;若进程恰在两者之间退出,重跑 finalize 幂等,但极端情况下可能留下空目录(无数据风险)。
- 路径守卫未下沉到 `packages/adapters/src/fs/guard.ts`(本模块自带按根解析的守卫,整理仍用 `sourceGuard`)。
## 整理事项闭环(QB-1)
- 整理成功(自动重试或手动,有资产入库)后,同一下载任务的 `organize_identity:<id>`(无法确定作品)与 `organize_blocked:<id>`
(整理冲突 / 做种义务)事项一并关闭,`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 条整理记录(含没有下载任务的手动整理)。
+19
View File
@@ -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,事项保持打开。
## 待处理只收相关的
| 结果 | 开事项? |
+99 -20
View File
@@ -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<QbInstanceConfig, "authMode" | "username" | "password" | "apiKey"> {
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_<port> / 名字含 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<string, unknown>): 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<void> {
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_<WebUI 端口>(容器内端口,不一定等于映射端口),
* 也可能被自定义 —— 一律取 Set-Cookie 里的会话 cookie 原名原值回传;
* - 失败:200 "Fails." / 401 → 凭据错误(auth);403 被封禁 / 429 → 限流(rate);5xx → protocol。
*/
private async doLogin(): Promise<void> {
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<Response> {
@@ -338,10 +406,13 @@ export class QbClient implements DownloaderClient {
}
private async sendOnce(endpoint: string, opts: ReqOpts, relogged: boolean): Promise<Response> {
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<string, string> = { 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();
@@ -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<T>(fn: () => Promise<T | null | undefined | false>, ms = 30_000): Promise<T> {
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<string> {
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<Record<string, unknown>> {
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<string, unknown>): Promise<void> {
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);
}
});
@@ -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_<port> 视为成功,会话 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_<port> / 自定义名都能取到", () => {
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();
+2
View File
@@ -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(),
+12 -1
View File
@@ -50,6 +50,13 @@ const Credential = z.string().min(1).max(4096);
export const DownloaderKind = z.enum(["qbittorrent"]);
export type DownloaderKind = z.infer<typeof DownloaderKind>;
/**
* 下载器认证方式(存 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<typeof DownloaderAuthKind>;
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<typeof DownloaderInstanceCreate>;
export const DownloaderInstanceUpdate = DownloaderInstanceCreate.partial();
export type DownloaderInstanceUpdate = z.infer<typeof DownloaderInstanceUpdate>;
+21
View File
@@ -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<typeof LibraryDetail>;
@@ -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<any>): Promise<void> {
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<any>): Promise<void> {
await execDdl(
db,
`drop index if exists download_task_media_entity_idx;
alter table download_task drop column if exists media_entity_id;`,
);
}
+1
View File
@@ -703,6 +703,7 @@ export interface DownloadTaskTable {
last_error: Json | null;
created_at: GeneratedTimestamptz;
updated_at: GeneratedTimestamptz;
media_entity_id: string | null;
}
export interface DownloaderInstanceTable {
+63 -5
View File
@@ -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 {
@@ -1,4 +1,5 @@
export {
deterministicContent,
inspectTorrent as inspectTorrentFixture,
type MadeTorrent,
type MakeTorrentInput,
+27 -6
View File
@@ -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_<port>、失败 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<MockQb> {
const faults = new Map<string, MockQbFault[]>();
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<MockQb> {
}
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");
+45 -1
View File
@@ -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();
});
});