import { createHash, randomUUID } from "node:crypto"; import { mkdir, readFile, stat, writeFile } from "node:fs/promises"; import { join, resolve } from "node:path"; import type { ProbeResult } from "@umt/adapters"; import type { FileProbeItem, FilesProbeRequest, FilesProbeResult, FilesScrapeBatchRequest, FilesScrapeBatchResult, FilesScrapeRequest, FilesScrapeResult, } from "@umt/contracts"; import { sql } from "@umt/db"; import type { Principal } from "../../http/auth.ts"; import { FilesError, type FmDeps, jsonb, resolveIn } from "./common.ts"; import { audit } from "./ledger.ts"; import { kindOfName, mediaSummary } from "./list.ts"; import { entitiesUnder } from "./records.ts"; // probe:ffprobe 摘要;已登记资产写回 media_asset.media_info.probe,未登记只缓存 24h(DATA_DIR/cache/probe)。 // scrape:只对已登记作品,等价 POST /media-entities/{id}/refresh(入队 refreshMetadata,字段锁优先、不动路径); // NFO/图片投影 W5 实现(TODO),请求时返回 not_implemented。 const CACHE_TTL_MS = 24 * 3600_000; export async function probeFiles(d: FmDeps, body: FilesProbeRequest): Promise { const items: FileProbeItem[] = []; const cacheDir = join(resolve(d.options.dataDir), "cache", "probe"); for (const it of body.items) { const r = await resolveIn(d, it.root, it.path); const base = { root: it.root as FileProbeItem["root"], path: r.rel }; if (!r.lst?.isFile() && !r.isSymlink) { items.push({ ...base, status: "unknown", reason: "not_a_file", summary: null, container: null, subtitleTracks: 0, savedToAsset: false, cached: false, }); continue; } if (kindOfName(r.abs) !== "video") { items.push({ ...base, status: "unknown", reason: "not_video", summary: null, container: null, subtitleTracks: 0, savedToAsset: false, cached: false, }); continue; } const st = await stat(r.abs); const key = createHash("sha1").update(`${r.abs}|${st.size}|${st.mtimeMs}`).digest("hex"); const cacheFile = join(cacheDir, `${key}.json`); let result: ProbeResult | null = null; let cached = false; try { const c = JSON.parse(await readFile(cacheFile, "utf8")) as { at: number; result: ProbeResult }; if (Date.now() - c.at < CACHE_TTL_MS) { result = c.result; cached = true; } } catch { // 无缓存 } result ??= await d.ao.ports.probe(r.abs); let saved = false; if (result.status === "ok" && r.def.libraryRootId) { const upd = await sql<{ id: string }>` update media_asset a set media_info = a.media_info || jsonb_build_object('probe', ${jsonb(result)}::jsonb) from asset_location l where l.media_asset_id = a.id and l.library_root_id = ${r.def.libraryRootId} and l.relative_path = ${r.rel} and l.status <> 'removed' returning a.id`.execute(d.ctx.db); saved = upd.rows.length > 0; } if (result.status === "ok" && !saved && !cached) { await mkdir(cacheDir, { recursive: true }).catch(() => undefined); const tmp = `${cacheFile}.${randomUUID()}.tmp`; await writeFile(tmp, JSON.stringify({ at: Date.now(), result })) .then(() => Deno.rename(tmp, cacheFile)) .catch(() => undefined); } items.push({ ...base, status: result.status, reason: result.status === "ok" ? null : result.reason, summary: result.status === "ok" ? mediaSummary({ probe: result }) : null, container: result.status === "ok" ? (result.container ?? null) : null, subtitleTracks: result.status === "ok" ? result.subtitles.length : 0, savedToAsset: saved, cached, }); } return { items }; } export async function scrape( d: FmDeps, principal: Principal | undefined, body: FilesScrapeRequest, ): Promise { const db = d.ctx.db; const e = await db .selectFrom("media_entity") .select(["id", "status", "field_locks"]) .where("id", "=", body.mediaEntityId) .executeTakeFirst(); if (!e) throw new FilesError("not_found", "作品不存在"); const registered = await db .selectFrom("media_asset") .select(["id"]) .where("media_entity_id", "=", e.id) .where("status", "=", "active") .executeTakeFirst(); if (!registered) throw new FilesError("bad_request", "作品没有已登记资产。先识别并整理入库,再刮削"); const { runId } = await d.ctx.engine.enqueue( "refreshMetadata", { mediaEntityId: e.id }, { idempotencyKey: `refreshMetadata:files:${body.idempotencyKey}`, traceId: randomUUID() }, ); const overrides = await db .selectFrom("field_override") .select(["field"]) .where("media_entity_id", "=", e.id) .where("locked", "=", true) .execute(); const lockedFields = [...new Set([...(e.field_locks ?? []), ...overrides.map((o) => o.field)])].sort(); // TODO(W5):writeNfo/images → 把最终字段投影成 NFO/海报写到绑定目录(临时文件 + 原子提交;非本系统 NFO 不覆盖) const result: FilesScrapeResult = { mediaEntityId: e.id, taskId: runId, lockedFields, nfo: body.writeNfo ? "not_implemented" : "not_requested", images: body.images ? "not_implemented" : "not_requested", }; await audit(d, principal, "scrape", { ...result }); return result; } /** * G5 批量刷新元数据:文件/目录归到其下已登记资产的作品,与显式作品合并去重,逐个入队 refreshMetadata。 * 只对有 active 资产的作品提交;其余进 skipped(前端汇总"已提交 n 个作品,锁定字段不变")。 */ export async function scrapeBatch( d: FmDeps, principal: Principal | undefined, body: FilesScrapeBatchRequest, ): Promise { const db = d.ctx.db; const skipped: FilesScrapeBatchResult["skipped"] = []; const ids = new Set(body.mediaEntityIds ?? []); for (const it of body.items ?? []) { const r = await resolveIn(d, it.root, it.path); const found = r.def.libraryRootId ? await entitiesUnder(db, r.def.libraryRootId, r.rel, !!r.lst?.isDirectory()) : []; if (found.length === 0) skipped.push({ ref: `${it.root}:${it.path}`, reason: "no_entity" }); for (const id of found) ids.add(id); } if (ids.size === 0 && skipped.length === 0) throw new FilesError("bad_request", "需要 items 或 mediaEntityIds"); if (ids.size > 200) throw new FilesError("bad_request", `一次最多刷新 200 个作品(本次 ${ids.size})`); const list = [...ids]; const ents = list.length ? await db.selectFrom("media_entity").select(["id", "title", "field_locks"]).where("id", "in", list).execute() : []; const active = new Set( list.length ? ( await db .selectFrom("media_asset") .select(["media_entity_id"]) .where("media_entity_id", "in", list) .where("status", "=", "active") .execute() ).map((a) => a.media_entity_id) : [], ); const locks = list.length ? await db .selectFrom("field_override") .select(["media_entity_id", "field"]) .where("media_entity_id", "in", list) .where("locked", "=", true) .execute() : []; const submitted: FilesScrapeBatchResult["submitted"] = []; for (const id of list) { const e = ents.find((x) => x.id === id); if (!e) { skipped.push({ ref: id, reason: "not_found" }); continue; } if (!active.has(id)) { skipped.push({ ref: id, reason: "not_registered" }); continue; } const { runId } = await d.ctx.engine.enqueue( "refreshMetadata", { mediaEntityId: id }, { idempotencyKey: `refreshMetadata:files-batch:${body.idempotencyKey}:${id}`, traceId: randomUUID() }, ); const lockedFields = [ ...new Set([...(e.field_locks ?? []), ...locks.filter((l) => l.media_entity_id === id).map((l) => l.field)]), ].sort(); submitted.push({ mediaEntityId: id, title: e.title, taskId: runId, lockedFields }); } await audit(d, principal, "scrape_batch", { submitted: submitted.map((s) => s.mediaEntityId), skipped, }); return { submitted, skipped }; }