- 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]>
145 lines
5.6 KiB
Python
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
|