Files
UHD_Media_Tool/apps/server/test/scrape/gen-scrape-layout-mp.py
T
rootandClaude Opus 5.5 87bf33b42b 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

145 lines
5.6 KiB
Python

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