feat(scrape): 任意目录就地刮削,按路径识别作品,每次刮削一条记录(SCR-2)

- POST /files/scrape 改为对目录 / 视频文件刮削:已登记用作品,否则按路径识别(本地 + TMDB), 认不准返回候选,选定后再提交;未登记直接用 TMDB 字段,addToLibrary 顺手建作品 - 作品详情重新刮削、入库后自动刮削改为对作品目录刮削,同一实现 - 迁移 0058_scrape_record 替代 media_scrape;GET /files/scrape-records - 与 MP ScrapingChain 的文件布局差分(7 个场景,3 处登记偏离) Co-Authored-By: Claude Opus 5.5 <[email protected]>
2026-10-01 00:08:17 +08:00
1 parent f60254e265
commit 87bf33b42b
@@ -0,0 +1,144 @@
+"""SCR-2 刮削差分(文件布局):用 MP 上游 ScrapingChain.scrape_metadata 的真实决策代码,在临时目录里对同一组目录结构
+"就地刮削",记录 MP 写了哪些文件,写 scrape-layout-mp.json。与 apps/server/test/file-manager/scrape-layout.test.ts 对照。
+
+用法(MP 仓库根目录,用其 .venv):
+  cd /opt/moviepilot-plus/upstream/MoviePilot && PYTHONPATH=. \
+  .venv/bin/python /opt/uhd_media_tool/apps/server/test/scrape/gen-scrape-layout-mp.py
+
+只记录行为输出,不复制 MP 源码。替身:
+- 存储:本地目录的最小实现(list_files / get_parent_item / get_file_item / upload_file / contains_bluray_subdirectories);
+- NFO 内容:固定占位(内容差分见 gen-scrape-mp.py);
+- 图片清单:按 MP themoviedb 模块 get_metadata_img 的规则给(作品 poster + backdrop;季 seasonNN-poster;集 episode-thumb);
+- 识别:recognize_media 原样返回传入的作品(路径识别不在此差分范围)。
+刮削策略用 MP 缺省(ScrapingConfig 默认全部 missingonly);手动刮削 overwrite=True(同 MP /media/scrape 接口)。
+"""
+
+import json
+import pathlib
+import shutil
+import tempfile
+from types import SimpleNamespace
+
+import app.chain.scraping as scraping_mod
+from app.chain.scraping import ScrapingChain, ScrapingConfig
+from app.domain.context import MediaInfo
+from app.schemas.types import MediaSource, MediaType
+from app.schemas.workflow import FileItem
+
+HERE = pathlib.Path(__file__).parent
+VIDEO_EXT = (".mkv", ".mp4", ".ts", ".m2ts", ".avi", ".iso")
+
+
+def item(p: pathlib.Path) -> FileItem:
+    return FileItem(
+        storage="local",
+        type="dir" if p.is_dir() else "file",
+        path=str(p) + ("/" if p.is_dir() else ""),
+        name=p.name,
+        basename=p.stem if p.is_file() else p.name,
+        extension=p.suffix[1:] if p.is_file() else None,
+    )
+
+
+class LocalStorage:
+    def list_files(self, fileitem, recursion=False):
+        return [item(c) for c in sorted(pathlib.Path(fileitem.path).iterdir())]
+
+    def get_parent_item(self, fileitem):
+        return item(pathlib.Path(fileitem.path.rstrip("/")).parent)
+
+    def get_file_item(self, storage, path):
+        p = pathlib.Path(path)
+        return item(p) if p.exists() else None
+
+    def upload_file(self, fileitem, path, new_name=None):
+        target = pathlib.Path(fileitem.path) / (new_name or path.name)
+        shutil.copyfile(path, target)
+        return item(target)
+
+    def contains_bluray_subdirectories(self, files):
+        return any(f.type == "dir" and f.name.upper() in ("BDMV", "VIDEO_TS") for f in files)
+
+
+def make_chain() -> ScrapingChain:
+    c = object.__new__(ScrapingChain)
+    c.storagechain = LocalStorage()
+    c.scraping_policies = ScrapingConfig()
+    c.runtime_config = SimpleNamespace(
+        video_extensions=VIDEO_EXT, audio_extensions=(".flac", ".mp3"), season_zero_names=("Specials", "SPs")
+    )
+    c.metadata_nfo = lambda **kw: "<nfo/>"
+
+    def metadata_img(mediainfo, season=None, episode=None):
+        if season is not None:
+            if episode:
+                return {"episode-thumb.jpg": "https://img/still.jpg"}
+            return {f"season{str(season).rjust(2, '0')}-poster.jpg": "https://img/season.jpg"}
+        return {"poster.jpg": "https://img/poster.jpg", "backdrop.jpg": "https://img/backdrop.jpg"}
+
+    c.metadata_img = metadata_img
+
+    def save_image(fileitem, path, url):
+        pathlib.Path(path).write_bytes(b"jpg")
+
+    c._download_and_save_image = save_image
+    return c
+
+
+class FakeMediaChain:
+    def recognize_media(self, meta=None, **kw):
+        return CURRENT["media"]
+
+    def recognize_by_meta(self, meta):
+        return CURRENT["media"]
+
+
+CURRENT: dict = {}
+scraping_mod.MediaChain = FakeMediaChain  # 集文件重新识别季集后仍是同一作品
+
+
+def media(kind: MediaType, tmdb_id: int, title: str) -> MediaInfo:
+    mi = MediaInfo()
+    mi.type = kind
+    mi.tmdb_id = tmdb_id
+    mi.media_source = MediaSource.TMDB
+    mi.media_id = str(tmdb_id)
+    mi.title = title
+    return mi
+
+
+SHOW = "人生切割术 (2022)"
+EP1 = f"{SHOW}/Season 1/人生切割术 - S01E01 - 有关地狱的好消息.mkv"
+EP2 = f"{SHOW}/Season 1/人生切割术 - S01E02 - 半循环.mkv"
+MOVIE = "星际穿越 (2014)"
+MOVIE_FILE = f"{MOVIE}/星际穿越 (2014) - 1080p.mkv"
+LOOSE = "Interstellar.2014.1080p.BluRay.mkv"
+TREE = [EP1, EP2, MOVIE_FILE, LOOSE, "Heat (1995)/BDMV/index.bdmv"]
+
+SCENARIOS = {
+    "show_dir": (SHOW, MediaType.TV),
+    "season_dir": (f"{SHOW}/Season 1", MediaType.TV),
+    "episode_file": (EP2, MediaType.TV),
+    "movie_dir": (MOVIE, MediaType.MOVIE),
+    "movie_file": (MOVIE_FILE, MediaType.MOVIE),
+    "movie_file_in_root": (LOOSE, MediaType.MOVIE),
+    "bluray_dir": ("Heat (1995)", MediaType.MOVIE),
+}
+
+out = {}
+for name, (target, kind) in SCENARIOS.items():
+    root = pathlib.Path(tempfile.mkdtemp(prefix="mp-scr-"))
+    for f in TREE:
+        (root / f).parent.mkdir(parents=True, exist_ok=True)
+        (root / f).write_bytes(b"video")
+    before = {str(p.relative_to(root)) for p in root.rglob("*") if p.is_file()}
+    CURRENT["media"] = media(kind, 95396 if kind == MediaType.TV else 157336, "x")
+    chain = make_chain()
+    ok, msg = chain.scrape_metadata(fileitem=item(root / target), mediainfo=CURRENT["media"], overwrite=True)
+    after = {str(p.relative_to(root)) for p in root.rglob("*") if p.is_file()}
+    out[name] = {"target": target, "ok": ok, "written": sorted(after - before)}
+    shutil.rmtree(root)
+
+(HERE / "scrape-layout-mp.json").write_text(json.dumps(out, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
+print(json.dumps(out, ensure_ascii=False, indent=2))
+# 生成后在本仓库跑一次格式化(lint 要求):deno run -A npm:@biomejs/biome check --write apps/server/test/scrape/scrape-layout-mp.json