feat(organize): 刮削写远端与 Alist 目录监控

- 目标在 Alist 时 NFO / 图片(含 clearlogo)本地生成后上传,已存在策略与本地一致
- 整理来源可选 Alist 根(手动 / 目录快照):按 MP 快照语义限深度、按目录修改时间跳过未变子树,
  列表失败保留基线,复用两次观察稳定与失败记录

Co-Authored-By: Claude Opus 5.5 <[email protected]>
Claude-Session: https://claude.ai/code/session_01Cxwgnn2pbzQ7rXv3rq7fei

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<void> {
+  // 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<T, R>(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);

apps/server/src/modules/file-manager/organize-sources.ts

@@ -74,14 +74,19 @@ export async function saveOrganizeSource(d: FmDeps, id: string, input: Omit<Orga
   const target = await resolveIn(d, input.target.root, input.target.path);
   if (
     target.isSymlink ||
+    target.def.kind !== "library" ||
     !target.lst?.isDirectory() ||
     !(await isWritableTarget(target.def, target.rootReal)) ||
+    (!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", "标签触发需要下载器和标签");

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<stri
   return { snapshot: { files }, eligible };
 }
 
-export async function walkSource(d: FmDeps, config: OrganizeSourceConfig, path = config.source.path) {
+export async function walkSource(
+  d: FmDeps,
+  config: OrganizeSourceConfig,
+  path = config.source.path,
+  prev?: { files: Record<string, Omit<SnapshotFile, "accepted">>; 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<string, Omit<SnapshotFile, "accepted">> = {};
   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<ReturnType<typeof resolveIn>>,
+  prev?: { files: Record<string, Omit<SnapshotFile, "accepted">>; since: number | null },
+) {
+  const opts = (await storageEntry(start.def.storageId!)).options;
+  const files: Record<string, Omit<SnapshotFile, "accepted">> = {};
+  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<ReturnType<typeof vreaddirStats>>;
+    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<string[]> {
   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);

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}`, "<movie>用户改的</movie>");
+    const missing = await run(film, "missing");
+    expect(missing[0]!.status).toMatch(/succeeded|partial|skipped/);
+    expect(text(a.mock.read(`/s1/movies/星际穿越 (2014)/${nfo}`))).toBe("<movie>用户改的</movie>");
+    // 全部覆盖:改写回来
+    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<OrganizeSourceConfig, "id"> = {
+      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<string, unknown> }).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");
+  });
+});

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)