feat: v0.3.8–v0.3.14 工作区汇总(依赖升级、整理来源、手动入库、原地入库等)

生产 v0.3.8~v0.3.13 均由 8013196 加工作区修改构建部署,此前未提交;本提交补齐源码,
并含未部署的 v0.3.14 原地入库。中间各版本源码状态未保存,无法逐版拆分,详见
docs/plan/17-full-upgrade.md 与 docs/plan/18-organize-release.md。

- v0.3.8 全面升级:Deno workspace 依赖、Router 8 / Vite 8、camofox 1.18.0 / Node 24、
  PostgreSQL 18.6、备份客户端 18.6、演示 qB 5.2.4
- v0.3.9 媒体整理:分类规则、整理来源与目录快照(0059)、文件管理整理入口
- v0.3.10 手动下载与入库:独立手动下载、详情页资源下载、TMDB / Local 元数据兜底
  (metadata.resolve_organize)、正常多版本 / 一次性洗版(0060)
- v0.3.11 搜索详情与桌面黑边修复
- v0.3.12 列表分页、搜索历史、首页概况
- v0.3.13 统一整理弹窗与队列
- v0.3.14 原地入库(0061,未部署)

同批审计修复:
- 刮削入队:确认原记录存在后才删重放产生的新记录;写回 run 关联失败时保持 queued,
  由处理器补关联(对齐 MP 先登记、只在入队失败时记失败)
- 识别升级链只在结果更确定时采用,不再用 no_match 顶替原有歧义候选
- bluray_dir 资产以原盘目录本身为刮削根(MP 同做法)
- metadata.resolve_organize 登记到入库排障 Skill 与能力地图
- db/src/types.ts 按迁移重新生成;contracts 整理动作枚举顺序与数据库对齐
- 测试:原地入库用例改为直接查记录表;domain 测试模板补 version
- 前端:通知铃铛弹层与用户菜单统一样式;站点卡片共用卡头、底栏贴底、分页间距

依赖升级证据只提交摘要文件,registry-snapshot.json 移至 .local/(SHA-256 记于文档)。

Co-Authored-By: Claude Opus 5.5 <[email protected]>
This commit is contained in:
rootandClaude Opus 5.5 committed 2026-10-02 15:47:17 +08:00
1 parent 8013196e3f
commit ceb66f8c77
265 files changed
+17751 -2483

No files matched your search

@@ -0,0 +1,142 @@
import { OpenAPIHono } from "@hono/zod-openapi";
import type { TmdbClient } from "@umt/adapters";
import { DEFAULT_CLASSIFICATION_YAML } from "@umt/domain";
import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest";
import type { AuthEnv } from "../../../http/auth.ts";
import { createSettingsReader } from "../../auth-settings/provided.ts";
import { registerSettingsRoutes } from "../../auth-settings/settings-routes.ts";
import { SourcesRuntime } from "../../sources-identity/runtime.ts";
import { classifyEntity } from "../services/classification.ts";
import { createHarness, type Harness, pgAvailable, seedEntity } from "./harness.ts";
const hasPg = await pgAvailable();
describe.skipIf(!hasPg)("TMDB classification API and accepted identities (real PG)", () => {
let h: Harness;
beforeAll(async () => {
h = await createHarness("classification");
const settingsApp = new OpenAPIHono<AuthEnv>();
registerSettingsRoutes(settingsApp, h.ctx);
h.app.route("/api/v1", settingsApp);
}, 60_000);
afterEach(() => vi.restoreAllMocks());
afterAll(async () => await h?.close());
it("legacy stored organize settings receive default YAML without losing preferences", async () => {
await h.setSettings("organize", { minSizeBytes: 456789, conflictDefault: "skip" });
const reader = createSettingsReader({ db: h.db, logger: h.ctx.logger });
expect(await reader.get("organize")).toMatchObject({
minSizeBytes: 456789,
classificationYaml: DEFAULT_CLASSIFICATION_YAML,
});
});
it("ordered categories and raw preview do not write settings", async () => {
const view = await h.api("GET", "/classification");
expect(view.status).toBe(200);
expect(view.body.data.yaml).toBe(DEFAULT_CLASSIFICATION_YAML);
expect(view.body.data.categories.movie).toEqual(["动画电影", "演唱会", "华语电影", "亚洲电影", "欧美电影"]);
const preview = await h.api("POST", "/classification/preview", {
yaml: "movie:\n 日本片:\n production_countries: JP\ntv: {}\n",
mediaType: "movie",
tmdb: { production_countries: [{ iso_3166_1: "JP" }] },
});
expect(preview.status).toBe(200);
expect(preview.body.data).toEqual({ category: "日本片", matchedFields: ["production_countries"], fallback: false });
expect((await h.ctx.settings.get("organize")).classificationYaml).toBe(DEFAULT_CLASSIFICATION_YAML);
});
it("PUT preserves organize preferences, audits and emits settings change; invalid writes reject", async () => {
await h.setSettings("organize", { ...(await h.ctx.settings.get("organize")), minSizeBytes: 765432 });
const publish = vi.spyOn(h.ctx.events, "publish");
const yaml = "movie:\n 新分类:\n original_language: zh\n 其它: null\ntv: {}\n";
const saved = await h.api("PUT", "/classification", { yaml });
expect(saved.status).toBe(200);
expect(saved.body.data.categories.movie).toEqual(["新分类", "其它"]);
expect(await h.ctx.settings.get("organize")).toMatchObject({ classificationYaml: yaml, minSizeBytes: 765432 });
expect(publish).toHaveBeenCalledWith("settings.changed", { keys: ["organize"] });
expect(
(await h.db.selectFrom("audit_event").selectAll().where("action", "=", "settings.update").execute()).length,
).toBeGreaterThan(0);
for (const invalid of ["movie:\n ../escape: null\ntv: {}", "movie:\n 同名: null\n 同名: null\ntv: {}"]) {
expect((await h.api("PUT", "/classification", { yaml: invalid })).status).toBe(422);
expect((await h.api("PATCH", "/settings", { organize: { classificationYaml: invalid } })).status).toBe(422);
}
expect((await h.ctx.settings.get("organize")).classificationYaml).toBe(yaml);
await h.api("PUT", "/classification", { yaml: DEFAULT_CLASSIFICATION_YAML });
});
it("PUT merges current row after waiting rather than overwriting a newer source edit", async () => {
const source = {
id: "33333333-3333-4333-8333-333333333333",
name: "Concurrent source",
enabled: true,
source: { root: "lib:11111111-1111-4111-8111-111111111111", path: "" },
target: { root: "lib:22222222-2222-4222-8222-222222222222", path: "" },
mediaType: "all",
category: null,
action: "copy",
trigger: "manual",
downloaderInstanceId: null,
tag: null,
intervalSec: 300,
};
const originalAll = h.ctx.settings.all.bind(h.ctx.settings);
vi.spyOn(h.ctx.settings, "all").mockImplementationOnce(async () => {
const stale = await originalAll();
await h.setSettings("organize", { ...stale.organize, minSizeBytes: 654321, sources: [source] });
return stale;
});
expect((await h.api("PUT", "/classification", { yaml: DEFAULT_CLASSIFICATION_YAML })).status).toBe(200);
const stored = await h.db
.selectFrom("settings")
.select("value")
.where("key", "=", "organize")
.executeTakeFirstOrThrow();
expect(stored.value).toMatchObject({
minSizeBytes: 654321,
sources: [source],
classificationYaml: DEFAULT_CLASSIFICATION_YAML,
});
});
it("requires accepted movie/tv identity, uses raw detail, and fails without fabricating fallback", async () => {
await h.setSettings("organize", {
...(await h.ctx.settings.get("organize")),
classificationYaml: DEFAULT_CLASSIFICATION_YAML,
});
const id = await seedEntity(h.db, { kind: "movie", title: "测试片", year: 2020 });
const entity = await h.db.selectFrom("media_entity").selectAll().where("id", "=", id).executeTakeFirstOrThrow();
const getJson = vi.fn().mockResolvedValue({ genres: [{ id: 16 }], original_language: "zh" });
const factory = vi.spyOn(SourcesRuntime.prototype, "tmdb").mockResolvedValue({ getJson } as unknown as TmdbClient);
expect((await classifyEntity(h.d, entity)).category).toBeNull();
expect(factory).not.toHaveBeenCalled();
const identity = await h.db
.insertInto("external_identity")
.values({ media_entity_id: id, provider: "tmdb", entity_type: "movie", external_id: "123", status: "proposed" })
.returning("id")
.executeTakeFirstOrThrow();
expect((await classifyEntity(h.d, entity)).category).toBeNull();
await h.db.updateTable("external_identity").set({ status: "accepted" }).where("id", "=", identity.id).execute();
expect(await classifyEntity(h.d, entity)).toEqual({
category: "动画电影",
matchedFields: ["genre_ids"],
fallback: false,
});
expect(getJson).toHaveBeenCalledWith("/movie/123");
getJson.mockRejectedValue(new Error("offline"));
expect(await classifyEntity(h.d, entity)).toEqual({ category: null, matchedFields: [], fallback: false });
getJson.mockResolvedValue({});
expect((await classifyEntity(h.d, entity)).category).toBeNull();
getJson.mockResolvedValue({ genres: [{ id: 10764 }], origin_country: ["CN"] });
const tvId = await seedEntity(h.db, { kind: "series", title: "测试剧", year: 2020 });
const tv = await h.db.selectFrom("media_entity").selectAll().where("id", "=", tvId).executeTakeFirstOrThrow();
await h.db
.insertInto("external_identity")
.values({ media_entity_id: tvId, provider: "tmdb", entity_type: "tv", external_id: "456", status: "accepted" })
.execute();
expect((await classifyEntity(h.d, tv)).category).toBe("综艺");
expect(getJson).toHaveBeenCalledWith("/tv/456");
factory.mockResolvedValue(null);
expect((await classifyEntity(h.d, entity)).category).toBeNull();
});
});
@@ -0,0 +1,207 @@
import type { TaskRunView } from "@umt/tasks";
import { createTempTree, type TempTree } from "@umt/testkit";
import { sql } from "kysely";
import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest";
import { enqueueSweep, enqueueTracked } from "../services/common.ts";
import { type QueueScrapeInput, queueScrape, runScrapeRecord } from "../services/scrape.ts";
import { createHarness, type Harness, pgAvailable, seedBase, seedEntity } from "./harness.ts";
const hasPg = await pgAvailable();
const runView = (runId: string, input: unknown, status: TaskRunView["status"] = "pending"): TaskRunView => ({
runId,
taskName: "scrapeMedia",
version: 2,
status,
attempts: 0,
idempotencyKey: null,
traceId: null,
priority: 0,
input,
output: null,
error: null,
cancelRequestedAt: null,
cancelReason: null,
availableAt: null,
createdAt: new Date(),
startedAt: null,
finishedAt: null,
});
describe.skipIf(!hasPg)("入队关联与巡检恢复", () => {
let h: Harness;
let lib: TempTree;
let rootId: string;
let entityId: string;
beforeAll(async () => {
lib = await createTempTree({}, { sentinel: true });
h = await createHarness("enqueue_recovery");
({ rootId } = await seedBase(h.db, { libraryPath: lib.root, cachePath: lib.root, qbUrl: "http://127.0.0.1:9" }));
entityId = await seedEntity(h.db, { kind: "movie", title: "恢复测试", year: 2020 });
await sql`create function reject_association() returns trigger language plpgsql as $$
begin
if new.engine_run_id is not null and new.engine_run_id is distinct from old.engine_run_id
then raise exception 'injected association failure'; end if;
return new;
end $$`.execute(h.db);
});
async function rejectAssociation(table: "task" | "scrape_record") {
await sql
.raw(`create trigger reject_association before update on ${table}
for each row execute function reject_association()`)
.execute(h.db);
}
async function allowAssociation() {
for (const table of ["task", "scrape_record"])
await sql.raw(`drop trigger if exists reject_association on ${table}`).execute(h.db);
}
afterEach(async () => {
vi.restoreAllMocks();
await allowAssociation();
});
afterAll(async () => {
await h?.close();
await lib?.cleanup();
});
const task = (key: string) => ({ kind: "reconcileLibrary" as const, idempotencyKey: key, input: {} });
const scrape = (path: string, idempotencyKey?: string): QueueScrapeInput => ({
rootId: `lib:${rootId}`,
rootPath: lib.root,
libraryRootId: rootId,
path,
match: {
mediaEntityId: entityId,
provider: null,
externalId: null,
kind: "movie",
title: "恢复测试",
year: 2020,
source: "chosen",
},
trigger: "manual",
mode: "missing",
...(idempotencyKey ? { idempotencyKey } : {}),
});
it("引擎查询异常跳过本轮,下一轮仍查原任务", async () => {
const t = task("query_error");
const q = await enqueueTracked(h.d, t);
const enqueue = vi.spyOn(h.engine, "enqueue");
const get = vi
.spyOn(h.engine, "get")
.mockRejectedValueOnce(new Error("temporary engine outage"))
.mockResolvedValue(runView(q.runId, {}, "running"));
expect(await enqueueSweep(h.d, t, "next")).toBeNull();
expect(await enqueueSweep(h.d, t, "later")).toBeNull();
expect(enqueue).not.toHaveBeenCalled();
expect(get).toHaveBeenNthCalledWith(2, q.runId);
expect(
(await h.db.selectFrom("task").select("engine_run_id").where("id", "=", q.taskId).executeTakeFirstOrThrow())
.engine_run_id,
).toBe(q.runId);
});
it.each(["missing", "completed"])("明确 %s 时允许补投", async (state) => {
const t = task(`retry_${state}`);
const q = await enqueueTracked(h.d, t);
vi.spyOn(h.engine, "get").mockResolvedValue(state === "missing" ? null : runView(q.runId, {}, "completed"));
const enqueue = vi.spyOn(h.engine, "enqueue");
expect(await enqueueSweep(h.d, t, "next")).not.toBeNull();
expect(enqueue).toHaveBeenCalledOnce();
expect(enqueue.mock.calls[0]?.[2].idempotencyKey).toBe(`${t.kind}:${t.idempotencyKey}:next`);
});
it.each([false, true])("关联写回失败后巡检按原 taskId 找回已接受的 run (旧关联=%s)", async (oldAssociation) => {
const t = task(`association_${oldAssociation}`);
if (oldAssociation) await enqueueTracked(h.d, t);
await rejectAssociation("task");
const q = await enqueueTracked(h.d, t, { engineKey: "new-slot" });
await h.db
.insertInto("tasks_run_meta")
.values({ run_id: q.runId, task_name: t.kind, task_version: 1, idempotency_key: "new-slot", trace_id: q.taskId })
.execute();
await allowAssociation();
vi.spyOn(h.engine, "get").mockResolvedValue(runView(q.runId, {}, "running"));
const enqueue = vi.spyOn(h.engine, "enqueue");
expect(await enqueueSweep(h.d, t, "later")).toBeNull();
expect(enqueue).not.toHaveBeenCalled();
expect(
(await h.db.selectFrom("task").select("engine_run_id").where("id", "=", q.taskId).executeTakeFirstOrThrow())
.engine_run_id,
).toBe(q.runId);
});
it("刮削已接受而关联失败时保持 queued,重试复用记录与原幂等键", async () => {
const input = scrape("association-retry");
await rejectAssociation("scrape_record");
const q = await queueScrape(h.d, input);
expect(
await h.db
.selectFrom("scrape_record")
.select(["status", "reason", "engine_run_id"])
.where("id", "=", q.recordId)
.executeTakeFirstOrThrow(),
).toMatchObject({ status: "queued", reason: null, engine_run_id: null });
const key = h.engine.enqueued.at(-1)?.key;
await allowAssociation();
const enqueue = vi.spyOn(h.engine, "enqueue").mockResolvedValue({ runId: q.taskId!, deduplicated: true });
vi.spyOn(h.engine, "get").mockResolvedValue(runView(q.taskId!, { recordId: q.recordId }));
expect(await queueScrape(h.d, input)).toEqual({ recordId: q.recordId, taskId: q.taskId, deduplicated: true });
expect(enqueue.mock.calls[0]?.[2].idempotencyKey).toBe(key);
expect((await h.db.selectFrom("scrape_record").select("id").where("path", "=", input.path).execute()).length).toBe(
1,
);
});
it("刮削处理器开始时补关联,不会把已接受的工作误标为入队失败", async () => {
await rejectAssociation("scrape_record");
const q = await queueScrape(h.d, scrape("worker-backfill"));
await allowAssociation();
await h.db.updateTable("library_root").set({ management_mode: "register_only" }).where("id", "=", rootId).execute();
const r = await runScrapeRecord(h.d, q.recordId, undefined, q.taskId!);
expect(r.reason).toBe("scrape_root_read_only");
expect(
(
await h.db
.selectFrom("scrape_record")
.select("engine_run_id")
.where("id", "=", q.recordId)
.executeTakeFirstOrThrow()
).engine_run_id,
).toBe(q.taskId);
});
it("确实入队失败的新刮削记录标失败", async () => {
vi.spyOn(h.engine, "enqueue").mockRejectedValue(new Error("enqueue rejected"));
const input = scrape("enqueue-failed");
await expect(queueScrape(h.d, input)).rejects.toThrow("enqueue rejected");
expect(
await h.db
.selectFrom("scrape_record")
.select(["status", "reason"])
.where("path", "=", input.path)
.executeTakeFirstOrThrow(),
).toMatchObject({ status: "failed", reason: "task_enqueue_failed" });
});
it("自动重放从原 run 输入找回未关联记录,只删除本次多建的记录", async () => {
const input = scrape("auto-replay", "auto-stable-key");
await rejectAssociation("scrape_record");
const q = await queueScrape(h.d, input);
await allowAssociation();
await h.db.updateTable("scrape_record").set({ status: "succeeded" }).where("id", "=", q.recordId).execute();
vi.spyOn(h.engine, "enqueue").mockResolvedValue({ runId: q.taskId!, deduplicated: true });
vi.spyOn(h.engine, "get").mockResolvedValue(runView(q.taskId!, { recordId: q.recordId }, "completed"));
expect(await queueScrape(h.d, input)).toEqual({ recordId: q.recordId, taskId: q.taskId, deduplicated: true });
const recs = await h.db
.selectFrom("scrape_record")
.select(["id", "status", "engine_run_id"])
.where("path", "=", input.path)
.execute();
expect(recs).toEqual([{ id: q.recordId, status: "succeeded", engine_run_id: q.taskId }]);
});
});
@@ -2,7 +2,7 @@ import type { DB } from "@umt/db";
import type { Kysely } from "kysely";
import { afterAll, beforeAll, describe, expect, it } from "vitest";
import { backfillVersionSpecs } from "../services/version-backfill.ts";
import { createHarness, type Harness, pgAvailable, seedEntity, seedSlot } from "./harness.ts";
import { createHarness, type Harness, pgAvailable, seedEntity, seedRelease, seedSlot } from "./harness.ts";
// LIBV-1:入库版本 API(每集多个版本、季汇总、单目标比对)与规格回填(dry-run / apply)。
@@ -272,4 +272,68 @@ describe.skipIf(!hasPg)("LIBV-1 入库版本", () => {
const dv = r.body.data.seasons[0].episodes[0].versions.find((x: { assetId: string }) => x.assetId === ids.e1dv);
expect(dv).toMatchObject({ specStored: true, hdr: ["DV", "HDR10"], releaseGroup: "FLUX" });
});
it("作品详情保留跨过旧条数上限的整理、下载和失败计划供分页查看", async () => {
const entityId = await seedEntity(h.db, { kind: "movie", title: "历史分页测试", year: 2026 });
const instance = await h.db
.insertInto("downloader_instance")
.values({ name: "历史下载器", base_url: "http://127.0.0.1:1" })
.returning("id")
.executeTakeFirstOrThrow();
await h.db
.insertInto("download_task")
.values(
Array.from({ length: 105 }, (_, i) => ({
instance_id: instance.id,
media_entity_id: entityId,
torrent_hash: String(i + 1).padStart(40, "0"),
name: `历史下载 ${i + 1}`,
})),
)
.execute();
const slotId = await seedSlot(h.db, { entityId, name: "历史版本" });
const site = await h.db
.insertInto("site")
.values({ key: "history-source", name: "历史来源", base_url: "http://127.0.0.1:1", adapter: "nexusphp" })
.returning("id")
.executeTakeFirstOrThrow();
const releaseId = await seedRelease(h.db, {
siteId: site.id,
entityId,
title: "历史分页测试",
size: 1000,
hash: "a".repeat(40),
});
await h.db
.insertInto("acquisition_plan")
.values(
Array.from({ length: 51 }, (_, i) => ({
slot_id: slotId,
source_release_id: releaseId,
status: "failed" as const,
payload_hash: `history-${i}`,
idempotency_key: `history:${i}`,
})),
)
.execute();
await h.db
.insertInto("organization_record")
.values(
Array.from({ length: 105 }, (_, i) => ({
media_entity_id: entityId,
trigger: "manual" as const,
actor_type: "user" as const,
action: "copy" as const,
status: "succeeded" as const,
source_path: `/history/${i}.mkv`,
target_path: `/library/history/${i}.mkv`,
})),
)
.execute();
const response = await h.api("GET", `/library/${entityId}`);
expect(response.status).toBe(200);
expect(response.body.data.organizationRecords).toHaveLength(105);
expect(response.body.data.downloads).toHaveLength(105);
expect(response.body.data.droppedPlans).toHaveLength(51);
});
});
@@ -0,0 +1,153 @@
import { ManualAcquisitionRequest } from "@umt/contracts";
import { createMockQb, createTempTree, makeTorrent } from "@umt/testkit";
import { describe, expect, it } from "vitest";
import { evaluateAcquisition } from "../services/acquisition.ts";
import { submitDownload } from "../services/downloads.ts";
import { planManualAcquisition } from "../services/manual-acquisition.ts";
import { createHarness, fakeDisk, pgAvailable, seedBase, seedEntity, seedRelease, seedSlot } from "./harness.ts";
const hasPg = await pgAvailable();
describe.skipIf(!hasPg)("独立手动下载", () => {
it("无订阅下载,未知资源确认,同hash沿用原路径", async () => {
const qb = await createMockQb({ addVisibleAfterPolls: 0 });
const cache = await createTempTree({});
const lib = await createTempTree({}, { sentinel: true });
const torrents = new Map<string, Uint8Array>();
const disk = fakeDisk(10_000_000);
disk.setPct(20);
const h = await createHarness("manual_acquisition", {
diskUsage: disk.fn,
torrentSource: async ({ release }) => ({ kind: "file", bytes: torrents.get(release.id)! }),
});
try {
const base = await seedBase(h.db, { libraryPath: lib.root, cachePath: cache.root, qbUrl: qb.url });
const entityId = await seedEntity(h.db, { kind: "movie", title: "Manual Movie", year: 2026 });
const torrent = makeTorrent({ name: "Manual.Movie.2026.1080p.mkv", length: 300_000 });
const relId = await seedRelease(h.db, {
siteId: base.siteId,
entityId,
title: torrent.name,
size: 300_000,
hash: torrent.hash,
});
torrents.set(relId, torrent.bytes);
const request = (id: string, opts = {}) => ManualAcquisitionRequest.parse({ sourceReleaseId: id, ...opts });
const first = await planManualAcquisition(h.ctx, request(relId));
expect(first).toMatchObject({ submitted: true, slotId: null });
expect(await h.db.selectFrom("subscription").selectAll().execute()).toHaveLength(0);
expect(await h.db.selectFrom("acquisition_slot").selectAll().execute()).toHaveLength(0);
const sub = await submitDownload(h.d, first.plan!.id);
expect(sub.status).toBe("submitted");
const rel2 = await seedRelease(h.db, {
siteId: base.siteId,
entityId,
title: "Other.Release.2160p.mkv",
size: 300_000,
hash: torrent.hash,
});
torrents.set(rel2, torrent.bytes);
const second = await planManualAcquisition(h.ctx, request(rel2, { savePath: `${cache.root}/other` }));
const reused = await submitDownload(h.d, second.plan!.id);
expect(reused).toMatchObject({ reason: "already_exists", downloadTaskId: sub.downloadTaskId });
const dt = await h.db
.selectFrom("download_task")
.selectAll()
.where("id", "=", sub.downloadTaskId!)
.executeTakeFirstOrThrow();
expect(dt.save_path).toBe(cache.root);
expect(dt.media_entity_id).toBe(entityId);
const unknown = makeTorrent({ name: "Unknown.1080p.mkv", length: 300_000, pieceSeed: 42 });
const uid = await seedRelease(h.db, {
siteId: base.siteId,
entityId,
title: unknown.name,
size: 300_000,
hash: unknown.hash,
});
await h.db.updateTable("source_release").set({ media_entity_id: null }).where("id", "=", uid).execute();
torrents.set(uid, unknown.bytes);
expect(await planManualAcquisition(h.ctx, request(uid))).toMatchObject({
submitted: false,
requiresConfirmation: true,
});
const confirmed = await planManualAcquisition(h.ctx, request(uid, { allowUnrecognized: true }));
const submitted = await submitDownload(h.d, confirmed.plan!.id);
expect(submitted.status).toBe("submitted");
expect(
(
await h.db
.selectFrom("download_task")
.select("media_entity_id")
.where("id", "=", submitted.downloadTaskId!)
.executeTakeFirstOrThrow()
).media_entity_id,
).toBeNull();
const autoEntity = await seedEntity(h.db, { kind: "movie", title: "Auto Movie", year: 2026 });
const slotId = await seedSlot(h.db, { entityId: autoEntity });
const autoTorrent = makeTorrent({
name: "Auto.Movie.2026.1080p.BluRay.x264-GRP.mkv",
length: 300_000,
pieceSeed: 100,
});
const autoRelease = await seedRelease(h.db, {
siteId: base.siteId,
entityId: autoEntity,
title: autoTorrent.name,
size: 300_000,
hash: autoTorrent.hash,
});
torrents.set(autoRelease, autoTorrent.bytes);
const auto = await evaluateAcquisition(h.d, { sourceReleaseId: autoRelease });
expect(auto.plans).toHaveLength(1);
const autoTask = await submitDownload(h.d, auto.plans[0]!);
const oldAsset = await h.db
.insertInto("media_asset")
.values({ media_entity_id: autoEntity, kind: "video", size_bytes: 300_000, status: "active" })
.returning("id")
.executeTakeFirstOrThrow();
await h.db
.updateTable("download_task")
.set({ status: "imported" })
.where("id", "=", autoTask.downloadTaskId!)
.execute();
const unchanged = await planManualAcquisition(h.ctx, request(autoRelease));
expect(unchanged.plan!.id).toBe(auto.plans[0]);
const wash = await planManualAcquisition(
h.ctx,
request(autoRelease, { importMode: "replace", replacementAssetIds: [oldAsset.id] }),
);
const washed = await submitDownload(h.d, wash.plan!.id);
expect(washed.downloadTaskId).toBe(autoTask.downloadTaskId);
const linked = await h.db
.selectFrom("download_task")
.selectAll()
.where("id", "=", autoTask.downloadTaskId!)
.executeTakeFirstOrThrow();
expect(linked).toMatchObject({
acquisition_plan_id: auto.plans[0],
import_plan_id: wash.plan!.id,
status: "completed",
});
expect(
(
await h.db
.selectFrom("acquisition_plan")
.select("slot_id")
.where("id", "=", auto.plans[0]!)
.executeTakeFirstOrThrow()
).slot_id,
).toBe(slotId);
expect(h.engine.enqueued).toContainEqual(
expect.objectContaining({
task: "verifyAndImport",
input: expect.objectContaining({ downloadTaskId: autoTask.downloadTaskId }),
}),
);
} finally {
await h.close();
await qb.close();
await cache.cleanup();
await lib.cleanup();
}
});
});
@@ -0,0 +1,248 @@
import { randomUUID } from "node:crypto";
import { join } from "node:path";
import type { TmdbClient } from "@umt/adapters";
import { DEFAULT_SETTINGS, type OrganizeSourceConfig } from "@umt/contracts";
import { createMockQb, createTempTree, type MockQb, type TempTree } from "@umt/testkit";
import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest";
import { SourcesRuntime } from "../../sources-identity/runtime.ts";
import { syncDownloads } from "../services/downloads.ts";
import { executeRecord, planImport } from "../services/organize.ts";
import { createHarness, type Harness, pgAvailable, seedBase, seedEntity } from "./harness.ts";
const hasPg = await pgAvailable();
const FILE = "Example.2020.1080p.BluRay.x264-GRP.mkv";
describe.skipIf(!hasPg)("configured organize sources (real PG, files, mock qB)", () => {
let h: Harness;
let lib: TempTree;
let cache: TempTree;
let qb: MockQb;
let otherQb: MockQb;
let base: Awaited<ReturnType<typeof seedBase>>;
let otherInstance: string;
beforeAll(async () => {
lib = await createTempTree({}, { sentinel: true });
cache = await createTempTree({ [`a/${FILE}`]: 200_000, [`b/${FILE}`]: 200_000, [`tagged/${FILE}`]: 200_000 });
qb = await createMockQb();
otherQb = await createMockQb();
h = await createHarness("source_config");
base = await seedBase(h.db, { libraryPath: lib.root, cachePath: cache.root, qbUrl: qb.url });
otherInstance = (
await h.db
.insertInto("downloader_instance")
.values({
name: "other-qb",
base_url: otherQb.url,
username: "admin",
settings: JSON.stringify({ savePath: cache.root }),
})
.returning("id")
.executeTakeFirstOrThrow()
).id;
}, 60_000);
afterEach(() => vi.restoreAllMocks());
afterAll(async () => {
await h?.close();
await qb?.close();
await otherQb?.close();
await lib?.cleanup();
await cache?.cleanup();
});
const profile = (patch: Partial<OrganizeSourceConfig> = {}): OrganizeSourceConfig => ({
id: randomUUID(),
name: "整理来源",
enabled: true,
source: { root: `dl:${base.instanceId}:0`, path: "a" },
target: { root: `lib:${base.rootId}`, path: "华语电影" },
mediaType: "movie",
category: "华语电影",
action: "copy",
trigger: "manual",
downloaderInstanceId: null,
tag: null,
intervalSec: 300,
...patch,
});
const settings = (
sources: OrganizeSourceConfig[],
classificationYaml = DEFAULT_SETTINGS.organize.classificationYaml,
) => h.setSettings("organize", { ...DEFAULT_SETTINGS.organize, minSizeBytes: 100_000, sources, classificationYaml });
const request = (entityId: string, source: OrganizeSourceConfig, folder = "a") => ({
sourcePaths: [cache.path(`${folder}/${FILE}`)],
mediaEntityId: entityId,
sourceConfigId: source.id,
trigger: "manual" as const,
actorType: "user" as const,
dryRun: true,
});
it("fixed category adds no duplicate; explicit same-category source selects own target and rejects wrong source", async () => {
const a = profile();
const b = profile({
source: { root: `dl:${base.instanceId}:0`, path: "b" },
target: { root: `lib:${base.rootId}`, path: "另一个目录/华语电影" },
});
await settings([a, b]);
const movie = await seedEntity(h.db, { kind: "movie", title: "固定片", year: 2020 });
const first = await planImport(h.d, request(movie, a));
expect(first.plan?.sourceConfigId).toBe(a.id);
expect(first.plan?.newBindings[0]?.relativePath).toBe("华语电影/固定片 (2020)");
const explicit = await planImport(h.d, { ...request(movie, a), targetDir: join(lib.root, "华语电影") });
expect(explicit.plan?.newBindings).toEqual([]);
const explicitDst = explicit.plan?.organize.files.find((f) => f.role === "main")?.dst;
expect(explicitDst?.startsWith(`${join(lib.root, "华语电影")}/`)).toBe(true);
expect(explicitDst?.includes("华语电影/华语电影")).toBe(false);
const second = await planImport(h.d, request(movie, b, "b"));
expect(second.plan?.sourceConfigId).toBe(b.id);
expect(second.plan?.newBindings[0]?.relativePath).toBe("另一个目录/华语电影/固定片 (2020)");
const mismatch = await planImport(h.d, request(movie, a, "b"));
expect(mismatch.reason).toBe("source_config_mismatch");
expect(mismatch.plan).toBeUndefined();
});
it("automatic category prefixes new binding; changed YAML/target preserves existing binding", async () => {
const automatic = profile({ category: null, target: { root: `lib:${base.rootId}`, path: "自动" } });
await settings([automatic]);
const movie = await seedEntity(h.db, { kind: "movie", title: "动画片", year: 2020 });
await h.db
.insertInto("external_identity")
.values({
media_entity_id: movie,
provider: "tmdb",
entity_type: "movie",
external_id: "987",
status: "accepted",
})
.execute();
const getJson = vi.fn().mockResolvedValue({ genres: [{ id: 16 }], original_language: "zh" });
vi.spyOn(SourcesRuntime.prototype, "tmdb").mockResolvedValue({ getJson } as unknown as TmdbClient);
const plan = await planImport(h.d, { ...request(movie, automatic), dryRun: false });
expect(plan.status).toBe("running");
const record = await h.db
.selectFrom("organization_record")
.select("plan")
.where("id", "=", plan.recordId!)
.executeTakeFirstOrThrow();
expect(record.plan).toMatchObject({
classification: { category: "动画电影" },
newBindings: [{ relativePath: "自动/动画电影/动画片 (2020)" }],
});
expect((await executeRecord(h.d, plan.recordId!)).status).toBe("succeeded");
await settings(
[{ ...automatic, target: { ...automatic.target, path: "新目录" } }],
"movie:\n 新分类: null\ntv: {}\n",
);
const updated = await planImport(h.d, request(movie, automatic));
expect(updated.plan?.classification?.category).toBe("新分类");
expect(updated.plan?.newBindings).toEqual([]);
expect(updated.plan?.organize.files.find((f) => f.role === "main")?.dst).toContain(
join(lib.root, "自动", "动画电影", "动画片 (2020)"),
);
expect(
(
await h.db
.selectFrom("path_binding")
.select("relative_path")
.where("media_entity_id", "=", movie)
.executeTakeFirstOrThrow()
).relative_path,
).toBe("自动/动画电影/动画片 (2020)");
});
it("explicit root creates an unbound path there and refuses to relocate an existing binding", async () => {
const alternate = await createTempTree({}, { sentinel: true });
const alternateRoot = await h.db
.insertInto("library_root")
.values({ name: "alternate", media_kind: "mixed", local_path: alternate.root, management_mode: "managed" })
.returning("id")
.executeTakeFirstOrThrow();
try {
const source = profile();
await settings([source]);
const fresh = await seedEntity(h.db, { kind: "movie", title: "新根片", year: 2020 });
const override = await planImport(h.d, { ...request(fresh, source), targetLibraryRootId: alternateRoot.id });
expect(override.plan?.libraryRootId).toBe(alternateRoot.id);
expect(override.plan?.newBindings[0]?.relativePath).toBe("新根片 (2020)");
const existing = await seedEntity(h.db, { kind: "movie", title: "已绑定片", year: 2020 });
const first = await planImport(h.d, { ...request(existing, source), dryRun: false });
expect((await executeRecord(h.d, first.recordId!)).status).toBe("succeeded");
const attempted = await planImport(h.d, { ...request(existing, source), targetLibraryRootId: alternateRoot.id });
expect(attempted.reason).toBe("binding_on_other_root");
expect(
await h.db
.selectFrom("path_binding")
.select(["library_root_id", "relative_path"])
.where("media_entity_id", "=", existing)
.executeTakeFirstOrThrow(),
).toMatchObject({ library_root_id: base.rootId, relative_path: "华语电影/已绑定片 (2020)" });
} finally {
await h.db.updateTable("library_root").set({ enabled: false }).where("id", "=", alternateRoot.id).execute();
await alternate.cleanup();
}
});
it("tag and source path adopt only matching instance; removing tag stops management", async () => {
const tag = "source-only-tag";
await settings([
profile({
source: { root: `dl:${base.instanceId}:0`, path: "tagged" },
trigger: "qb_tag",
downloaderInstanceId: base.instanceId,
tag,
}),
]);
const hash = (n: number) => n.toString(16).padStart(40, "0");
qb.seed({ hash: hash(1), name: FILE, save_path: cache.path("tagged"), tags: [tag] });
qb.seed({ hash: hash(2), name: "No.Tag.mkv", save_path: cache.path("tagged"), tags: [] });
qb.seed({ hash: hash(3), name: "Wrong.Path.mkv", save_path: cache.path("b"), tags: [tag] });
otherQb.seed({ hash: hash(1), name: FILE, save_path: cache.path("tagged"), tags: [tag] });
const result = await syncDownloads(h.d);
expect(result.adopted).toBe(1);
const rows = await h.db.selectFrom("download_task").selectAll().execute();
expect(rows).toHaveLength(1);
expect(rows[0]).toMatchObject({ instance_id: base.instanceId, torrent_hash: hash(1), managed: true });
expect(rows.some((r) => r.instance_id === otherInstance)).toBe(false);
qb.get(hash(1))!.tags = [];
await syncDownloads(h.d);
expect(
(await h.db.selectFrom("download_task").selectAll().where("id", "=", rows[0]!.id).executeTakeFirstOrThrow())
.managed,
).toBe(false);
qb.get(hash(1))!.tags = [tag];
await syncDownloads(h.d);
expect(
(await h.db.selectFrom("download_task").selectAll().where("id", "=", rows[0]!.id).executeTakeFirstOrThrow())
.managed,
).toBe(true);
});
it("automatic source with movie-only media type refuses a TV entity instead of generic organization", async () => {
const tag = "movie-source";
await settings([
profile({
source: { root: `dl:${base.instanceId}:0`, path: "tagged" },
trigger: "qb_tag",
downloaderInstanceId: base.instanceId,
tag,
}),
]);
const hash = "4".padStart(40, "0");
const name = "Only.TV.S01E01.1080p.WEB-DL.x264-GRP.mkv";
qb.seed({ hash, name, save_path: cache.path("tagged"), tags: [tag] });
await qb.complete(hash, { files: [{ path: name, size: 200_000 }] });
await syncDownloads(h.d);
const row = await h.db
.selectFrom("download_task")
.selectAll()
.where("torrent_hash", "=", hash)
.executeTakeFirstOrThrow();
const entity = await seedEntity(h.db, { kind: "series", title: "Only TV", year: 2020 });
await h.db.updateTable("download_task").set({ media_entity_id: entity }).where("id", "=", row.id).execute();
const planned = await planImport(h.d, {
downloadTaskId: row.id,
trigger: "auto",
actorType: "system",
dryRun: true,
});
expect(planned.reason).toBe("source_config_mismatch");
expect(planned.plan).toBeUndefined();
});
});
@@ -108,6 +108,33 @@ describe.skipIf(skip)("整理记录筛选与统计", () => {
expect(movies.body.data).toMatchObject({ total: 1, succeeded: 1, failed: 0 });
});
it("历史分页和统计排除排队与运行记录,显式active仍可查询", async () => {
const inserted = await h.db
.insertInto("organization_record")
.values([
{ trigger: "manual", actor_type: "user", action: "copy", status: "queued", source_path: "/queued.mkv" },
{ trigger: "manual", actor_type: "user", action: "move", status: "running", source_path: "/running.mkv" },
])
.returning("id")
.execute();
try {
expect(await list("phase=history")).toHaveLength(4);
expect(idsOf(await list("phase=active"))).toEqual(inserted.map((item) => item.id).sort());
expect(await list("phase=history&status=queued")).toHaveLength(0);
const stats = await h.api("GET", "/organization-records/stats?phase=history");
expect(stats.body.data).toMatchObject({ total: 4, running: 0 });
} finally {
await h.db
.deleteFrom("organization_record")
.where(
"id",
"in",
inserted.map((item) => item.id),
)
.execute();
}
});
it("列表项带海报、大小、下载器名称与 hash", async () => {
const [film] = await list("q=Arrival.2016");
expect(film).toMatchObject({
@@ -0,0 +1,440 @@
import { randomUUID } from "node:crypto";
import { existsSync } from "node:fs";
import { mkdir, readFile, writeFile } from "node:fs/promises";
import { dirname } from "node:path";
import { DEFAULT_SETTINGS } from "@umt/contracts";
import { createMockQb, createTempTree, type MockQb, type TempTree } from "@umt/testkit";
import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
import { executeRecord, planImport } from "../services/organize.ts";
import { createHarness, type Harness, pgAvailable, seedBase, seedEntity } from "./harness.ts";
const hasPg = await pgAvailable();
describe.skipIf(!hasPg)("one-time replacement (real PG and library files)", () => {
let h: Harness;
let lib: TempTree;
let src: TempTree;
let rootId: string;
let instanceId: string;
let siteId: string;
let qb: MockQb;
beforeAll(async () => {
lib = await createTempTree({}, { sentinel: true });
src = await createTempTree({
"Movie.2020.1080p.BluRay.x264-GRP.mkv": 200_000,
"Same.2020.1080p.BluRay.x264-GRP.mkv": 200_000,
...Object.fromEntries(
Array.from({ length: 6 }, (_, i) => [
`Show.S01E${String(i + 1).padStart(2, "0")}.1080p.WEB-DL.x264-GRP.mkv`,
200_000,
]),
),
});
qb = await createMockQb();
h = await createHarness("replacement");
({ rootId, instanceId, siteId } = await seedBase(h.db, {
libraryPath: lib.root,
cachePath: src.root,
qbUrl: qb.url,
}));
await h.setSettings("naming", {
tv: "{{title}} ({{year}})/Season {{season}}/{{title}} - {{season_episode}} - {{videoFormat}}{{fileExt}}",
});
}, 60_000);
afterAll(async () => {
await h?.close();
await qb?.close();
await lib?.cleanup();
await src?.cleanup();
});
const old = async (entityId: string, path: string, parent?: string) => {
await mkdir(dirname(lib.path(path)), { recursive: true });
await writeFile(lib.path(path), Buffer.alloc(200_001, 7));
const asset = await h.db
.insertInto("media_asset")
.values({
media_entity_id: entityId,
kind: parent ? "subtitle" : "video",
parent_asset_id: parent ?? null,
status: "active",
size_bytes: 200_001,
})
.returning("id")
.executeTakeFirstOrThrow();
await h.db
.insertInto("asset_location")
.values({
media_asset_id: asset.id,
location_kind: "library",
library_root_id: rootId,
relative_path: path,
status: "available",
is_primary: true,
})
.execute();
return asset.id;
};
const request = (entityId: string, paths: string[], assetIds: string[] = []) => ({
mediaEntityId: entityId,
sourcePaths: paths.map((p) => src.path(p)),
trigger: "manual" as const,
actorType: "user" as const,
action: "copy" as const,
importMode: "replace" as const,
replacementAssetIds: assetIds,
});
const status = async (id: string) =>
(await h.db.selectFrom("media_asset").select("status").where("id", "=", id).executeTakeFirstOrThrow()).status;
it("new movie registers first, retires only selected old and its private subtitle, preserving unselected/shared/cache", async () => {
const e = await seedEntity(h.db, {
kind: "movie",
title: "Movie",
year: 2020,
});
const path = "Movie (2020)/Movie (2020) - 720p.mkv";
const selected = await old(e, path);
const subtitle = await old(e, path.replace(".mkv", ".en.srt"), selected);
const kept = await old(e, "Movie (2020)/Movie (2020) - 2160p.mkv");
await writeFile(lib.path("Movie (2020)/movie.nfo"), "shared");
const planned = await planImport(h.d, request(e, ["Movie.2020.1080p.BluRay.x264-GRP.mkv"], [selected]));
expect(planned.status).toBe("running");
expect(existsSync(lib.path(path))).toBe(true);
expect((await executeRecord(h.d, planned.recordId!)).status).toBe("succeeded");
expect(await status(selected)).toBe("retired");
expect(await status(subtitle)).toBe("retired");
expect(await status(kept)).toBe("active");
expect(existsSync(lib.path(path))).toBe(false);
expect(existsSync(lib.path("Movie (2020)/movie.nfo"))).toBe(true);
expect(existsSync(src.path("Movie.2020.1080p.BluRay.x264-GRP.mkv"))).toBe(true);
expect(await h.db.selectFrom("trash_item").selectAll().execute()).toHaveLength(2);
await executeRecord(h.d, planned.recordId!);
expect(await h.db.selectFrom("trash_item").selectAll().execute()).toHaveLength(2);
});
it("same-path replaces selected old asset with a fresh asset, but normal/unselected cannot overwrite", async () => {
const e = await seedEntity(h.db, {
kind: "movie",
title: "Same",
year: 2020,
});
const selected = await old(e, "Same (2020)/Same (2020) - 1080p.mkv");
const normal = await planImport(h.d, {
...request(e, ["Same.2020.1080p.BluRay.x264-GRP.mkv"]),
importMode: "normal",
dryRun: true,
});
expect(normal.plan?.organize.files[0]?.action).toBe("hold");
const planned = await planImport(h.d, request(e, ["Same.2020.1080p.BluRay.x264-GRP.mkv"], [selected]));
expect(planned.status).toBe("running");
expect((await executeRecord(h.d, planned.recordId!)).status).toBe("succeeded");
expect(await status(selected)).toBe("retired");
const active = await h.db
.selectFrom("media_asset")
.select("id")
.where("media_entity_id", "=", e)
.where("status", "=", "active")
.execute();
expect(active).toHaveLength(1);
expect(active[0]!.id).not.toBe(selected);
});
it("selected grants cannot overwrite an unselected path, and changed old files survive cleanup", async () => {
const e = await seedEntity(h.db, {
kind: "movie",
title: "Protected",
year: 2020,
});
const lower = await old(e, "Protected (2020)/Protected (2020) - 720p.mkv");
const occupied = await old(e, "Protected (2020)/Protected (2020) - 1080p.mkv");
const blocked = await planImport(h.d, {
...request(e, ["Movie.2020.1080p.BluRay.x264-GRP.mkv"], [lower]),
dryRun: true,
});
expect(blocked.plan?.organize.files[0]?.action).toBe("hold");
expect(await status(occupied)).toBe("active");
const other = await seedEntity(h.db, {
kind: "movie",
title: "Stale",
year: 2020,
});
const oldPath = "Stale (2020)/Stale (2020) - 720p.mkv";
const stale = await old(other, oldPath);
const planned = await planImport(h.d, request(other, ["Movie.2020.1080p.BluRay.x264-GRP.mkv"], [stale]));
await writeFile(lib.path(oldPath), "changed-after-plan");
expect((await executeRecord(h.d, planned.recordId!)).status).toBe("succeeded");
expect(await status(stale)).toBe("active");
expect((await readFile(lib.path(oldPath))).toString()).toBe("changed-after-plan");
});
it("same-path failed final probe restores old bytes and asset, even with delete policy", async () => {
const e = await seedEntity(h.db, {
kind: "movie",
title: "Same",
year: 2021,
});
const path = "Same (2021)/Same (2021) - 1080p.mkv";
const selected = await old(e, path);
const savedProbe = h.d.ports.probe;
const savedConfig = h.ctx.config;
h.ctx.config = { ...savedConfig, FS_REPLACE_OLD: "delete" };
h.d.ports.probe = async () => ({
status: "ok",
video: [],
audio: [],
subtitles: [],
});
try {
const planned = await planImport(h.d, request(e, ["Same.2020.1080p.BluRay.x264-GRP.mkv"], [selected]));
expect(planned.status).toBe("running");
expect((await executeRecord(h.d, planned.recordId!)).status).toBe("failed");
expect((await readFile(lib.path(path))).equals(Buffer.alloc(200_001, 7))).toBe(true);
expect(await status(selected)).toBe("active");
expect(
(
await h.db
.selectFrom("asset_location")
.select("status")
.where("media_asset_id", "=", selected)
.executeTakeFirstOrThrow()
).status,
).toBe("available");
await executeRecord(h.d, planned.recordId!);
expect((await readFile(lib.path(path))).length).toBe(200_001);
} finally {
h.d.ports.probe = savedProbe;
h.ctx.config = savedConfig;
}
});
it("old E01-E10 only loses successful new E01-E06; failed E04 and old merged range survive", async () => {
const e = await seedEntity(h.db, {
kind: "series",
title: "Show",
year: 2020,
});
const ids: string[] = [];
for (let i = 1; i <= 10; i++) {
ids.push(await old(e, `Show (2020)/Season 1/Show - S01E${String(i).padStart(2, "0")} - 720p.mkv`));
}
const merged = await old(e, "Show (2020)/Season 1/Show - S01E01-E10 - 480p.mkv");
const saved = h.d.ports.probe;
h.d.ports.probe = async (path) =>
path.includes("E04")
? { status: "ok", video: [], audio: [], subtitles: [] }
: { status: "unknown", reason: "ffprobe_unavailable" };
try {
const paths = Array.from(
{ length: 6 },
(_, i) => `Show.S01E${String(i + 1).padStart(2, "0")}.1080p.WEB-DL.x264-GRP.mkv`,
);
const planned = await planImport(h.d, request(e, paths, [...ids, merged]));
expect(planned.status).toBe("running");
expect((await executeRecord(h.d, planned.recordId!)).status).toBe("partial");
for (let i = 1; i <= 10; i++) {
expect(await status(ids[i - 1]!)).toBe(i <= 6 && i !== 4 ? "retired" : "active");
}
expect(await status(merged)).toBe("active");
} finally {
h.d.ports.probe = saved;
}
});
it("unknown manual task resolves Local, classifies and registers in the same import, then explicit reused-hash replacement uses its new plan", async () => {
const name = "Unknown.2022.1080p.WEB-DL.x264-GRP.mkv";
const hash = "a".repeat(40);
qb.seed({ hash, name, tags: ["umt"], save_path: src.root });
await qb.complete(hash, {
writeFiles: true,
files: [{ path: name, size: 200_000 }],
});
const release = await h.db
.insertInto("source_release")
.values({
site_id: siteId,
site_release_id: randomUUID(),
title: name,
size_bytes: 200_000,
raw: JSON.stringify({}),
resolution_status: "unresolved",
})
.returning("id")
.executeTakeFirstOrThrow();
const makePlan = async (evaluation: object) =>
(
await h.db
.insertInto("acquisition_plan")
.values({
slot_id: null,
source_release_id: release.id,
status: "submitted",
payload_hash: randomUUID(),
idempotency_key: randomUUID(),
evaluation: JSON.stringify(evaluation),
})
.returning("id")
.executeTakeFirstOrThrow()
).id;
const initialPlan = await makePlan({
manual: true,
mediaEntityId: null,
importMode: "normal",
replacementAssetIds: [],
});
const dt = await h.db
.insertInto("download_task")
.values({
instance_id: instanceId,
torrent_hash: hash,
name,
save_path: src.root,
content_path: src.path(name),
managed: true,
tags: ["umt"],
status: "completed",
source_release_id: release.id,
acquisition_plan_id: initialPlan,
import_plan_id: initialPlan,
})
.returning("id")
.executeTakeFirstOrThrow();
const local = await seedEntity(h.db, {
kind: "movie",
title: "本地作品",
year: 2022,
});
await h.db
.updateTable("media_entity")
.set({
metadata: JSON.stringify({
title: {
value: "本地作品",
source: "web",
evidence: { sourceUrl: "https://example.test/detail" },
},
originalLanguage: { value: "zh", source: "web" },
genres: { value: ["剧情"], source: "web" },
}),
})
.where("id", "=", local)
.execute();
const sourceId = randomUUID();
await h.setSettings("organize", {
...DEFAULT_SETTINGS.organize,
minSizeBytes: 100_000,
scrapeAfterImport: false,
sources: [
{
id: sourceId,
name: "Local来源",
enabled: true,
source: { root: `dl:${instanceId}:0`, path: "" },
target: { root: `lib:${rootId}`, path: "" },
mediaType: "movie",
category: null,
action: "copy",
trigger: "manual",
downloaderInstanceId: null,
tag: null,
intervalSec: 300,
},
],
});
const has = vi.spyOn(h.ctx.capabilities, "has").mockReturnValue(true);
const invoke = vi.spyOn(h.ctx.capabilities, "invoke").mockResolvedValue({
mediaEntityId: local,
mode: "local",
enrichmentId: randomUUID(),
reason: null,
});
try {
const planned = await planImport(h.d, {
downloadTaskId: dt.id,
sourceConfigId: sourceId,
trigger: "auto",
actorType: "system",
});
expect(planned.status).toBe("running");
expect(
(
await h.db
.selectFrom("organization_record")
.select("plan")
.where("id", "=", planned.recordId!)
.executeTakeFirstOrThrow()
).plan,
).toMatchObject({ metadataMode: "local" });
expect(invoke).toHaveBeenCalledWith(
"metadata.resolve_organize",
{ sourceReleaseId: release.id },
expect.objectContaining({ source: "task" }),
);
expect((await executeRecord(h.d, planned.recordId!)).status).toBe("succeeded");
expect(
await h.db
.selectFrom("media_asset")
.select("id")
.where("media_entity_id", "=", local)
.where("status", "=", "active")
.execute(),
).toHaveLength(1);
expect(
(
await h.db
.selectFrom("path_binding")
.select("relative_path")
.where("media_entity_id", "=", local)
.executeTakeFirstOrThrow()
).relative_path,
).toBe("华语电影/本地作品 (2022)");
expect(h.engine.enqueued.some((j) => j.task === "scrapeMedia")).toBe(true);
const selected = await old(local, "华语电影/本地作品 (2022)/本地作品 (2022) - 720p.mkv");
const secondPlan = await makePlan({
manual: true,
mediaEntityId: local,
importMode: "replace",
replacementAssetIds: [selected],
});
await h.db
.updateTable("download_task")
.set({ import_plan_id: secondPlan, status: "completed" })
.where("id", "=", dt.id)
.execute();
const rerun = await planImport(h.d, {
downloadTaskId: dt.id,
sourceConfigId: sourceId,
trigger: "auto",
actorType: "system",
});
expect(rerun.status).toBe("running");
expect(
(
await h.db
.selectFrom("organization_record")
.select("plan")
.where("id", "=", rerun.recordId!)
.executeTakeFirstOrThrow()
).plan,
).toMatchObject({ metadataMode: "local" });
expect(invoke).toHaveBeenCalledTimes(1);
await executeRecord(h.d, rerun.recordId!);
expect(await status(selected)).toBe("retired");
expect(
(
await h.db
.selectFrom("acquisition_plan")
.select("status")
.where("id", "=", secondPlan)
.executeTakeFirstOrThrow()
).status,
).toBe("completed");
expect(
(
await h.db
.selectFrom("download_task")
.select("acquisition_plan_id")
.where("id", "=", dt.id)
.executeTakeFirstOrThrow()
).acquisition_plan_id,
).toBe(initialPlan);
} finally {
has.mockRestore();
invoke.mockRestore();
}
});
});
@@ -3,7 +3,7 @@ import { join, relative } from "node:path";
import { createTempTree, type TempTree } from "@umt/testkit";
import { afterAll, beforeAll, describe, expect, it } from "vitest";
import { executeRecord, planImport } from "../services/organize.ts";
import { runScrapeRecord, SCRAPE_TASK, scrapeEntity } from "../services/scrape.ts";
import { entityScrapeLocations, runScrapeRecord, SCRAPE_TASK, scrapeEntity } from "../services/scrape.ts";
import { createHarness, type Harness, pgAvailable, seedBase, seedEntity } from "./harness.ts";
// 刮削(SCR-1 / SCR-2):入库成功后自动对作品目录排 scrapeMedia → 写 NFO 与图片(临时库根);只补缺失 / 全部覆盖;
@@ -299,6 +299,21 @@ describe.skipIf(!hasPg)("刮削:NFO 与图片写进媒体库目录", () => {
expect(nfo).toContain("<runtime>169</runtime>");
}, 60_000);
it("蓝光原盘资产:原盘目录本身就是刮削目录,不取上一级", async () => {
const disc = await seedEntity(h.db, { kind: "movie", title: "原盘", year: 2020 });
const root = await h.db.selectFrom("library_root").select("id").executeTakeFirstOrThrow();
const asset = await h.db
.insertInto("media_asset")
.values({ media_entity_id: disc, kind: "bluray_dir" })
.returning("id")
.executeTakeFirstOrThrow();
await h.db
.insertInto("asset_location")
.values({ media_asset_id: asset.id, library_root_id: root.id, relative_path: "Movies/原盘 (2020)" })
.execute();
expect((await entityScrapeLocations(h.d, disc, "movie")).map((l) => l.path)).toEqual(["Movies/原盘 (2020)"]);
});
it("写入失败记原因;只登记库根不写文件;没有入库文件的作品 412", async () => {
const files = await tree(lib.root);
const video = files.find((p) => p.includes("星际穿越") && p.endsWith(".mkv"))!;
@@ -249,16 +249,19 @@ describe.skipIf(!hasPg)("订阅增强(真实 PG + mock qB + 临时目录)", () =
.transaction()
.execute((trx) => applyEpisodeTotal(trx, { entityId, season: 1, fresh: 3, now: new Date() }));
const r2 = await release(entityId, "Unknown.Total.S01E02.1080p.WEB-DL.x264-GRP");
await completeAndImport(r2.hash, await submit(await planFor(r2.id)));
const acceptedTask = await submit(await planFor(r2.id));
expect(await subRow(subId)).toMatchObject({ status: "completed", status_reason: "download_accepted" });
expect((await slotRow(slotId)).status).toBe("downloading");
await completeAndImport(r2.hash, acceptedTask);
expect((await slotRow(slotId)).status).toBe("satisfied");
expect(await subRow(subId)).toMatchObject({ status: "completed", status_reason: "slot_satisfied" });
expect(await subRow(subId)).toMatchObject({ status: "completed", status_reason: "download_accepted" });
const hist = await db()
.selectFrom("subscription_history")
.selectAll()
.where("subscription_id", "=", subId)
.execute();
expect(hist).toHaveLength(1);
expect(hist[0]).toMatchObject({ reason: "slot_satisfied", media_entity_id: entityId, mode: "once" });
expect(hist[0]).toMatchObject({ reason: "download_accepted", media_entity_id: entityId, mode: "once" });
expect((hist[0]!.progress as { seasons: { complete: boolean }[] }).seasons[0]!.complete).toBe(true);
// 历史 API + 重新订阅(改回 active)
const api = await h.api("GET", `/subscription-history?subscriptionId=${subId}`);
@@ -267,6 +270,32 @@ describe.skipIf(!hasPg)("订阅增强(真实 PG + mock qB + 临时目录)", () =
expect(re.body.data.status).toBe("active");
}, 90_000);
it("多个目标版本按各自已接收单元完成,单版本入库不提前完成订阅", async () => {
const { entityId, subId } = await seriesSlot("双版本剧", { mode: "once" });
await db()
.insertInto("acquisition_slot")
.values({
media_entity_id: entityId,
subscription_id: subId,
name: "720p",
hard_conditions: JSON.stringify({ resolutions: ["720p"] }),
})
.execute();
await db()
.transaction()
.execute((trx) => applyEpisodeTotal(trx, { entityId, season: 1, fresh: 1, now: new Date() }));
const first = await release(entityId, "Two.Versions.S01E01.1080p.WEB-DL.x264-GRP");
const firstTask = await submit(await planFor(first.id));
expect((await subRow(subId)).status).toBe("active");
await completeAndImport(first.hash, firstTask);
expect((await subRow(subId)).status).toBe("active");
const second = await release(entityId, "Two.Versions.S01E01.720p.WEB-DL.x264-GRP");
await submit(await planFor(second.id));
expect(await subRow(subId)).toMatchObject({ status: "completed", status_reason: "download_accepted" });
await qb.complete(second.hash, { writeFiles: true });
await syncDownloads(h.d);
}, 60_000);
it("pending(TMDB 未就绪)照常下载但不完成;暂停的订阅不评估", async () => {
const { entityId, slotId, subId } = await seriesSlot("待定剧", { mode: "once", status: "pending" });
await db()
@@ -532,7 +532,8 @@ export function registerAoCapabilities(registry: CapabilityRegistry, deps: () =>
if (!dt) throw new CapabilityError("not_found", "下载任务不存在");
if (!["completed", "importing", "imported"].includes(dt.status))
throw new CapabilityError("precondition", `下载任务尚未完成(${dt.status})`);
const action: "copy" | "move" = input.action === "move" ? "move" : "copy";
const action: "copy" | "move" | "in_place" =
input.action === "in_place" ? "in_place" : input.action === "move" ? "move" : "copy";
const request = {
downloadTaskId: input.downloadTaskId,
action,
@@ -603,7 +604,7 @@ export function registerAoCapabilities(registry: CapabilityRegistry, deps: () =>
if (!art) throw new CapabilityError("not_found", "整理计划不存在");
if (art.content_hash !== input.payloadHash) throw new CapabilityError("conflict", "payloadHash 与计划不一致");
const content = art.content as {
request: { downloadTaskId: string; action: "copy" | "move"; targetLibraryRootId?: string };
request: { downloadTaskId: string; action: "copy" | "move" | "in_place"; targetLibraryRootId?: string };
};
const req = content.request;
// 已执行过(同计划)→ 返回已有记录
@@ -45,13 +45,7 @@ async function taskRow(
runId,
...(subject ? { subjectType: subject.type, subjectId: subject.id } : {}),
}));
if (given)
await d.ctx.db
.updateTable("task")
.set({ engine_run_id: runId })
.where("id", "=", given)
.where("engine_run_id", "is", null)
.execute();
if (given) await d.ctx.db.updateTable("task").set({ engine_run_id: runId }).where("id", "=", given).execute();
await markTask(d.ctx.db, taskId, "running");
return taskId;
}
@@ -197,7 +191,7 @@ export function registerAoTasks(registry: TaskRegistry, d: AoDeps) {
retryPolicy: { maximumAttempts: 1 },
handler: async (t) =>
t.step.run("scrape", async () => {
const r = await runScrapeRecord(d, t.input.recordId, t.abortSignal);
const r = await runScrapeRecord(d, t.input.recordId, t.abortSignal, t.runId);
return { status: r.status, written: r.written, skipped: r.skipped, failed: r.failed, reason: r.reason };
}),
}),
@@ -7,6 +7,7 @@ import { registerAoCapabilities } from "./capabilities.ts";
import { registerAoTasks } from "./handlers.ts";
import { type AoOptions, type AoPorts, createPorts } from "./ports.ts";
import { registerAssetRoutes } from "./routes/assets.ts";
import { registerClassificationRoutes } from "./routes/classification.ts";
import { registerDashboardRoutes } from "./routes/dashboard.ts";
import { registerDownloadRoutes } from "./routes/downloads.ts";
import { registerLibraryRoutes } from "./routes/library.ts";
@@ -77,6 +78,7 @@ export function createAcquisitionOrganizeModule(overrides: AoOverrides = {}): Se
registerDashboardRoutes(app, d);
registerAssetRoutes(app, d);
registerNamingRoutes(app, d);
registerClassificationRoutes(app, d);
},
registerTasks(registry, ctx) {
registerAoTasks(registry, lazy(ctx));
@@ -31,6 +31,7 @@ import {
createEntityFromTmdbCandidate,
enqueueNewEntityRefresh,
} from "../sources-identity/service/entity-from-candidate.ts";
import { fetchPublicImage } from "../sources-identity/service/image-proxy.ts";
import { metadataLanguages, metadataTextOptions } from "../sources-identity/service/metadata-rich.ts";
import { identifyParsed, type NameIdentifyResult } from "../sources-identity/service/resolve.ts";
import type { ServerContext } from "../types.ts";
@@ -200,6 +201,11 @@ export function createPorts(
return { kind: "url", url: t.url };
},
async scrapeImage(tmdbPath) {
if (/^https?:\/\//i.test(tmdbPath)) {
const got = await fetchPublicImage(externalFetch, tmdbPath, { maxBytes: 8 * 1024 * 1024, label: "封面" });
if (!got.ok) throw new Error(got.message);
return { content: got.bytes, mime: got.mime };
}
images ??= new ImageCache({
dir: join(dataPaths(ctx.config).dataDir, "cache", "images"),
fetch: externalFetch,
@@ -0,0 +1,63 @@
import { Api, type ClassificationConfigView } from "@umt/contracts";
import { sql } from "@umt/db";
import { classifyTmdb, parseClassificationYaml } from "@umt/domain";
import { ApiException } from "../../../http/errors.ts";
import { audit, endpoint } from "../../auth-settings/endpoint.ts";
import { type CachedSettingsReader, SETTINGS_CHANGED_EVENT } from "../../auth-settings/provided.ts";
import type { ModuleApp } from "../../types.ts";
import type { AoDeps } from "../services/common.ts";
const isCached = (s: unknown): s is CachedSettingsReader =>
typeof (s as CachedSettingsReader)?.invalidate === "function";
export function classificationView(yaml: string): ClassificationConfigView {
try {
const config = parseClassificationYaml(yaml);
return { yaml, categories: { movie: config.movie.map((r) => r.category), tv: config.tv.map((r) => r.category) } };
} catch (err) {
throw new ApiException("validation_failed", err instanceof Error ? err.message : String(err));
}
}
export function registerClassificationRoutes(app: ModuleApp, d: AoDeps) {
const settings = async () =>
isCached(d.ctx.settings) ? (await d.ctx.settings.fresh()).organize : await d.ctx.settings.get("organize");
endpoint(app, { ...Api.classification.get, tags: ["classification"] }, async () =>
classificationView((await settings()).classificationYaml),
);
endpoint(app, { ...Api.classification.put, tags: ["classification"] }, async ({ body, actor }) => {
const view = classificationView(body.yaml);
const defaults = await settings();
await d.ctx.db.transaction().execute(async (trx) => {
await sql`select pg_advisory_xact_lock(hashtext('organize.sources'))`.execute(trx);
const row = await trx
.selectFrom("settings")
.select("value")
.where("key", "=", "organize")
.forUpdate()
.executeTakeFirst();
const current = row?.value && typeof row.value === "object" && !Array.isArray(row.value) ? row.value : {};
const next = { ...defaults, ...current, classificationYaml: body.yaml };
await trx
.insertInto("settings")
.values({ key: "organize", value: JSON.stringify(next), updated_at: new Date() })
.onConflict((oc) => oc.column("key").doUpdateSet({ value: JSON.stringify(next), updated_at: new Date() }))
.execute();
});
if (isCached(d.ctx.settings)) d.ctx.settings.invalidate();
void d.ctx.events.publish(SETTINGS_CHANGED_EVENT, { keys: ["organize"] });
await audit(
d.ctx,
actor,
"settings.update",
{ type: "settings", id: "organize" },
{ patch: { organize: { classificationYaml: body.yaml } } },
);
return view;
});
endpoint(app, { ...Api.classification.preview, tags: ["classification"] }, async ({ body }) => {
const yaml = body.yaml ?? (await settings()).classificationYaml;
classificationView(yaml);
return classifyTmdb(parseClassificationYaml(yaml), body.mediaType, body.tmdb);
});
}
@@ -302,7 +302,6 @@ export function registerLibraryRoutes(app: ModuleApp, d: AoDeps) {
]),
)
.orderBy("dt.created_at", "desc")
.limit(100)
.execute();
const entityRecs = await db()
.selectFrom("organization_record")
@@ -321,7 +320,6 @@ export function registerLibraryRoutes(app: ModuleApp, d: AoDeps) {
])
.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
@@ -356,7 +354,6 @@ export function registerLibraryRoutes(app: ModuleApp, d: AoDeps) {
),
)
.orderBy("p.updated_at", "desc")
.limit(50)
.execute()
: [];
const rootPath = new Map(roots.map((r) => [r.id, r.local_path]));
@@ -266,11 +266,13 @@ function recordFilters<Q extends SelectQueryBuilder<DB, "organization_record", a
): Q {
// biome-ignore lint/suspicious/noExplicitAny: kysely 泛型在链式 where 后无法保持 Q
let q = q0 as any;
if (query.phase === "active") q = q.where("status", "in", ["queued", "running"]);
else if (query.phase === "history") q = q.where("status", "not in", ["queued", "running"]);
if (query.trigger) q = q.where("trigger", "=", query.trigger);
if (query.mediaEntityId) q = q.where("media_entity_id", "=", query.mediaEntityId);
if (query.downloadTaskId) q = q.where("download_task_id", "=", query.downloadTaskId);
if (query.since) q = q.where("created_at", ">=", new Date(query.since));
if (query.organizeOnly) q = q.where("action", "in", ["copy", "move", "hardlink"]);
if (query.organizeOnly) q = q.where("action", "in", ["copy", "move", "in_place", "hardlink"]);
if (query.kind === "none") q = q.where("media_entity_id", "is", null);
else if (query.kind)
q = q.where("media_entity_id", "in", (eb: ExpressionBuilder<DB, "organization_record">) =>
@@ -0,0 +1,105 @@
import type { ClassificationPreviewResult } from "@umt/contracts";
import type { MediaEntityRow } from "@umt/db";
import { classifyTmdb, DEFAULT_CLASSIFICATION_YAML, parseClassificationYaml } from "@umt/domain";
import { SourcesRuntime } from "../../sources-identity/runtime.ts";
import type { AoDeps } from "./common.ts";
/** Use accepted TMDB identity and the existing shared raw-detail cache. */
export async function classifyEntity(d: AoDeps, entity: MediaEntityRow): Promise<ClassificationPreviewResult> {
const none: ClassificationPreviewResult = {
category: null,
matchedFields: [],
fallback: false,
};
const config = parseClassificationYaml(
(await d.ctx.settings.get("organize")).classificationYaml ?? DEFAULT_CLASSIFICATION_YAML,
);
const mediaType = entity.kind === "movie" ? "movie" : "tv";
if (!config[mediaType].length) return none;
const identity = await d.ctx.db
.selectFrom("external_identity")
.select("external_id")
.where("media_entity_id", "=", entity.id)
.where("provider", "=", "tmdb")
.where("entity_type", "=", mediaType)
.where("content_unit_id", "is", null)
.where("status", "=", "accepted")
.orderBy("id")
.executeTakeFirst();
if (!identity) {
const metadata = entity.metadata as Record<string, unknown>;
const value = (field: string) => {
const entry = metadata[field];
return entry && typeof entry === "object" && "value" in entry ? (entry as { value: unknown }).value : undefined;
};
if (!Object.values(metadata).some((v) => v && typeof v === "object" && "value" in v)) return none;
const genres = value("genres");
const country = value("originCountry");
const year = value("year") ?? entity.year;
const genreIds: Record<string, number> = {
动画: 16,
动漫: 16,
animation: 16,
儿童: 10762,
kids: 10762,
纪录: 99,
纪录片: 99,
documentary: 99,
综艺: 10764,
reality: 10764,
talk: 10767,
音乐: 10402,
music: 10402,
动作: 28,
action: 28,
冒险: 12,
adventure: 12,
喜剧: 35,
comedy: 35,
犯罪: 80,
crime: 80,
剧情: 18,
drama: 18,
家庭: 10751,
family: 10751,
奇幻: 14,
fantasy: 14,
历史: 36,
history: 36,
恐怖: 27,
horror: 27,
悬疑: 9648,
mystery: 9648,
爱情: 10749,
romance: 10749,
科幻: 878,
"science fiction": 878,
惊悚: 53,
thriller: 53,
战争: 10752,
war: 10752,
西部: 37,
western: 37,
};
const raw = {
original_language: value("originalLanguage"),
origin_country: country,
production_countries: country,
genre_ids: (Array.isArray(genres) ? genres : genres ? [genres] : []).map(
(g) => genreIds[String(g).toLowerCase()] ?? g,
),
...(year ? { release_date: String(year), first_air_date: String(year) } : {}),
};
return classifyTmdb(config, mediaType, raw);
}
if (!/^\d+$/.test(identity.external_id)) return none;
try {
const tmdb = await new SourcesRuntime(d.ctx).tmdb();
if (!tmdb) return none;
const raw = await tmdb.getJson<Record<string, unknown>>(`/${mediaType}/${identity.external_id}`);
return classifyTmdb(config, mediaType, raw);
} catch (err) {
d.ctx.logger.warn({ mediaEntityId: entity.id, err }, "TMDB 分类数据读取失败");
return none;
}
}
@@ -3,7 +3,7 @@ import { type DB, type DownloaderInstanceRow, type Kysely, type LibraryRootRow,
import { type PathMapping, toDownloaderPath, toHostPath } from "@umt/domain";
import type { ServerContext } from "../../types.ts";
import type { AoPorts } from "../ports.ts";
import { hashJson, jsonb } from "../util.ts";
import { errMessage, hashJson, jsonb } from "../util.ts";
// 本模块各服务共用:决策(待处理)、候选评估记录、业务任务行、实例拓扑(路径映射/缓存卷)。
@@ -132,7 +132,7 @@ export async function markTask(
* 巡检补投等需要"同一业务行、按时间槽重投"的场景传 engineKey。
*/
export async function enqueueTracked(
d: AoDeps,
d: Pick<AoDeps, "ctx">,
t: Omit<TaskRowInput, "runId">,
opts: { engineKey?: string } = {},
): Promise<{ taskId: string; runId: string }> {
@@ -145,7 +145,12 @@ export async function enqueueTracked(
traceId: taskId,
},
);
await d.ctx.db.updateTable("task").set({ engine_run_id: runId }).where("id", "=", taskId).execute();
try {
await d.ctx.db.updateTable("task").set({ engine_run_id: runId }).where("id", "=", taskId).execute();
} catch (err) {
// run 已被接受;处理器和下一轮巡检会按 taskId 补回关联。
d.ctx.logger.warn({ taskId, runId, err: { message: errMessage(err) } }, "任务已入队,业务关联暂未写回");
}
return { taskId, runId };
}
@@ -163,13 +168,34 @@ export async function enqueueSweep(
): Promise<{ taskId: string; runId: string } | null> {
const prev = await d.ctx.db
.selectFrom("task")
.select(["engine_run_id"])
.select(["id", "engine_run_id"])
.where("kind", "=", t.kind)
.where("idempotency_key", "=", t.idempotencyKey)
.executeTakeFirst();
if (prev?.engine_run_id) {
const run = await d.ctx.engine.get(prev.engine_run_id).catch(() => null);
if (run && ACTIVE_RUN.has(run.status)) return null;
if (prev) {
try {
// 关联写回失败时,用入队已有的 traceId 找回 run,不随时间槽生成另一个任务。
const runId =
(
await d.ctx.db
.selectFrom("tasks_run_meta")
.select("run_id")
.where("task_name", "=", t.kind)
.where("trace_id", "=", prev.id)
.orderBy("created_at", "desc")
.executeTakeFirst()
)?.run_id ?? prev.engine_run_id;
if (runId) {
const run = await d.ctx.engine.get(runId);
if (prev.engine_run_id !== runId && run) {
await d.ctx.db.updateTable("task").set({ engine_run_id: runId }).where("id", "=", prev.id).execute();
}
if (run && ACTIVE_RUN.has(run.status)) return null;
}
} catch (err) {
d.ctx.logger.warn({ taskId: prev.id, err: { message: errMessage(err) } }, "任务状态查询失败,跳过本轮补投");
return null;
}
}
return enqueueTracked(d, t, { engineKey: `${t.kind}:${t.idempotencyKey}:${slot}` });
}
@@ -250,6 +250,7 @@ export async function buildSubscriptionModels(
: r.dt_status === "completed" || r.dt_status === "importing"
? "verifying"
: "downloading";
if (!r.slot_id) continue;
const list = inflightBySlot.get(r.slot_id) ?? [];
const prev = list.find((x) => x.id === r.id);
if (!prev) list.push({ id: r.id, stage, unitKeys: r.unit_keys });
@@ -167,14 +167,14 @@ export function planWaitOf(status: string, deferred: string | null): PlanWait {
}
/** 等待中的计划 → 面板行 */
async function planItems(db: Exec, rows: { id: string; slot_id: string; source_release_id: string }[]) {
async function planItems(db: Exec, rows: { id: string; slot_id: string | null; source_release_id: string }[]) {
if (!rows.length)
return new Map<string, Omit<DownloadBoardPlanItem, "kind" | "id" | "createdAt" | "purpose" | "status">>();
const ids = rows.map((r) => r.id);
const [extra, approvals, tasks] = await Promise.all([
db
.selectFrom("acquisition_plan as p")
.innerJoin("acquisition_slot as s", "s.id", "p.slot_id")
.leftJoin("acquisition_slot as s", "s.id", "p.slot_id")
.leftJoin("media_entity as e", "e.id", "s.media_entity_id")
.leftJoin("source_release as r", "r.id", "p.source_release_id")
.select([
@@ -39,6 +39,7 @@ import {
downloaderPathOf,
type Exec,
enqueueSweep,
enqueueTracked,
hostPathOf,
type InstanceTopology,
loadInstances,
@@ -46,8 +47,10 @@ import {
timeSlot,
upsertDecision,
} from "./common.ts";
import { matchesQbSource, qbSourceTags } from "./organize-config.ts";
import { hnrSatisfied, retentionElapsed } from "./retention.ts";
import {
afterDownloadAccepted,
afterSlotVerified,
type SlotCompletion,
slotComplete,
@@ -107,10 +110,11 @@ async function verifiedContext(
export async function applySlotEvent(
d: AoDeps,
trx: Exec,
slotId: string,
slotId: string | null,
event: SlotEvent,
extra: { currentAssetId?: string | null; reason?: string; planId?: string | null } = {},
): Promise<boolean> {
if (!slotId) return false;
const slot = await trx
.selectFrom("acquisition_slot")
.selectAll()
@@ -217,10 +221,21 @@ export async function submitDownload(d: AoDeps, planId: string, opts: { runId?:
const db = d.ctx.db;
const plan = await db.selectFrom("acquisition_plan").selectAll().where("id", "=", planId).executeTakeFirst();
if (!plan) return { status: "skipped", reason: "plan_not_found" };
if (plan.status !== "reserved") return { status: "skipped", reason: `plan_${plan.status}` };
if (plan.status !== "reserved") {
if (["submitted", "completed"].includes(plan.status) && plan.status_reason !== "accepted_unconfirmed")
await afterDownloadAccepted(d, plan.id);
return { status: "skipped", reason: `plan_${plan.status}` };
}
const blocked = await automationBlock(d);
if (blocked) return { status: "deferred", reason: blocked };
const ev = (plan.evaluation ?? {}) as { instanceId?: string; volumeKey?: string; savePath?: string };
const ev = (plan.evaluation ?? {}) as {
instanceId?: string;
volumeKey?: string;
savePath?: string;
mediaEntityId?: string | null;
manual?: boolean;
importMode?: string;
};
const topo = (await loadInstances(db)).find((t) => t.inst.id === ev.instanceId);
if (!topo?.savePath) return failPlan(d, plan.id, plan.slot_id, "downloader_unavailable");
@@ -313,16 +328,17 @@ 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,
insert into download_task (instance_id, torrent_hash, infohash_v1, infohash_v2, acquisition_plan_id, import_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}, (select media_entity_id from acquisition_slot where id = ${plan.slot_id}),
${plan.id}, ${ev.manual ? plan.id : null}, ${rel.id}, coalesce((select media_entity_id from acquisition_slot where id = ${plan.slot_id}), ${ev.mediaEntityId ?? rel.media_entity_id}::uuid),
${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),
import_plan_id = coalesce(excluded.import_plan_id, download_task.import_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()
@@ -357,6 +373,7 @@ export async function submitDownload(d: AoDeps, planId: string, opts: { runId?:
d.ctx.logger.warn({ planId: plan.id, err: { message: errMessage(err) } }, "季包选文件失败,按整包下载"),
);
}
if (result.kind !== "accepted_unconfirmed") await afterDownloadAccepted(d, plan.id);
if (!dtId) {
// 账本已有结果(重放):按 hash 找回任务行
dtId = (
@@ -368,6 +385,20 @@ export async function submitDownload(d: AoDeps, planId: string, opts: { runId?:
.executeTakeFirst()
)?.id;
}
if (result.kind === "already_exists" && dtId && ev.manual && ev.importMode === "replace") {
const existing = await db.selectFrom("download_task").select("status").where("id", "=", dtId).executeTakeFirst();
if (existing?.status === "imported") {
await db.updateTable("download_task").set({ status: "completed" }).where("id", "=", dtId).execute();
await enqueueTracked(d, {
kind: "verifyAndImport",
idempotencyKey: `manual-import:${plan.id}`,
input: { downloadTaskId: dtId },
subjectType: "download_task",
subjectId: dtId,
triggeredBy: "user",
});
}
}
return { status: "submitted", reason: result.kind, ...(dtId ? { downloadTaskId: dtId } : {}) };
}
@@ -385,6 +416,7 @@ async function selectPackFiles(
plan: AcquisitionPlanRow,
rel: SourceReleaseRow,
): Promise<void> {
if (!plan.slot_id) return;
const parsed = parsedOf(rel);
const isPack = !parsed.episodes && !parsed.airDate;
const multi = !!parsed.episodes && parsed.episodes.end > parsed.episodes.start;
@@ -465,7 +497,7 @@ async function selectPackFiles(
);
}
async function failPlanTx(d: AoDeps, trx: Exec, planId: string, slotId: string, reason: string) {
async function failPlanTx(d: AoDeps, trx: Exec, planId: string, slotId: string | null, reason: string) {
await trx
.updateTable("acquisition_plan")
.set({ status: "failed", status_reason: reason.slice(0, 500) })
@@ -502,7 +534,7 @@ async function releaseRemovedPlan(d: AoDeps, trx: Exec, planId: string) {
);
}
async function failPlan(d: AoDeps, planId: string, slotId: string, reason: string): Promise<SubmitOutcome> {
async function failPlan(d: AoDeps, planId: string, slotId: string | null, reason: string): Promise<SubmitOutcome> {
await d.ctx.db.transaction().execute((trx) => failPlanTx(d, trx, planId, slotId, reason));
return { status: "failed", reason };
}
@@ -593,9 +625,14 @@ async function syncInstance(d: AoDeps, topo: InstanceTopology, stats: SyncStats)
const db = d.ctx.db;
const now = d.ports.now();
const client = await d.ports.downloader(topo.inst);
const sourceTags = await qbSourceTags(d, topo);
let list: QbTorrent[];
try {
list = await client.list({ tag: topo.inst.managed_tag });
const tags = [...new Set([topo.inst.managed_tag, ...sourceTags.map((s) => s.config.tag!)])];
const batches = await Promise.all(tags.map((tag) => client.list({ tag })));
list = [...new Map(batches.flat().map((t) => [t.hash, t])).values()].filter(
(t) => t.tags.includes(topo.inst.managed_tag) || matchesQbSource(t, topo, sourceTags),
);
} catch (err) {
// 下载器失联:不改任何任务状态(失败 ≠ 不存在)
stats.unreachable.push(topo.inst.id);
@@ -722,6 +759,7 @@ async function observeLocked(
.set({
status: merged.status,
pause_reason: merged.status === "paused" ? merged.pauseReason : null,
managed: true,
raw_state: t.rawState,
progress: Math.min(1, Math.max(0, t.progress)),
size_bytes: sizeOf(t),
@@ -0,0 +1,149 @@
import type { ManualAcquisitionRequest, ManualAcquisitionResult } from "@umt/contracts";
import { sql } from "@umt/db";
import { ApiException } from "../../../http/errors.ts";
import type { ServerContext } from "../../types.ts";
import { hashJson, jsonb, toApi } from "../util.ts";
import { subscriptionSavePath } from "./acquisition.ts";
import { enqueueTracked, loadInstances } from "./common.ts";
/** Manual choice is a one-off plan, independent of subscription filters and slots. */
export async function planManualAcquisition(
ctx: ServerContext,
body: ManualAcquisitionRequest,
): Promise<ManualAcquisitionResult> {
const db = ctx.db;
const rel = await db
.selectFrom("source_release")
.selectAll()
.where("id", "=", body.sourceReleaseId)
.executeTakeFirst();
if (!rel) throw new ApiException("not_found", "资源不存在");
const entityId = body.entityId ?? rel.media_entity_id;
if (!entityId && !body.allowUnrecognized)
return {
submitted: false,
requiresConfirmation: true,
plan: null,
slotId: null,
taskId: null,
rejected: [
{ slotId: null, code: "unrecognized", message: "资源尚未识别作品,确认后可以先下载,整理时补全元数据" },
],
};
if (entityId && !(await db.selectFrom("media_entity").select("id").where("id", "=", entityId).executeTakeFirst()))
throw new ApiException("not_found", "作品不存在");
const instances = await loadInstances(db, body.downloaderInstanceId ? { ids: [body.downloaderInstanceId] } : {});
const topo = instances.find((i) => i.savePath);
if (!topo?.savePath) throw new ApiException("precondition_failed", "没有可用的下载器或保存目录");
const replacementIds = body.importMode === "replace" ? [...new Set(body.replacementAssetIds)] : [];
if (body.importMode === "replace" && (!entityId || replacementIds.length === 0))
throw new ApiException("bad_request", "洗版入库需要明确作品及要替换的已有版本");
if (replacementIds.length) {
const assets = await db
.selectFrom("media_asset")
.select(["id", "media_entity_id"])
.where("id", "in", replacementIds)
.execute();
if (assets.length !== replacementIds.length || assets.some((a) => a.media_entity_id !== entityId))
throw new ApiException("bad_request", "要替换的版本必须属于本次下载的作品");
}
const knownHash = rel.infohash_v1 ?? rel.infohash_v2;
if (knownHash && body.importMode !== "replace") {
const existing = await db
.selectFrom("download_task")
.select(["id", "acquisition_plan_id", "import_plan_id"])
.where("instance_id", "=", topo.inst.id)
.where("torrent_hash", "=", knownHash)
.where("status", "=", "imported")
.where("removed_at", "is", null)
.executeTakeFirst();
if (existing) {
const id = existing.import_plan_id ?? existing.acquisition_plan_id;
const p = id
? await db.selectFrom("acquisition_plan").selectAll().where("id", "=", id).executeTakeFirst()
: undefined;
return {
submitted: true,
requiresConfirmation: false,
plan: p ? (toApi(p) as ManualAcquisitionResult["plan"]) : null,
slotId: p?.slot_id ?? null,
taskId: p?.task_id ?? null,
rejected: [],
};
}
}
const savePath = body.savePath ? subscriptionSavePath(topo, body.savePath) : topo.savePath;
if (!savePath) throw new ApiException("bad_request", "保存目录不在该下载器的路径映射范围内");
const evaluation = {
manual: true,
mediaEntityId: entityId,
instanceId: topo.inst.id,
volumeKey: topo.volumeKey,
savePath,
importMode: body.importMode,
replacementAssetIds: replacementIds,
reasons: ["manual_selection"],
note: body.note ?? null,
};
// Pending requests converge. Completed/failed plans do not prevent another explicit choice.
const key = `manual:${hashJson({ release: rel.id, ...evaluation })}`;
const plan = await db.transaction().execute(async (trx) => {
await sql`select pg_advisory_xact_lock(hashtextextended(${key}, 0))`.execute(trx);
const existing = await trx
.selectFrom("acquisition_plan")
.selectAll()
.where("source_release_id", "=", rel.id)
.where("slot_id", "is", null)
.where("status", "in", ["reserved", "submitted"])
.where("payload_hash", "=", hashJson(evaluation))
.where(sql<boolean>`evaluation->>'instanceId' = ${topo.inst.id}`)
.orderBy("created_at", "desc")
.executeTakeFirst();
if (existing) return existing;
return trx
.insertInto("acquisition_plan")
.values({
slot_id: null,
source_release_id: rel.id,
task_id: null,
purpose: body.importMode === "replace" ? "upgrade" : "fill",
status: "reserved",
priority: "immediate",
evaluation: jsonb(evaluation),
reserved_bytes: rel.size_bytes ?? 0,
budget_reservation: jsonb({}),
payload_hash: hashJson(evaluation),
idempotency_key: `${key}:${crypto.randomUUID()}`,
expires_at: null,
status_reason: "manual_selection",
file_selection: null,
})
.returningAll()
.executeTakeFirstOrThrow();
});
const { taskId } = await enqueueTracked(
{ ctx },
{
kind: "submitDownload",
idempotencyKey: `plan:${plan.id}`,
input: { acquisitionPlanId: plan.id },
subjectType: "acquisition_plan",
subjectId: plan.id,
triggeredBy: "user",
},
);
const updated = await db
.updateTable("acquisition_plan")
.set({ task_id: taskId })
.where("id", "=", plan.id)
.returningAll()
.executeTakeFirstOrThrow();
return {
submitted: true,
requiresConfirmation: false,
plan: toApi(updated) as ManualAcquisitionResult["plan"],
slotId: null,
taskId,
rejected: [],
};
}
@@ -0,0 +1,74 @@
import { join } from "node:path";
import type { OrganizeSourceConfig } from "@umt/contracts";
import { loadRootDefs, within } from "../../file-manager/common.ts";
import type { AoDeps, InstanceTopology } from "./common.ts";
import { hostPathOf } from "./common.ts";
export async function configuredSources(d: AoDeps) {
const settings = await d.ctx.settings.get("organize");
if (!settings.sources?.length) return [];
const roots = await loadRootDefs(d.ctx.db);
return (settings.sources ?? []).flatMap((config) => {
const root = roots.find((r) => r.id === config.source.root);
return root ? [{ config, sourcePath: join(root.hostPath, config.source.path) }] : [];
});
}
export async function resolveOrganizeSource(
d: AoDeps,
input: {
id?: string | undefined;
paths: string[];
mediaType: "movie" | "tv";
instanceId?: string | undefined;
tags?: string[] | undefined;
},
): Promise<OrganizeSourceConfig | null> {
const sources = await configuredSources(d);
const compatible = sources.filter(
({ config, sourcePath }) =>
(config.mediaType === "all" || config.mediaType === input.mediaType) &&
input.paths.length > 0 &&
input.paths.every((p) => within(p, sourcePath)),
);
if (input.id) {
const selected = compatible.find(({ config }) => config.id === input.id);
if (!selected) throw new Error("整理来源不存在,或文件/媒体类型不属于该来源");
return selected.config;
}
// qB automation selects only its configured tag; manual rows require explicit selection.
const selected = sources
.filter(
({ config, sourcePath }) =>
config.enabled &&
config.trigger === "qb_tag" &&
config.downloaderInstanceId === input.instanceId &&
!!config.tag &&
input.tags?.includes(config.tag) &&
input.paths.length > 0 &&
input.paths.every((p) => within(p, sourcePath)),
)
.sort((a, b) => b.sourcePath.length - a.sourcePath.length)[0]?.config;
if (selected && selected.mediaType !== "all" && selected.mediaType !== input.mediaType)
throw new Error("媒体类型与 qB 标签来源配置不匹配");
return selected ?? null;
}
export async function qbSourceTags(d: AoDeps, topology: InstanceTopology) {
return (await configuredSources(d)).filter(
({ config }) =>
config.enabled && config.trigger === "qb_tag" && config.downloaderInstanceId === topology.inst.id && !!config.tag,
);
}
export function matchesQbSource(
t: { tags: string[]; contentPath: string | null; savePath: string; name: string },
topology: InstanceTopology,
sources: Awaited<ReturnType<typeof qbSourceTags>>,
) {
const path = hostPathOf(topology, t.contentPath ?? join(t.savePath, t.name));
return (
!!path &&
sources.some(({ config, sourcePath }) => !!config.tag && t.tags.includes(config.tag) && within(path, sourcePath))
);
}
@@ -135,7 +135,7 @@ export function requestKey(body: {
sourcePaths?: string[] | undefined;
mediaEntityId?: string | undefined;
contentUnitId?: string | undefined;
action: "copy" | "move";
action: "copy" | "move" | "in_place";
targetLibraryRootId?: string | undefined;
}): string {
return canonicalJson({
@@ -0,0 +1,20 @@
import type { AoDeps } from "./common.ts";
/** Count complete successful histories for each local main-video source. */
export async function successfulOrganizeCounts(db: AoDeps["ctx"]["db"], paths: string[]) {
const counts = new Map<string, number>();
if (!paths.length) return counts;
const rows = await db
.selectFrom("organization_file_result as f")
.innerJoin("organization_record as r", "r.id", "f.organization_record_id")
.select(["f.source_path", "r.id"])
.distinct()
.where("r.status", "=", "succeeded")
.where("r.source_path_view", "=", "local")
.where("f.role", "=", "main")
.where("f.status", "=", "succeeded")
.where("f.source_path", "in", paths)
.execute();
for (const row of rows) counts.set(row.source_path, (counts.get(row.source_path) ?? 0) + 1);
return counts;
}
File diff suppressed because it is too large. Load diff
@@ -30,7 +30,7 @@ export async function planWhys(db: Exec, planIds: readonly string[]): Promise<Ma
if (!ids.length) return out;
const plans = await db
.selectFrom("acquisition_plan as p")
.innerJoin("acquisition_slot as s", "s.id", "p.slot_id")
.leftJoin("acquisition_slot as s", "s.id", "p.slot_id")
.leftJoin("source_release as r", "r.id", "p.source_release_id")
.leftJoin("site", "site.id", "r.site_id")
.select([
@@ -100,7 +100,7 @@ export async function planWhys(db: Exec, planIds: readonly string[]): Promise<Ma
if (agent?.agent_run_id) {
const goal = (agent.goal ?? "").trim().replace(/\s+/g, " ");
source = { kind: "agent", runId: agent.agent_run_id, name: goal.length > 40 ? `${goal.slice(0, 40)}…` : goal };
} else if (inv.length) source = { kind: "manual" };
} else if (inv.length || evOf(p).manual === true) source = { kind: "manual" };
else if (sub) {
const title = sub.title ?? sub.title_query ?? "";
source = {
@@ -0,0 +1,502 @@
import { randomUUID } from "node:crypto";
import { mkdir, readdir, realpath, rm } from "node:fs/promises";
import { basename, dirname, join, relative } from "node:path";
import { DEFAULT_ROOT_SENTINEL, type FileFacts } from "@umt/adapters";
import { sql } from "@umt/db";
import { type ReplacementScope, replacementCovers, replacementScope } from "@umt/domain";
import { jsonb } from "../util.ts";
import { type AoDeps, upsertDecision } from "./common.ts";
import { type FrozenPlan, fsRuntime } from "./organize.ts";
export interface ReplacementFile {
assetId: string;
locationId: string;
rootId: string;
rootPath: string;
sentinel: string;
path: string;
facts: Pick<FileFacts, "size" | "mtimeMs" | "ino" | "dev">;
trashId: string;
trashDay: string;
}
export interface ReplacementAsset {
assetId: string;
scope: ReplacementScope;
files: ReplacementFile[];
}
export interface FrozenReplacement {
mode: "replace";
assets: ReplacementAsset[];
}
export async function freezeReplacement(
d: AoDeps,
entityId: string,
mediaType: "movie" | "tv",
ids: string[],
): Promise<FrozenReplacement> {
const selected = [...new Set(ids)];
if (!selected.length) throw new Error("洗版入库必须选择旧版本");
const assets = await d.ctx.db.selectFrom("media_asset").selectAll().where("id", "in", selected).execute();
if (
assets.length !== selected.length ||
assets.some(
(a) => a.media_entity_id !== entityId || a.status !== "active" || !["video", "bluray_dir"].includes(a.kind),
)
) {
throw new Error("旧版本必须是当前作品的有效库内视频资产");
}
const out: ReplacementAsset[] = [];
for (const asset of assets) {
const locations = await d.ctx.db
.selectFrom("asset_location as l")
.innerJoin("library_root as r", "r.id", "l.library_root_id")
.select([
"l.id",
"l.media_asset_id",
"l.relative_path",
"r.id as root_id",
"r.local_path",
"r.sentinel_file",
"r.management_mode",
])
.where("l.media_asset_id", "=", asset.id)
.where("l.location_kind", "=", "library")
.where("l.status", "=", "available")
.execute();
if (!locations.length || locations.some((l) => l.management_mode !== "managed")) {
throw new Error("旧版本必须有可写的库内位置");
}
const units = await d.ctx.db
.selectFrom("asset_unit_coverage as c")
.innerJoin("content_unit as u", "u.id", "c.content_unit_id")
.select(["u.season_number", "u.episode_number"])
.where("c.media_asset_id", "=", asset.id)
.where("c.coverage", "=", "full")
.execute();
const seasons = new Set(units.map((u) => u.season_number).filter((s) => s !== null));
const explicit =
seasons.size === 1
? {
season: [...seasons][0]!,
episodes: units.flatMap((u) => (u.episode_number === null ? [] : [u.episode_number])),
}
: undefined;
const scope = replacementScope(
locations[0]!.relative_path,
mediaType,
explicit?.episodes.length ? explicit : undefined,
);
if (seasons.size > 1) scope.episodes = [];
const files: ReplacementFile[] = [];
for (const loc of locations) {
const children = await d.ctx.db
.selectFrom("media_asset as a")
.innerJoin("asset_location as l", "l.media_asset_id", "a.id")
.select(["a.id as media_asset_id", "l.id", "l.relative_path"])
.where("a.parent_asset_id", "=", asset.id)
.where("a.status", "=", "active")
.where("l.library_root_id", "=", loc.root_id)
.where("l.status", "=", "available")
.execute();
const stem = basename(loc.relative_path).replace(/\.[^.]+$/, "");
const privateChildren = children.filter(
(c) =>
dirname(c.relative_path) === dirname(loc.relative_path) && basename(c.relative_path).startsWith(`${stem}.`),
);
for (const file of [loc, ...privateChildren]) {
const path = join(loc.local_path, file.relative_path);
const st = await fsRuntime(d).proxy.statSafe(path, {
root: loc.local_path,
sentinel: loc.sentinel_file ?? DEFAULT_ROOT_SENTINEL,
});
if (st.kind !== "ok" || !st.facts.isFile || st.facts.isSymbolicLink) {
throw new Error(`旧版本文件不可核验:${file.relative_path}`);
}
files.push({
assetId: file.media_asset_id,
locationId: file.id,
rootId: loc.root_id,
rootPath: loc.local_path,
sentinel: loc.sentinel_file ?? DEFAULT_ROOT_SENTINEL,
path: file.relative_path,
facts: {
size: st.facts.size,
mtimeMs: st.facts.mtimeMs,
ino: st.facts.ino,
dev: st.facts.dev,
},
trashId: randomUUID(),
trashDay: d.ports.now().toISOString().slice(0, 10).replace(/-/g, ""),
});
}
}
out.push({ assetId: asset.id, scope, files });
}
return { mode: "replace", assets: out };
}
/** Called after successful new assets commit; only library locations and private sidecars are eligible. */
export async function cleanupReplacement(d: AoDeps, frozen: FrozenPlan, recordId: string): Promise<void> {
if (!frozen.replacement) return;
const db = d.ctx.db;
const successful = await db
.selectFrom("organization_file_result as f")
.innerJoin("media_asset as a", "a.id", "f.media_asset_id")
.select(["f.source_path", "f.target_path", "f.media_asset_id"])
.where("f.organization_record_id", "=", recordId)
.where("f.role", "=", "main")
.where((eb) =>
eb.or([
eb("f.status", "=", "succeeded"),
eb.and([eb("f.status", "=", "skipped"), eb("f.skip_reason", "=", "already_in_library")]),
]),
)
.where("a.status", "=", "active")
.execute();
const registeredUnits = successful.length
? await db
.selectFrom("asset_unit_coverage as c")
.innerJoin("content_unit as u", "u.id", "c.content_unit_id")
.select(["c.media_asset_id", "u.season_number", "u.episode_number"])
.where(
"c.media_asset_id",
"in",
successful.map((f) => f.media_asset_id!),
)
.where("c.coverage", "=", "full")
.execute()
: [];
const scopes = successful.flatMap((f) => {
const units = registeredUnits.filter(
(u) => u.media_asset_id === f.media_asset_id && u.season_number !== null && u.episode_number !== null,
);
if (units.length) {
return [...new Set(units.map((u) => u.season_number!))].map((season) =>
replacementScope(f.source_path, frozen.mediaType, {
season,
episodes: units.filter((u) => u.season_number === season).map((u) => u.episode_number!),
}),
);
}
const facts = frozen.facts[f.source_path];
return [
replacementScope(
f.source_path,
frozen.mediaType,
facts?.episodes
? {
season: facts.season,
episodes: Array.from(
{ length: facts.episodes.end - facts.episodes.start + 1 },
(_, i) => facts.episodes!.start + i,
),
}
: undefined,
),
];
});
for (const old of frozen.replacement.assets) {
if (successful.some((f) => f.media_asset_id === old.assetId)) continue;
if (!replacementCovers(old.scope, scopes)) {
if (successful.length) {
await upsertDecision(d, db, {
kind: "other",
severity: "info",
title: "旧版本未完整覆盖,保留文件",
summary: "本次成功文件未明确覆盖旧文件的全部季集或 Part;旧文件保留,可核对范围后处理",
subjectType: "media_asset",
subjectId: old.assetId,
dedupeKey: `replacement_scope:${recordId}:${old.assetId}`,
evidence: { recordId, oldScope: old.scope, successfulScopes: scopes },
});
}
continue;
}
for (const file of old.files) {
const opId = `replacement:${recordId}:${file.locationId}`;
const existing = await db
.selectFrom("asset_operation")
.select(["status", "result"])
.where("operation_id", "=", opId)
.executeTakeFirst();
if (existing?.status === "succeeded") continue;
const location = await db
.selectFrom("asset_location")
.select(["status", "media_asset_id", "relative_path"])
.where("id", "=", file.locationId)
.executeTakeFirst();
if (!location || location.media_asset_id !== file.assetId || location.relative_path !== file.path) continue;
// A selected same-path replacement has already been quarantined by copyWithVerify and its old location retired.
if (location.status === "removed") continue;
if (successful.some((f) => frozen.libraryRootId === file.rootId && f.target_path === file.path)) continue;
try {
const fs = fsRuntime(d);
if ((await fs.proxy.checkRoot(file.rootPath, file.sentinel)).kind !== "online") {
throw new Error("旧版本库根离线");
}
const rootReal = await realpath(file.rootPath);
const abs = join(rootReal, file.path);
const parent = await realpath(dirname(abs));
const rel = relative(rootReal, parent);
if (rel.startsWith("..") || file.path.split("/").some((s) => s === "..")) throw new Error("旧文件位置越界");
const trash = join(rootReal, ".umt-trash", file.trashDay, file.trashId, file.path);
const st = await fs.proxy.statSafe(abs, {
root: rootReal,
sentinel: file.sentinel,
});
const moved = await fs.proxy.statSafe(trash);
const resumed =
existing?.status === "started" &&
st.kind === "missing" &&
(fs.replaceOld === "delete" || moved.kind === "ok");
if (
!resumed &&
(st.kind !== "ok" ||
st.facts.isSymbolicLink ||
Object.entries(file.facts).some(([k, v]) => st.facts[k as keyof typeof file.facts] !== v))
) {
throw new Error("旧文件已变化,保留待核对");
}
await db
.insertInto("asset_operation")
.values({
operation_id: opId,
phase: "cleanup",
kind: "delete_target",
status: "started",
organization_record_id: recordId,
media_asset_id: file.assetId,
intent: jsonb({ file, mode: fs.replaceOld }),
started_at: d.ports.now(),
})
.onConflict((oc) => oc.column("operation_id").doUpdateSet({ status: "started" }))
.execute();
if (!resumed) {
if (fs.replaceOld === "delete") await rm(abs);
else {
await mkdir(dirname(trash), { recursive: true });
const result = await fs.proxy.moveSafe(abs, trash);
if (result.status !== "moved") {
throw new Error(`旧文件回收失败:${result.status}`);
}
}
}
await db.transaction().execute(async (trx) => {
if (fs.replaceOld === "trash") {
await trx
.insertInto("trash_item")
.values({
id: file.trashId,
root_id: `lib:${file.rootId}`,
library_root_id: file.rootId,
root_host_path: rootReal,
original_path: file.path,
trash_path: trash,
is_dir: false,
size_bytes: file.facts.size,
actor: "system:organize",
deleted_at: d.ports.now(),
expires_at: new Date(d.ports.now().getTime() + 7 * 86_400_000),
asset_snapshot: jsonb({
source: "one_time_replacement",
recordId,
assets: [
{
assetId: file.assetId,
locationId: file.locationId,
},
],
}),
})
.onConflict((oc) => oc.column("id").doNothing())
.execute();
}
await trx
.updateTable("asset_location")
.set({ status: "removed" })
.where("id", "=", file.locationId)
.execute();
await sql`update media_asset set status='retired' where id=${file.assetId} and not exists(select 1 from asset_location where media_asset_id=${file.assetId} and status <> 'removed')`.execute(
trx,
);
await trx
.updateTable("asset_operation")
.set({
status: "succeeded",
result: jsonb({
disposition: fs.replaceOld,
path: fs.replaceOld === "trash" ? trash : null,
}),
finished_at: d.ports.now(),
})
.where("operation_id", "=", opId)
.execute();
});
} catch (error) {
await upsertDecision(d, db, {
kind: "other",
title: "旧版本清理未完成",
summary: error instanceof Error ? error.message : String(error),
subjectType: "media_asset",
subjectId: file.assetId,
dedupeKey: `replacement_cleanup:${recordId}:${file.locationId}`,
evidence: { recordId, file },
});
}
}
}
}
/** Keep the old same-path file until the new asset has passed final validation. */
export async function settleSamePathReplacement(d: AoDeps, frozen: FrozenPlan, recordId: string): Promise<void> {
if (!frozen.replacement) return;
const results = await d.ctx.db
.selectFrom("organization_file_result")
.selectAll()
.where("organization_record_id", "=", recordId)
.execute();
const bySource = new Map(results.map((f) => [f.source_path, f]));
for (const pf of frozen.organize.files) {
if (pf.action !== "replace") continue;
const old = frozen.replacement.assets.flatMap((a) => a.files).find((f) => join(f.rootPath, f.path) === pf.dst);
if (!old) continue;
const main = bySource.get(pf.mainSrc ?? pf.src);
const file = bySource.get(pf.src);
if (!file) continue;
const opId = `org:${recordId}:${file.ordinal}`;
const op = await d.ctx.db
.selectFrom("asset_operation")
.select("result")
.where("operation_id", "=", opId)
.executeTakeFirst();
const result = op?.result as
| {
replaced?: { disposition: string; path?: string; trashItemId?: string };
rollback?: boolean;
}
| undefined;
const backup = result?.replaced;
if (!backup?.path || backup.disposition !== "trash" || result?.rollback) {
continue;
}
try {
const fs = fsRuntime(d);
if (main?.status === "succeeded" && main.media_asset_id) {
if (fs.replaceOld === "delete") {
await rm(backup.path, { force: true });
if (backup.trashItemId) {
await d.ctx.db.deleteFrom("trash_item").where("id", "=", backup.trashItemId).execute();
}
await d.ctx.db
.updateTable("asset_operation")
.set({
result: jsonb({
...result,
replaced: { ...backup, disposition: "deleted" },
}),
})
.where("operation_id", "=", opId)
.execute();
}
continue;
}
if (main?.status !== "failed") continue;
if ((await fs.proxy.checkRoot(old.rootPath, old.sentinel)).kind !== "online") {
throw new Error("旧版本库根离线,保留回收副本");
}
const rejected = join(old.rootPath, ".umt-trash", old.trashDay, old.trashId, old.path);
const oldBackup = await fs.proxy.statSafe(backup.path);
const target = await fs.proxy.statSafe(pf.dst);
const alreadyRestored =
oldBackup.kind === "missing" &&
target.kind === "ok" &&
Object.entries(old.facts).every(([key, value]) => target.facts[key as keyof typeof old.facts] === value);
if (!alreadyRestored) {
if (
oldBackup.kind !== "ok" ||
oldBackup.facts.size !== old.facts.size ||
oldBackup.facts.ino !== old.facts.ino
) {
throw new Error("旧文件回收副本已变化");
}
if (!file.checksum?.startsWith("sha256:")) {
throw new Error("失败目标缺少全量校验,旧文件留在回收站");
}
if (target.kind === "ok") {
if ((await fs.proxy.sha256(pf.dst)).digest !== file.checksum.slice(7)) {
throw new Error("失败目标已变化,旧文件留在回收站");
}
await mkdir(dirname(rejected), { recursive: true });
const parked = await fs.proxy.moveSafe(pf.dst, rejected);
if (parked.status !== "moved") throw new Error("失败新文件无法暂存");
} else if (target.kind !== "missing" || (await fs.proxy.sha256(rejected)).digest !== file.checksum.slice(7)) {
throw new Error("失败新文件暂存无法核验");
}
const restored = await fs.proxy.moveSafe(backup.path, pf.dst);
if (restored.status !== "moved") {
throw new Error("旧文件恢复未完成,副本保留");
}
}
await d.ctx.db.transaction().execute(async (trx) => {
if (backup.trashItemId) {
await trx.deleteFrom("trash_item").where("id", "=", backup.trashItemId).execute();
}
await trx
.insertInto("trash_item")
.values({
id: old.trashId,
root_id: `lib:${old.rootId}`,
library_root_id: old.rootId,
root_host_path: old.rootPath,
original_path: old.path,
trash_path: rejected,
is_dir: false,
size_bytes: pf.size,
actor: "system:organize",
deleted_at: d.ports.now(),
expires_at: new Date(d.ports.now().getTime() + 7 * 86_400_000),
asset_snapshot: jsonb({ source: "rejected_replacement", recordId }),
})
.onConflict((oc) => oc.column("id").doNothing())
.execute();
await trx
.updateTable("asset_operation")
.set({
result: jsonb({
...result,
rollback: true,
replaced: { ...backup, disposition: "restored" },
}),
})
.where("operation_id", "=", opId)
.execute();
});
} catch (error) {
await upsertDecision(d, d.ctx.db, {
kind: "other",
title: "洗版旧文件恢复待核对",
summary: error instanceof Error ? error.message : String(error),
subjectType: "media_asset",
subjectId: old.assetId,
dedupeKey: `replacement_restore:${recordId}:${old.locationId}`,
evidence: { recordId, old, backup },
});
}
}
}
export async function stagedReplacementBackup(d: AoDeps, file: ReplacementFile): Promise<string | null> {
const stage = join(file.rootPath, ".umt-trash", "replacement-staging", file.trashId);
for (const dir of await readdir(stage).catch(() => [] as string[])) {
const candidate = join(stage, dir, basename(file.path));
const st = await fsRuntime(d).proxy.statSafe(candidate);
if (
st.kind === "ok" &&
!st.facts.isSymbolicLink &&
Object.entries(file.facts).every(([key, value]) => st.facts[key as keyof typeof file.facts] === value)
)
return candidate;
}
return null;
}
@@ -23,7 +23,13 @@ const credits = (v: unknown): Credit[] => (Array.isArray(v) ? (v as Credit[]) :
const person = (c: Credit): NfoPerson | null => {
const name = str(c.name);
if (!name) return null;
return { tmdbId: numOf(c.id), name, role: str(c.role), profilePath: str(c.profilePath) };
const id = numOf(c.id);
return {
tmdbId: id !== null && Number.isInteger(id) && id > 0 ? id : null,
name,
role: str(c.role),
profilePath: str(c.profilePath),
};
};
/** 制片国:ISO 3166-1 代码 → 中文名(Intl);认不出的保留原码 */
@@ -132,7 +138,10 @@ export function nfoEpisode(season: number, episode: number, u: UnitForNfo | unde
/** 图片路径:作品海报 / 背景图,季海报,集剧照 */
export function imagePathsOf(entityMeta: unknown) {
return { posterPath: str(val(entityMeta, "posterPath")), backdropPath: str(val(entityMeta, "backdropPath")) };
return {
posterPath: str(val(entityMeta, "posterPath")) ?? str(val(entityMeta, "posterUrl")),
backdropPath: str(val(entityMeta, "backdropPath")),
};
}
export const unitPosterPath = (m: unknown) => str(val(m, "posterPath"));
export const unitStillPath = (m: unknown) => str(val(m, "stillPath"));
@@ -170,51 +170,75 @@ export async function queueScrape(
(cur.media_entity_id ?? null) === input.match.mediaEntityId &&
((cur.match as ScrapeMatch | null)?.externalId ?? null) === input.match.externalId;
if (cur && sameMatch && cur.engine_run_id) return { recordId: cur.id, taskId: cur.engine_run_id, deduplicated: true };
const rec = await db
.insertInto("scrape_record")
.values({
root_id: input.rootId,
root_path: input.rootPath,
library_root_id: input.libraryRootId,
path: input.path,
media_entity_id: input.match.mediaEntityId,
match: jsonb(input.match),
status: "queued",
trigger: input.trigger,
mode,
organization_record_id: input.organizationRecordId ?? null,
actor_type: input.actor?.type ?? (input.trigger === "auto" ? "system" : null),
actor_id: input.actor?.id ?? null,
})
.returning("id")
.executeTakeFirstOrThrow();
const reused = !!cur && sameMatch;
const rec = reused
? cur
: await db
.insertInto("scrape_record")
.values({
root_id: input.rootId,
root_path: input.rootPath,
library_root_id: input.libraryRootId,
path: input.path,
media_entity_id: input.match.mediaEntityId,
match: jsonb(input.match),
status: "queued",
trigger: input.trigger,
mode,
organization_record_id: input.organizationRecordId ?? null,
actor_type: input.actor?.type ?? (input.trigger === "auto" ? "system" : null),
actor_id: input.actor?.id ?? null,
})
.returning("id")
.executeTakeFirstOrThrow();
let queued: { runId: string; deduplicated: boolean };
try {
const { runId, deduplicated } = await d.ctx.engine.enqueue(
queued = await d.ctx.engine.enqueue(
SCRAPE_TASK,
{ recordId: rec.id },
{ idempotencyKey: input.idempotencyKey ?? `${SCRAPE_TASK}:${rec.id}`, traceId: randomUUID() },
);
if (deduplicated) {
// 同一幂等键已排过(自动刮削重放):这条记录作废,返回已有的
await db.deleteFrom("scrape_record").where("id", "=", rec.id).execute();
const prev = await db
.selectFrom("scrape_record")
.select(["id"])
.where("engine_run_id", "=", runId)
.executeTakeFirst();
return { recordId: prev?.id ?? rec.id, taskId: runId, deduplicated: true };
}
await db.updateTable("scrape_record").set({ engine_run_id: runId }).where("id", "=", rec.id).execute();
return { recordId: rec.id, taskId: runId, deduplicated: false };
} catch (err) {
await db
.updateTable("scrape_record")
.set({ status: "failed", reason: "task_enqueue_failed", finished_at: new Date() })
.where("id", "=", rec.id)
.execute()
.catch(() => {});
// 只有新记录的入队失败才标失败;复用的记录可能已有接受但未关联的 run。
if (!reused)
await db
.updateTable("scrape_record")
.set({ status: "failed", reason: "task_enqueue_failed", finished_at: new Date() })
.where("id", "=", rec.id)
.where("status", "=", "queued")
.execute()
.catch(() => {});
throw err;
}
const { runId, deduplicated } = queued;
let recordId = rec.id;
if (deduplicated) {
// 自动刮削重放时,返回引擎真正持有的记录;确认之前不能删除新记录。
const prev = await db
.selectFrom("scrape_record")
.select("id")
.where("engine_run_id", "=", runId)
.executeTakeFirst();
if (prev) recordId = prev.id;
else {
const run = await d.ctx.engine.get(runId);
const original = (run?.input as { recordId?: unknown } | undefined)?.recordId;
if (typeof original !== "string") throw new Error(`刮削任务缺少记录:${runId}`);
const existing = await db.selectFrom("scrape_record").select("id").where("id", "=", original).executeTakeFirst();
if (!existing) throw new Error(`刮削记录不存在:${original}`);
recordId = existing.id;
}
}
try {
await db.transaction().execute(async (trx) => {
await trx.updateTable("scrape_record").set({ engine_run_id: runId }).where("id", "=", recordId).execute();
if (recordId !== rec.id) await trx.deleteFrom("scrape_record").where("id", "=", rec.id).execute();
});
} catch (err) {
// run 已被接受,保持 queued;处理器开始时补回关联,重试也复用原记录/幂等键。
d.ctx.logger.warn({ recordId, runId, err: { message: errMessage(err) } }, "刮削已入队,记录关联暂未写回");
}
return { recordId, taskId: runId, deduplicated };
}
// ── 作品 → 目录(作品详情「重新刮削」、入库后自动刮削) ──
@@ -267,7 +291,7 @@ export async function entityScrapeLocations(
.selectFrom("media_asset as a")
.innerJoin("asset_location as l", "l.media_asset_id", "a.id")
.innerJoin("library_root as r", "r.id", "l.library_root_id")
.select(["l.relative_path", "r.id as root_id", "r.local_path"])
.select(["a.kind", "l.relative_path", "r.id as root_id", "r.local_path"])
.where("a.media_entity_id", "=", entityId)
.where("a.status", "=", "active")
.where("a.kind", "in", ["video", "bluray_dir"])
@@ -291,7 +315,10 @@ export async function entityScrapeLocations(
!!b.relative_path &&
(l.relative_path === b.relative_path || l.relative_path.startsWith(`${b.relative_path}/`)),
);
const path = bound?.relative_path ?? inferWorkDir(kind, l.relative_path) ?? l.relative_path;
// 原盘目录本身就是作品目录(MP:BDMV 只对原盘根目录刮削)
const path =
bound?.relative_path ??
(l.kind === "bluray_dir" ? l.relative_path : (inferWorkDir(kind, l.relative_path) ?? l.relative_path));
const key = `${l.root_id}\u0000${path}`;
if (!out.has(key))
out.set(key, { rootId: `lib:${l.root_id}`, rootPath: l.local_path, libraryRootId: l.root_id, path });
@@ -335,12 +362,53 @@ export async function scrapeEntity(
return out;
}
/** 整理入库成功后:设置打开时对作品目录自动刮削(失败只记日志,不影响整理结果) */
export async function scrapeAfterImport(d: AoDeps, entityId: string | null, recordId: string): Promise<void> {
/** 整理入库成功后:设置打开或本次 Local 补全要求写本地元数据时,复用作品目录刮削任务。 */
export async function scrapeAfterImport(
d: AoDeps,
entityId: string | null,
recordId: string,
opts: { force?: boolean; enabled?: boolean | undefined } = {},
): Promise<void> {
if (!entityId) return;
if (opts.enabled === false) return;
try {
const s = await scrapeSettings(d);
if (!s.scrapeAfterImport) return;
if (!s.scrapeAfterImport && !opts.force && opts.enabled !== true) return;
const organization = await d.ctx.db
.selectFrom("organization_record")
.select(["action", "target_library_root_id"])
.where("id", "=", recordId)
.executeTakeFirst();
if (organization?.action === "in_place" && organization.target_library_root_id) {
const match = await entityScrapeMatch(d, entityId, "library");
const root = await d.ctx.db
.selectFrom("library_root")
.selectAll()
.where("id", "=", organization.target_library_root_id)
.executeTakeFirst();
if (!match || !root || root.management_mode !== "managed") return;
const files = await d.ctx.db
.selectFrom("organization_file_result")
.select("target_path")
.where("organization_record_id", "=", recordId)
.where("role", "=", "main")
.where("status", "=", "succeeded")
.orderBy("ordinal")
.execute();
for (const file of files)
if (file.target_path)
await queueScrape(d, {
rootId: `lib:${root.id}`,
rootPath: root.local_path,
libraryRootId: root.id,
path: file.target_path,
match,
trigger: "auto",
organizationRecordId: recordId,
idempotencyKey: `${SCRAPE_TASK}:record:${recordId}:${file.target_path}`,
});
return;
}
await scrapeEntity(d, entityId, { trigger: "auto", organizationRecordId: recordId });
} catch (err) {
d.ctx.logger.warn({ entityId, recordId, err: { message: errMessage(err) } }, "入库后自动刮削没排上");
@@ -469,7 +537,12 @@ interface Fields {
episodeUnit: Map<string, UnitForNfo>;
}
export async function runScrapeRecord(d: AoDeps, recordId: string, signal?: AbortSignal): Promise<ScrapeRecord> {
export async function runScrapeRecord(
d: AoDeps,
recordId: string,
signal?: AbortSignal,
runId?: string,
): Promise<ScrapeRecord> {
const db = d.ctx.db;
const rec = await db.selectFrom("scrape_record").selectAll().where("id", "=", recordId).executeTakeFirst();
if (!rec) throw new Error(`刮削记录不存在:${recordId}`);
@@ -477,7 +550,7 @@ export async function runScrapeRecord(d: AoDeps, recordId: string, signal?: Abor
const now = () => d.ports.now();
await db
.updateTable("scrape_record")
.set({ status: "running", started_at: now(), reason: null })
.set({ status: "running", started_at: now(), reason: null, ...(runId ? { engine_run_id: runId } : {}) })
.where("id", "=", recordId)
.execute();
const settings = await scrapeSettings(d);
@@ -5,6 +5,7 @@ import {
episodeKey,
movieCompletion,
type ParsedRelease,
parseRelease,
parseUnitKey,
priorityLadder,
type QualityFacts,
@@ -319,8 +320,19 @@ export async function slotComplete(
>,
mediaType: "movie" | "tv",
newAssets: readonly string[],
opts: { includeSubmitted?: boolean } = {},
): Promise<SlotCompletion> {
const completion = (slot.completion ?? {}) as { kind?: string; cuts?: string[] };
const accepted = opts.includeSubmitted
? await exec
.selectFrom("acquisition_plan as p")
.innerJoin("source_release as r", "r.id", "p.source_release_id")
.select(["p.id as plan_id", "p.unit_keys", "r.parsed", "r.title"])
.where("p.slot_id", "=", slot.id)
.where("p.status", "in", ["submitted", "completed"])
.where("p.status_reason", "!=", "accepted_unconfirmed")
.execute()
: [];
if (mediaType === "movie") {
const assets = newAssets.length
? await exec
@@ -329,14 +341,20 @@ export async function slotComplete(
.where("id", "in", [...newAssets])
.execute()
: [];
const complete = movieCompletion(
completion.cuts as never,
assets.map((a) => ({
const complete = movieCompletion(completion.cuts as never, [
...assets.map((a) => ({
assetId: a.id,
verified: true,
...(a.content_edition ? { cut: a.content_edition as never } : {}),
})),
).complete;
...accepted.map((p) => ({
assetId: p.plan_id,
verified: true,
...(((p.parsed as { cut?: string } | null)?.cut ?? parseRelease(p.title).cut)
? { cut: ((p.parsed as { cut?: string } | null)?.cut ?? parseRelease(p.title).cut) as never }
: {}),
})),
]).complete;
return { complete, seasons: [], suspectedMissing: [] };
}
const sub = slot.subscription_id
@@ -357,6 +375,12 @@ export async function slotComplete(
const u = parseUnitKey(k);
if (u?.episode !== undefined) inflightEps.set(u.season, [...(inflightEps.get(u.season) ?? []), u.episode]);
}
for (const p of accepted)
for (const key of p.unit_keys) {
const unit = parseUnitKey(key);
if (unit?.episode !== undefined)
covered.episodes.set(unit.season, new Set([...(covered.episodes.get(unit.season) ?? []), unit.episode]));
}
const ledgers = await seasonLedgers(exec, slot.media_entity_id);
const seasons = await scopeSeasons(exec, slot, sub, covered.episodes, ledgers);
const range = rangeOf(sub);
@@ -490,7 +514,25 @@ export async function afterSlotVerified(
.executeTakeFirst();
const seasonDone = sub?.media_kind !== "movie" && sub?.season_scope === "current_season";
if (sub && sub.status === "active" && (sub.mode === "once" || seasonDone)) {
await completeSubscription(d, trx, sub, "slot_satisfied", { seasons: input.completion.seasons });
const siblings = await trx
.selectFrom("acquisition_slot")
.selectAll()
.where("subscription_id", "=", sub.id)
.where("id", "!=", slot.id)
.execute();
const others = await Promise.all(
siblings.map((s) =>
slotComplete(
trx,
s,
sub.media_kind === "movie" ? "movie" : "tv",
s.current_asset_id ? [s.current_asset_id] : [],
{ includeSubmitted: true },
),
),
);
if (others.every((p) => p.complete))
await completeSubscription(d, trx, sub, "slot_satisfied", { seasons: input.completion.seasons });
}
}
// 计划结束:Slot 到期重评(忙时延后的发布、连续更新的下一集)
@@ -856,3 +898,46 @@ export async function subscriptionProgress(exec: Exec, subId: string): Promise<S
(sub.media_kind === "movie" || (seasons.length > 0 && seasons.every((s) => s.totalKnown)));
return { subscriptionId: sub.id, status: sub.status, completable, seasons, slots: slotOut };
}
/** MP ordinary subscription progress is settled on confirmed downloader acceptance.
* Reserved/uncertain tasks and season wildcard claims never count as covered episodes.
*/
export async function afterDownloadAccepted(d: AoDeps, planId: string): Promise<void> {
await d.ctx.db.transaction().execute(async (trx) => {
const plan = await trx
.selectFrom("acquisition_plan")
.select(["slot_id"])
.where("id", "=", planId)
.executeTakeFirst();
if (!plan?.slot_id) return;
const slot = await trx.selectFrom("acquisition_slot").selectAll().where("id", "=", plan.slot_id).executeTakeFirst();
if (!slot?.subscription_id) return;
const sub = await trx
.selectFrom("subscription")
.selectAll()
.where("id", "=", slot.subscription_id)
.forUpdate()
.executeTakeFirst();
if (sub?.status !== "active") return;
await touchProgress(trx, { subscriptionId: sub.id }, d.ports.now());
if (sub.mode !== "once" && sub.season_scope !== "current_season") return;
const slots = await trx.selectFrom("acquisition_slot").selectAll().where("subscription_id", "=", sub.id).execute();
if (!slots.length) return;
const progress = await Promise.all(
slots.map((s) =>
slotComplete(
trx,
s,
sub.media_kind === "movie" ? "movie" : "tv",
s.current_asset_id ? [s.current_asset_id] : [],
{ includeSubmitted: true },
),
),
);
if (progress.every((p) => p.complete))
await completeSubscription(d, trx, sub, "download_accepted", {
slots: progress,
seasons: progress.flatMap((p) => p.seasons),
});
});
}