feat(organize): identify takeover downloads by file name and close organize decisions
Takeover identify (item 2): - Adopted tasks without source release / plan: on completion planImport parses main video path (file + parent + grandparent dirs, like MP MetaInfoPath) merged with the torrent name, then runs ID-2 matching via identifyParsed (local recall + TMDB; trigger "takeover" may query TMDB). auto -> bind/create entity, backfill download_task.media_entity_id and organize right away; no pool entry, no identity decision. - Failure / ambiguity: organize_identity decision with evidence and options: candidates, select entity (local / TMDB), provisional entity, dismiss. Decision closure (item 3): - decision-hooks registry: the decisions resolve route runs the owning module's executor before closing; errors keep the decision open. - organize_identity: backfill entity, freeze manual plan, enqueue run. - organize_blocked with target_exists: replace (verify + trash old) / keep both (rename "<name> (n)" with sidecars) / skip. - Successful organize closes both organize_identity and organize_blocked of the same download task (resolution manual_organize), also for manual path organizes matched to download tasks via path mappings. Co-Authored-By: Claude Opus 5.5 <[email protected]>
This commit is contained in:
rootandClaude Opus 5.5 committed
1 parent
05d1f76b9f
commit
e7de7e2e69
13 files changed
+1127
-17
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,
|
||||
);
|
||||
});
|
||||
});
|
||||
@@ -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);
|
||||
|
||||
@@ -15,7 +15,16 @@ import {
|
||||
type TorrentSource,
|
||||
type VerifyMethod,
|
||||
} from "@umt/adapters";
|
||||
import type { ActorType } from "@umt/contracts";
|
||||
import type { DownloaderInstanceRow, MediaServerInstanceRow, SiteRow, SourceReleaseRow } from "@umt/db";
|
||||
import type { ParsedRelease } from "@umt/domain";
|
||||
import { SourcesRuntime } from "../sources-identity/runtime.ts";
|
||||
import {
|
||||
createEntityFromTmdbCandidate,
|
||||
enqueueNewEntityRefresh,
|
||||
} from "../sources-identity/service/entity-from-candidate.ts";
|
||||
import { metadataLanguages } from "../sources-identity/service/metadata-rich.ts";
|
||||
import { identifyParsed, type NameIdentifyResult } from "../sources-identity/service/resolve.ts";
|
||||
import type { ServerContext } from "../types.ts";
|
||||
|
||||
// 外部依赖端口:默认实现接真实 qB / 站点 / 文件系统 / Emby;测试注入 fake(mock qB、临时目录、固定磁盘用量)。
|
||||
@@ -58,9 +67,21 @@ export interface AoPorts {
|
||||
diskUsage(path: string): Promise<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,
|
||||
@@ -90,8 +111,31 @@ export function createPorts(
|
||||
}
|
||||
};
|
||||
|
||||
let sources: SourcesRuntime | null = null;
|
||||
const rt = () => {
|
||||
sources ??= new SourcesRuntime(ctx);
|
||||
return sources;
|
||||
};
|
||||
|
||||
const ports: AoPorts = {
|
||||
now: () => new Date(),
|
||||
async identify({ parsed, evidence }) {
|
||||
const r = await identifyParsed(rt(), parsed, { trigger: "takeover", ...(evidence ? { evidence } : {}) });
|
||||
if (r.created && r.mediaEntityId) await enqueueNewEntityRefresh(ctx, r.mediaEntityId);
|
||||
return r;
|
||||
},
|
||||
async entityFromTmdb(ref, actor) {
|
||||
const type: ActorType = (["user", "agent", "system", "api_key"] as const).find((t) => t === actor.type) ?? "user";
|
||||
const r = await createEntityFromTmdbCandidate(
|
||||
ctx.db,
|
||||
() => rt().tmdb(),
|
||||
{ kind: ref.type, externalId: ref.id },
|
||||
{ type, id: actor.id },
|
||||
{ languages: await metadataLanguages(ctx) },
|
||||
);
|
||||
if (r.created) await enqueueNewEntityRefresh(ctx, r.mediaEntityId);
|
||||
return r.mediaEntityId;
|
||||
},
|
||||
async downloader(inst) {
|
||||
const stamp = `${inst.updated_at.toISOString()}|${inst.base_url}`;
|
||||
const hit = clients.get(inst.id);
|
||||
|
||||
@@ -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]!;
|
||||
|
||||
@@ -46,6 +46,16 @@ import { PathGuardError } from "./organize-guard.ts";
|
||||
import { type FrozenTransfer, finalizeTransfer, planTransfer, type TransferSpec } from "./organize-transfer.ts";
|
||||
import { hnrSatisfied } from "./retention.ts";
|
||||
import { slotComplete as sharedSlotComplete, subscriptionLibraryRoot } from "./subscriptions.ts";
|
||||
import {
|
||||
CONFLICT_OPTIONS,
|
||||
downloadTasksUnder,
|
||||
identifyTakeover,
|
||||
identityOptions,
|
||||
keepBothTargets,
|
||||
type QbFileLike,
|
||||
type TakeoverIdentify,
|
||||
takeoverSummary,
|
||||
} from "./takeover.ts";
|
||||
|
||||
// 入库:冻结整理计划(organization_record + 文件明细)→ 逐文件可恢复执行(操作账本 fencing)→
|
||||
// 资产/位置/路径绑定同事务登记 → Slot 推进 → Move 后移除 qB 任务记录(不删数据)。
|
||||
@@ -70,6 +80,8 @@ export interface ImportRequest {
|
||||
onExisting?: ExistingPolicy | undefined;
|
||||
/** 已获替换授权(审批/手动确认) */
|
||||
grantReplace?: boolean | undefined;
|
||||
/** QB-1 目标已存在时"保留两者":新文件(连同附属文件)改名加序号放在旁边,已有文件不动 */
|
||||
keepBoth?: boolean | undefined;
|
||||
/**
|
||||
* W5-A2 逐文件覆写(文件管理整理向导的人工确认,优先于文件名解析):键为源文件或目录的绝对路径
|
||||
* (目录对其下文件生效,最长前缀优先)。season/episodes 决定剧集季目录与命名以及集单元覆盖;
|
||||
@@ -312,6 +324,7 @@ export async function planImport(d: AoDeps, req: ImportRequest): Promise<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);
|
||||
@@ -329,6 +342,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;
|
||||
@@ -389,16 +403,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;
|
||||
}
|
||||
@@ -612,6 +651,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,
|
||||
@@ -705,9 +745,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}`,
|
||||
@@ -718,13 +766,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" });
|
||||
}
|
||||
@@ -1604,6 +1654,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) {
|
||||
@@ -1654,7 +1726,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
|
||||
@@ -1671,7 +1743,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;
|
||||
}
|
||||
@@ -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(),
|
||||
})
|
||||
|
||||
@@ -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,
|
||||
};
|
||||
}
|
||||
@@ -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 清理", () => {
|
||||
|
||||
Reference in new issue
Block a user