From c1bc389058fe281be9c06e9598f690eb1223d93c Mon Sep 17 00:00:00 2001 From: root Date: Tue, 6 Oct 2026 23:52:39 +0800 Subject: [PATCH] =?UTF-8?q?feat(organize):=20=E5=88=AE=E5=89=8A=E5=86=99?= =?UTF-8?q?=E8=BF=9C=E7=AB=AF=E4=B8=8E=20Alist=20=E7=9B=AE=E5=BD=95?= =?UTF-8?q?=E7=9B=91=E6=8E=A7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 目标在 Alist 时 NFO / 图片(含 clearlogo)本地生成后上传,已存在策略与本地一致 - 整理来源可选 Alist 根(手动 / 目录快照):按 MP 快照语义限深度、按目录修改时间跳过未变子树, 列表失败保留基线,复用两次观察稳定与失败记录 Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Cxwgnn2pbzQ7rXv3rq7fei --- .../acquisition-organize/services/scrape.ts | 19 +- .../modules/file-manager/organize-sources.ts | 13 +- .../modules/file-manager/source-monitor.ts | 69 +++++- .../test/storage/alist-scrape-monitor.test.ts | 213 ++++++++++++++++++ packages/contracts/src/organize-sources.ts | 2 +- 5 files changed, 305 insertions(+), 11 deletions(-) create mode 100644 apps/server/test/storage/alist-scrape-monitor.test.ts diff --git a/apps/server/src/modules/acquisition-organize/services/scrape.ts b/apps/server/src/modules/acquisition-organize/services/scrape.ts index 69951c1f..7a67f10b 100644 --- a/apps/server/src/modules/acquisition-organize/services/scrape.ts +++ b/apps/server/src/modules/acquisition-organize/services/scrape.ts @@ -1,7 +1,7 @@ import { createHash, randomUUID } from "node:crypto"; -import { lstat, mkdir, readdir, readFile, realpath, rename, rm, writeFile } from "node:fs/promises"; +import { mkdir, rename, rm, writeFile } from "node:fs/promises"; import { dirname, isAbsolute, join, relative, resolve } from "node:path"; -import { checkRoot, DEFAULT_ROOT_SENTINEL } from "@umt/adapters"; +import { DEFAULT_ROOT_SENTINEL } from "@umt/adapters"; import { DEFAULT_SETTINGS, type OrganizeSettings, @@ -24,6 +24,15 @@ import { seasonOfDirName, workNfo, } from "@umt/domain"; +import { + isRemotePath, + vlstat as lstat, + vreadFile as readFile, + vrealpath as realpath, + vcheckRoot, + vreaddirents, + vwriteFileAtomic, +} from "../../../storage/vfs.ts"; import { createEntityEpisodeMapper, type EntityEpisodeMapper, @@ -587,6 +596,8 @@ export async function scrapeAfterImport( /** 原子写:同目录临时文件 → rename;目录不存在时建(只在登记目录内) */ async function atomicWrite(abs: string, content: Uint8Array | string): Promise { + // Alist 目标:本地生成后上传到对应目录(同 MP 刮削 _save_file → upload_file),已存在时上传旁名后替换 + if (isRemotePath(abs)) return vwriteFileAtomic(abs, content); await mkdir(dirname(abs), { recursive: true }); const tmp = join(dirname(abs), `.${randomUUID()}.umt-scrape.tmp`); try { @@ -621,7 +632,7 @@ async function pool(items: readonly T[], n: number, fn: (t: T) => Promise< } async function listDir(rootReal: string, rel: string) { - const names = await readdir(join(rootReal, rel), { withFileTypes: true }).catch(() => []); + const names = await vreaddirents(join(rootReal, rel)).catch(() => []); const visible = names.filter((n) => !n.name.startsWith(".")).sort((a, b) => a.name.localeCompare(b.name)); return { videos: visible @@ -948,7 +959,7 @@ export async function runScrapeRecord( .executeTakeFirst(); if (!lr?.enabled) return await fail("scrape_root_offline"); if (lr.management_mode !== "managed") return await fail("scrape_root_read_only"); - const health = await checkRoot(lr.local_path, lr.sentinel_file ?? DEFAULT_ROOT_SENTINEL); + const health = await vcheckRoot(lr.local_path, lr.sentinel_file ?? DEFAULT_ROOT_SENTINEL); if (health.kind !== "online") return await fail("scrape_root_offline"); } const rootReal = await realpath(rec.root_path).catch(() => null); diff --git a/apps/server/src/modules/file-manager/organize-sources.ts b/apps/server/src/modules/file-manager/organize-sources.ts index 87224f15..4a82467c 100644 --- a/apps/server/src/modules/file-manager/organize-sources.ts +++ b/apps/server/src/modules/file-manager/organize-sources.ts @@ -74,14 +74,19 @@ export async function saveOrganizeSource(d: FmDeps, id: string, input: Omit true, - () => false, - )) + (!target.def.storageId && + !(await access(target.abs, constants.W_OK).then( + () => true, + () => false, + ))) ) throw new FilesError("not_writable", "目标必须为可写的纳管媒体库目录"); + // Alist 来源(对照 MP 远程目录监控):只能手动或目录快照触发,qB 标签对网盘没有意义 + if (source.def.storageId && input.trigger === "qb_tag") + throw new FilesError("bad_request", "Alist 来源只能用手动或目录快照触发"); if (input.trigger === "qb_tag") { if (!input.downloaderInstanceId || !input.tag?.trim()) throw new FilesError("bad_request", "标签触发需要下载器和标签"); diff --git a/apps/server/src/modules/file-manager/source-monitor.ts b/apps/server/src/modules/file-manager/source-monitor.ts index 9f0fee84..a36fa6cc 100644 --- a/apps/server/src/modules/file-manager/source-monitor.ts +++ b/apps/server/src/modules/file-manager/source-monitor.ts @@ -3,6 +3,8 @@ import { basename, join } from "node:path"; import { isDownloadComplete } from "@umt/adapters"; import type { FileRef, OrganizeSourceConfig } from "@umt/contracts"; import { sql } from "@umt/db"; +import { storageEntry } from "../../storage/registry.ts"; +import { vreaddirStats } from "../../storage/vfs.ts"; import { hostPathOf, loadInstances, upsertDecision } from "../acquisition-organize/services/common.ts"; import { planImport } from "../acquisition-organize/services/organize.ts"; import { downloadTasksUnder } from "../acquisition-organize/services/takeover.ts"; @@ -15,6 +17,7 @@ import { jsonb, recheck, relOf, + remoteFilesError, resolveIn, rootDef, within, @@ -46,11 +49,17 @@ export function advanceSnapshot(previous: Snapshot | null, observed: Record>; since: number | null }, +) { if (!within(path, config.source.path) && config.source.path !== "") throw new FilesError("path_escape", "扫描路径必须在来源目录内"); const start = await resolveIn(d, config.source.root, path); if (start.isSymlink || !start.lst?.isDirectory()) throw new FilesError("bad_request", "来源必须为目录"); + if (start.def.storageId) return walkRemoteSource(start, prev); const files: Record> = {}; const stack = [start.abs]; let seen = 0; @@ -78,6 +87,59 @@ export async function walkSource(d: FmDeps, config: OrganizeSourceConfig, path = return { files, truncated, rootReal: start.rootReal }; } +/** + * Alist 来源快照(对照 MP StorageBase.snapshot,docs/design/storage-alist.md §8):BFS 列目录,深度 ≤ snapshotMaxDepth; + * 开了「按目录修改时间跳过」时,子目录修改时间不晚于上次扫描就沿用上次快照里该子树的文件(根目录每轮都列,用于发现删除); + * 子目录列表失败保留旧快照(MP「查询异常时保留基线」);根目录列表失败抛错(记 last_error,保留旧快照)。 + * 请求经实例限速器(QPS / 并发),不会打爆网盘。 + */ +async function walkRemoteSource( + start: Awaited>, + prev?: { files: Record>; since: number | null }, +) { + const opts = (await storageEntry(start.def.storageId!)).options; + const files: Record> = {}; + const carry = (relDir: string) => { + for (const [k, v] of Object.entries(prev?.files ?? {})) + if (k === relDir || k.startsWith(`${relDir}/`)) files[k] = { size: v.size, mtimeMs: v.mtimeMs }; + }; + let seen = 0; + let truncated = false; + const queue: { abs: string; depth: number }[] = [{ abs: start.abs, depth: 0 }]; + while (queue.length && !truncated) { + const { abs, depth } = queue.shift()!; + let items: Awaited>; + try { + items = await vreaddirStats(abs); + } catch (e) { + if (depth === 0) throw remoteFilesError(e, start.def); + carry(relOf(start.rootReal, abs)); + continue; + } + for (const it of items) { + if (++seen > SOURCE_SCAN_LIMIT) { + truncated = true; + break; + } + if (it.name.startsWith(".") || isTempName(it.name)) continue; + const child = join(abs, it.name); + const rel = relOf(start.rootReal, child); + if (it.stats.isDirectory()) { + if (depth + 1 >= opts.snapshotMaxDepth) continue; + const unchanged = + opts.snapshotCheckFolderModtime && + prev?.since != null && + it.stats.mtimeMs > 0 && + it.stats.mtimeMs <= prev.since; + if (unchanged) carry(rel); + else queue.push({ abs: child, depth: depth + 1 }); + } else if (kindOfName(it.name) === "video" && Number(it.stats.size) > 0) + files[rel] = { size: Number(it.stats.size), mtimeMs: Number(it.stats.mtimeMs) }; + } + } + return { files, truncated, rootReal: start.rootReal }; +} + /** Query unfiltered QB state once: an untagged download may never have a DB task row. */ async function activeDownloaderPaths(d: FmDeps, config: OrganizeSourceConfig): Promise { const source = await rootDef(d.ctx.db, config.source.root); @@ -220,7 +282,10 @@ export async function scanOrganizeSource(d: FmDeps, input: { sourceId: string }) const now = new Date(); try { const liveActive = await activeDownloaderPaths(d, config); - const scan = await walkSource(d, config); + const scan = await walkSource(d, config, config.source.path, { + files: row?.baseline_at ? ((row.snapshot as unknown as Snapshot).files ?? {}) : {}, + since: row?.baseline_at && row.last_scan_at ? new Date(row.last_scan_at).getTime() : null, + }); if (scan.truncated) throw new FilesError("too_large", "来源超过单次扫描上限，请选择更小的目录"); const baseline = !row?.baseline_at; const advance = advanceSnapshot(baseline ? null : (row!.snapshot as unknown as Snapshot), scan.files); diff --git a/apps/server/test/storage/alist-scrape-monitor.test.ts b/apps/server/test/storage/alist-scrape-monitor.test.ts new file mode 100644 index 00000000..ecb93653 --- /dev/null +++ b/apps/server/test/storage/alist-scrape-monitor.test.ts @@ -0,0 +1,213 @@ +import { randomUUID } from "node:crypto"; +import type { OrganizeSourceConfig } from "@umt/contracts"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { runScrapeRecord, scrapeEntity } from "../../src/modules/acquisition-organize/services/scrape.ts"; +import { saveOrganizeSource } from "../../src/modules/file-manager/organize-sources.ts"; +import { scanOrganizeSource, scanStock } from "../../src/modules/file-manager/source-monitor.ts"; +import { createFmHarness, type FmHarness, pgAvailable, seedEntity } from "../file-manager/harness.ts"; +import { addAlist, addAlistLibrary, type TestAlist, text } from "./alist.ts"; + +// 刮削写远端(NFO / 海报 / 背景 / clearlogo 本地生成后上传;已存在策略与本地一致) +// 与目录监控发现 Alist 新文件(快照两次稳定才整理、深度上限、按目录修改时间跳过未变子树、列表失败保留基线)。 + +const skip = !(await pgAvailable()); +const JPEG = new Uint8Array([0xff, 0xd8, 0xff, 0xe0, 0x00, 0x10, 0x4a, 0x46, 0x49, 0x46, 0x00]); +const PNG = new Uint8Array([0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a]); + +describe.skipIf(skip)("Alist:刮削写远端", () => { + let h: FmHarness; + let a: TestAlist; + let lib: { libraryRootId: string; root: string }; + beforeAll(async () => { + h = await createFmHarness( + "alist_scrape", + { + ao: { + async scrapeImage(p) { + return p.endsWith(".png") ? { content: PNG, mime: "image/png" } : { content: JPEG, mime: "image/jpeg" }; + }, + async scrapeTmdbDetail() { + throw new Error("不应请求 TMDB:本地作品有元数据"); + }, + }, + }, + { runTransfers: false }, + ); + a = await addAlist(h.db, { name: "OL", rootPath: "/s1" }); + lib = await addAlistLibrary(h.db, a, "/s1/movies"); + }); + afterAll(async () => h?.close()); + + const run = async (entity: string, mode: "missing" | "overwrite") => { + const recs = await scrapeEntity(h.module.deps().ao, entity, { trigger: "manual", mode }); + for (const r of recs) await runScrapeRecord(h.module.deps().ao, r.recordId); + return h.db + .selectFrom("scrape_record") + .selectAll() + .where( + "id", + "in", + recs.map((r) => r.recordId), + ) + .execute(); + }; + + it("NFO 与图片(含 clearlogo)写到 Alist 上作品目录;只补缺失跳过已有,全部覆盖改写", async () => { + const film = await seedEntity(h.db, { kind: "movie", title: "星际穿越", year: 2014 }); + await h.db + .updateTable("media_entity") + .set({ + metadata: JSON.stringify({ + overview: { value: "穿越虫洞" }, + posterPath: { value: "/inter.jpg" }, + backdropPath: { value: "/inter-bg.jpg" }, + logoPath: { value: "/inter-logo.png" }, + }), + }) + .where("id", "=", film) + .execute(); + await h.db + .insertInto("external_identity") + .values({ + media_entity_id: film, + provider: "tmdb", + entity_type: "movie", + external_id: "157336", + status: "accepted", + }) + .execute(); + a.mock.put("/s1/movies/星际穿越 (2014)/星际穿越 (2014) - 1080p.mkv", "video"); + const asset = await h.db + .insertInto("media_asset") + .values({ media_entity_id: film, kind: "video", size_bytes: 5 }) + .returning("id") + .executeTakeFirstOrThrow(); + await h.db + .insertInto("asset_location") + .values({ + media_asset_id: asset.id, + location_kind: "library", + library_root_id: lib.libraryRootId, + relative_path: "星际穿越 (2014)/星际穿越 (2014) - 1080p.mkv", + status: "available", + is_primary: true, + }) + .execute(); + + const first = await run(film, "overwrite"); + expect(first[0]!.status, JSON.stringify(first[0])).toBe("succeeded"); + const names = a.mock.listNames("/s1/movies/星际穿越 (2014)"); + expect(names.some((n) => n.endsWith(".nfo"))).toBe(true); + expect(names.some((n) => n.includes("poster"))).toBe(true); + expect(names).toContain("clearlogo.png"); + expect(names.filter((n) => n.includes(".umt-new") || n.endsWith(".uhd-partial"))).toEqual([]); + const nfo = names.find((n) => n.endsWith(".nfo"))!; + expect(text(a.mock.read(`/s1/movies/星际穿越 (2014)/${nfo}`))).toContain("星际穿越"); + + // 用户改过 NFO:只补缺失不动它 + a.mock.put(`/s1/movies/星际穿越 (2014)/${nfo}`, "用户改的"); + const missing = await run(film, "missing"); + expect(missing[0]!.status).toMatch(/succeeded|partial|skipped/); + expect(text(a.mock.read(`/s1/movies/星际穿越 (2014)/${nfo}`))).toBe("用户改的"); + // 全部覆盖:改写回来 + await run(film, "overwrite"); + expect(text(a.mock.read(`/s1/movies/星际穿越 (2014)/${nfo}`))).toContain("星际穿越"); + }); +}); + +describe.skipIf(skip)("Alist:目录监控(远端快照)", () => { + let h: FmHarness; + let a: TestAlist; + let sourceId: string; + beforeAll(async () => { + h = await createFmHarness("alist_monitor", {}, { catalog: true, runTransfers: false }); + a = await addAlist(h.db, { name: "OL", rootPath: "/s1", mockOptions: { token: "t" } }); + a.mock.put("/s1/incoming/Old.Film.2010.1080p.mkv", "old video bytes"); + a.mock.put("/s1/incoming/deep/a/b/c/d/e/Too.Deep.2011.mkv", "too deep"); + const film = await seedEntity(h.db, { kind: "movie", title: "New Film", year: 2021 }); + await h.db + .insertInto("external_identity") + .values({ + media_entity_id: film, + provider: "tmdb", + external_id: "654321", + status: "accepted", + entity_type: "movie", + }) + .execute(); + const config: Omit = { + name: "Alist 来源", + enabled: true, + source: { root: a.root as OrganizeSourceConfig["source"]["root"], path: "incoming" }, + target: { root: h.libRoot as OrganizeSourceConfig["target"]["root"], path: "" }, + mediaType: "all", + category: "欧美电影", + action: "copy", + trigger: "snapshot", + downloaderInstanceId: null, + tag: null, + intervalSec: 60, + }; + sourceId = randomUUID(); + await saveOrganizeSource(h.module.deps(), sourceId, config); + }); + afterAll(async () => h?.close()); + + it("首扫建基线不整理;新文件两次观察稳定后整理;深度超限不进快照", async () => { + const d = h.module.deps(); + const base = await scanOrganizeSource(d, { sourceId }); + expect(base).toMatchObject({ baseline: true, queued: 0 }); + const snap0 = await h.db.selectFrom("organize_source_snapshot").selectAll().executeTakeFirstOrThrow(); + const files0 = Object.keys((snap0.snapshot as { files: Record }).files); + expect(files0).toContain("incoming/Old.Film.2010.1080p.mkv"); + expect(files0.some((f) => f.includes("Too.Deep"))).toBe(false); + + a.mock.put("/s1/incoming/New.Film.2021.tmdb-654321.1080p.WEB-DL.mkv", "new video bytes!!"); + const first = await scanOrganizeSource(d, { sourceId }); + expect(first.eligible).toBe(0); + const second = await scanOrganizeSource(d, { sourceId }); + expect(second.eligible).toBe(1); + expect(second.queued).toBe(1); + const rec = await h.db.selectFrom("organization_record").select(["id", "trigger"]).executeTakeFirstOrThrow(); + expect(rec.trigger).toBe("auto"); + const files = await h.db + .selectFrom("organization_file_result") + .select("source_path") + .where("organization_record_id", "=", rec.id) + .execute(); + expect(files.map((f) => f.source_path)).toContain(a.v("/s1/incoming/New.Film.2021.tmdb-654321.1080p.WEB-DL.mkv")); + expect(h.engine.enqueued.some((e) => e.task === "verifyAndImport")).toBe(true); + }); + + it("按目录修改时间跳过未变子树:子目录没变就不再列", async () => { + const d = h.module.deps(); + a.mock.put("/s1/incoming/sub/Inner.Film.2012.mkv", "inner"); + // sub 目录修改时间早于上次扫描 → 沿用旧快照(新文件要等目录时间变化才会被看到) + a.mock.dirs.set("/s1/incoming/sub", 1); + a.mock.requests.length = 0; + await scanOrganizeSource(d, { sourceId }); + const listed = a.mock.requests.filter((r) => r.path === "/api/fs/list").map((r) => r.body!.path); + expect(listed).toContain("/s1/incoming"); + expect(listed).not.toContain("/s1/incoming/sub"); + // 目录时间更新后再扫就会进去 + a.mock.dirs.set("/s1/incoming/sub", Date.now() + 60_000); + a.mock.requests.length = 0; + await scanOrganizeSource(d, { sourceId }); + expect(a.mock.requests.filter((r) => r.path === "/api/fs/list").map((r) => r.body!.path)).toContain( + "/s1/incoming/sub", + ); + }); + + it("Alist 连不上:记 last_error,保留基线;存量整理也能列出远端文件", async () => { + const d = h.module.deps(); + const before = await h.db.selectFrom("organize_source_snapshot").selectAll().executeTakeFirstOrThrow(); + a.mock.failNext(10); + await scanOrganizeSource(d, { sourceId }); + const after = await h.db.selectFrom("organize_source_snapshot").selectAll().executeTakeFirstOrThrow(); + expect(after.last_error).toBeTruthy(); + expect(after.baseline_at).toEqual(before.baseline_at); + a.mock.failNext(0); + const stock = await scanStock(d, sourceId); + expect(stock.items.map((i) => i.path)).toContain("incoming/Old.Film.2010.1080p.mkv"); + }); +}); diff --git a/packages/contracts/src/organize-sources.ts b/packages/contracts/src/organize-sources.ts index 4e7109a2..154616b8 100644 --- a/packages/contracts/src/organize-sources.ts +++ b/packages/contracts/src/organize-sources.ts @@ -3,7 +3,7 @@ import { Id, JsonObject, Timestamp } from "./common.ts"; // Independent reference shape avoids settings → instances → settings import cycles. const SourceRef = z.object({ - root: z.string().regex(/^(lib:[0-9a-f-]{36}|dl:[0-9a-f-]{36}:\d{1,3})$/i), + root: z.string().regex(/^(lib:[0-9a-f-]{36}|dl:[0-9a-f-]{36}:\d{1,3}|st:[0-9a-f-]{36})$/i), path: z .string() .max(4096)