"""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: "" 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