#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ media_scraper.py —— 自研媒体刮削器（核心库 + 命令行） 流程：接收一个媒体单元（单个电影文件 / 电影目录 / 剧集目录） 1. 从文件名/目录名解析标题、年份、类型（电影/剧集）、季/集 2. 查询 TMDB（电影 search/movie，剧集 search/tv） 3. 把媒体文件「复制」到成品目录（落库） 4. 在成品目录就地生成 Kodi 规范的 nfo + 下载海报/背景图 命令行用法： ./media_scraper.py --source <路径> --dest <成品目录> [--config config.json] [--mock] [--dry-run] [--recurse] 依赖：requests（TMDB 调用）。TMDB API Key 通过 config.json 或环境变量 TMDB_API_KEY 提供。 未配置 Key 时可用 --mock 模式跑通整条管线（用模拟元数据，验证 nfo 生成 + 复制落库）。 """ import argparse import json import os import re import shutil import sys import time import xml.etree.ElementTree as ET from datetime import datetime from urllib.parse import quote import requests # --------------------------------------------------------------------------- # 常量与配置 # --------------------------------------------------------------------------- VIDEO_EXTS = {".mkv", ".mp4", ".mov", ".avi", ".wmv", ".flv", ".ts", ".webm", ".rmvb", ".iso", ".m2ts", ".mts"} SUB_EXTS = {".srt", ".ass", ".sub", ".ssa"} TMDB_IMAGE_BASE = "https://image.tmdb.org/t/p/w500" TMDB_IMAGE_ORIG = "https://image.tmdb.org/t/p/original" DEFAULT_LANGUAGE = "zh-CN" # 刮削文案语言，可按需改 FALLBACK_LANGUAGE = "en-US" class Config: """从 config.json / 环境变量读取配置。""" def __init__(self, path=None): self.tmdb_api_key = os.environ.get("TMDB_API_KEY", "") self.language = DEFAULT_LANGUAGE self.mock = False self.timeout = 20 if path and os.path.isfile(path): data = json.load(open(path, encoding="utf-8")) self.tmdb_api_key = data.get("tmdb_api_key", self.tmdb_api_key) self.language = data.get("language", self.language) self.mock = data.get("mock", self.mock) self.timeout = data.get("timeout", self.timeout) # --------------------------------------------------------------------------- # 文件名解析 # --------------------------------------------------------------------------- def clean_ext(name): return os.path.splitext(name)[1].lower() def parse_media_name(name): """从单个文件名或目录名解析媒体信息。 返回 dict： kind: "movie" | "tv" title: 标题（尽量去除年份/季集标记） year: 年份(int) 或 None season/episode: 剧集时存在 """ base = os.path.basename(name) base = os.path.splitext(base)[0] info = {"kind": "movie", "title": base, "year": None, "season": None, "episode": None} # 剧集标记：S01E02 / s1e2 / 第01集 / E01 / 第 1 季 第 2 集 m = re.search(r"[Ss](\d{1,2})[Ee](\d{1,3})", base) # SxxExx if m: info["kind"] = "tv" info["season"] = int(m.group(1)) info["episode"] = int(m.group(2)) else: m = re.search(r"第\s*(\d{1,2})\s*[集話话]", base) # 第01集 if m: info["kind"] = "tv" info["season"] = 1 info["episode"] = int(m.group(1)) else: m2 = re.search(r"[Ee](\d{1,3})", base) # E01 if m2: info["kind"] = "tv" info["season"] = 1 info["episode"] = int(m2.group(1)) # 年份 m = re.search(r"\((\d{4})\)", base) or re.search(r"(\b19\d{2}\b|\b20\d{2}\b)", base) if m: info["year"] = int(m.group(1)) base = base.replace(f"({m.group(1)})", "") # 去除季集标记，得到标题 base = re.sub(r"[Ss]\d{1,2}[Ee]\d{1,3}", "", base) base = re.sub(r"第\s*\d{1,2}\s*[集話话]", "", base) base = re.sub(r"[Ee]\d{1,3}", "", base) base = re.sub(r"\((\d{4})\)", "", base) base = re.sub(r"(\b19\d{2}\b|\b20\d{2}\b)", "", base) base = re.sub(r"[\s._\-]+", " ", base).strip() info["title"] = base if base else info["title"] return info # --------------------------------------------------------------------------- # TMDB 客户端 # --------------------------------------------------------------------------- class TMDBApi: def __init__(self, config): self.cfg = config self.sess = requests.Session() self.sess.headers.update({ "Authorization": f"Bearer {config.tmdb_api_key}", "Content-Type": "application/json;charset=utf-8", }) self.base = "https://api.themoviedb.org/3" def _get(self, path, params): if not self.cfg.tmdb_api_key: raise RuntimeError("未配置 TMDB API Key（config.json 或环境变量 TMDB_API_KEY）") r = self.sess.get(f"{self.base}/{path}", params=params, timeout=self.cfg.timeout) r.raise_for_status() return r.json() def search(self, kind, title, year=None): path = "search/movie" if kind == "movie" else "search/tv" params = {"query": title, "language": self.cfg.language} if year: params["year" if kind == "movie" else "first_air_date_year"] = year data = self._get(path, params) results = data.get("results") or [] # 优先标题精确匹配 + 年份匹配 exact = [r for r in results if r.get("title" if kind == "movie" else "name", "").lower() == title.lower()] if year and exact: exact = [r for r in exact if str(r.get("release_date" if kind == "movie" else "first_air_date", "") or "") .startswith(str(year))] if exact: return exact[0] if results: return results[0] return None def get_details(self, kind, tmdb_id): path = f"movie/{tmdb_id}" if kind == "movie" else f"tv/{tmdb_id}" return self._get(path, {"language": self.cfg.language}) def get_credits(self, kind, tmdb_id): """电影 cast/crew；剧集 cast。用于 nfo 的演员、导演、制片厂字段。""" path = f"movie/{tmdb_id}/credits" if kind == "movie" else f"tv/{tmdb_id}/credits" return self._get(path, {"language": self.cfg.language}) def image_url(self, path, size="w500"): if not path: return None base = TMDB_IMAGE_BASE if size == "w500" else TMDB_IMAGE_ORIG return f"{base}{path}" # --------------------------------------------------------------------------- # nfo 生成 # --------------------------------------------------------------------------- def _esc(s): if s is None: return "" return str(s).replace("&", "&").replace("<", "<").replace(">", ">") def _txt(node, tag, value): if value is not None and str(value).strip(): ET.SubElement(node, tag).text = str(value).strip() def build_movie_nfo(meta): """meta: TMDB movie details 的整理字段。返回 movie.nfo XML 字符串。""" root = ET.Element("movie") _txt(root, "title", meta.get("title")) _txt(root, "originaltitle", meta.get("original_title")) _txt(root, "year", meta.get("year")) _txt(root, "rating", meta.get("vote_average")) _txt(root, "votes", meta.get("vote_count")) _txt(root, "runtime", meta.get("runtime")) _txt(root, "plot", meta.get("overview")) _txt(root, "tagline", meta.get("tagline")) _txt(root, "premiered", meta.get("release_date")) _txt(root, "release", meta.get("release_date")) for g in meta.get("genres", []) or []: _txt(root, "genre", g) for d in meta.get("directors", []) or []: _txt(root, "director", d) for s in meta.get("studios", []) or []: _txt(root, "studio", s) _txt(root, "mpaa", meta.get("certification")) if meta.get("tmdb_id"): ET.SubElement(root, "uniqueid", attrib={"type": "tmdb"}).text = str(meta["tmdb_id"]) if meta.get("imdb_id"): ET.SubElement(root, "uniqueid", attrib={"type": "imdb"}).text = str(meta["imdb_id"]) if meta.get("poster_path"): ET.SubElement(root, "thumb").text = TMDB_IMAGE_BASE + meta["poster_path"] if meta.get("fanart_path"): f = ET.SubElement(root, "fanart") ET.SubElement(f, "thumb").text = TMDB_IMAGE_ORIG + meta["fanart_path"] for a in meta.get("cast", []) or []: ac = ET.SubElement(root, "actor") _txt(ac, "name", a.get("name")) _txt(ac, "role", a.get("role")) if a.get("thumb"): _txt(ac, "thumb", TMDB_IMAGE_BASE + a["thumb"]) return '\n' + ET.tostring(root, encoding="unicode") def build_tvshow_nfo(meta): root = ET.Element("tvshow") _txt(root, "title", meta.get("name")) _txt(root, "originaltitle", meta.get("original_name")) _txt(root, "year", meta.get("year")) _txt(root, "rating", meta.get("vote_average")) _txt(root, "votes", meta.get("vote_count")) _txt(root, "plot", meta.get("overview")) _txt(root, "premiered", meta.get("first_air_date")) _txt(root, "status", meta.get("status")) for g in meta.get("genres", []) or []: _txt(root, "genre", g) for s in meta.get("studios", []) or []: _txt(root, "studio", s) if meta.get("tmdb_id"): ET.SubElement(root, "uniqueid", attrib={"type": "tmdb"}).text = str(meta["tmdb_id"]) if meta.get("imdb_id"): ET.SubElement(root, "uniqueid", attrib={"type": "imdb"}).text = str(meta["imdb_id"]) if meta.get("poster_path"): ET.SubElement(root, "thumb").text = TMDB_IMAGE_BASE + meta["poster_path"] for a in meta.get("cast", []) or []: ac = ET.SubElement(root, "actor") _txt(ac, "name", a.get("name")) _txt(ac, "role", a.get("role")) if a.get("thumb"): _txt(ac, "thumb", TMDB_IMAGE_BASE + a["thumb"]) return '\n' + ET.tostring(root, encoding="unicode") def build_episode_nfo(meta): root = ET.Element("episodedetails") _txt(root, "title", meta.get("name")) _txt(root, "season", meta.get("season_number")) _txt(root, "episode", meta.get("episode_number")) _txt(root, "plot", meta.get("overview")) _txt(root, "aired", meta.get("air_date")) _txt(root, "rating", meta.get("vote_average")) if meta.get("still_path"): ET.SubElement(root, "thumb").text = TMDB_IMAGE_BASE + meta["still_path"] return '\n' + ET.tostring(root, encoding="unicode") # --------------------------------------------------------------------------- # 刮削器 # --------------------------------------------------------------------------- class Scraper: def __init__(self, config, dest, source_root=None, dry_run=False, log=None): self.cfg = config self.dest = os.path.abspath(dest) self.source_root = os.path.abspath(source_root) if source_root else None self.dry_run = dry_run self.log = log or (lambda msg: print(msg)) self.api = None if config.mock else TMDBApi(config) def logi(self, msg): ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S") self.log(f"[{ts}] {msg}") # ---- 复制媒体到成品目录 ---- def _relative(self, path): """以源根为基准计算相对路径；不在源根下则退回 basename。""" path = os.path.abspath(path) if self.source_root and os.path.commonpath([path, self.source_root]) == self.source_root: return os.path.relpath(path, self.source_root) return os.path.basename(path) def _copy_media(self, src_path, rel_src): """把源媒体文件/目录复制到成品目录，返回成品里的完整路径。 安全保护：目标路径绝不允许等于源路径，否则直接报错，绝不误删源。 """ dst = os.path.join(self.dest, rel_src) if os.path.abspath(dst) == os.path.abspath(src_path): raise RuntimeError(f"目标路径与源路径相同，已终止以防误删: {src_path}") if self.dry_run: self.logi(f"[DRY] 将复制 {src_path} -> {dst}") return dst os.makedirs(os.path.dirname(dst), exist_ok=True) if os.path.isdir(src_path): if os.path.exists(dst): shutil.rmtree(dst) shutil.copytree(src_path, dst) else: shutil.copy2(src_path, dst) return dst # ---- 电影刮削（单文件或电影目录）---- def scrape_movie(self, src): info = parse_media_name(src) rel = self._relative(src) dst = self._copy_media(src, rel) base = os.path.basename(dst) base_noext = os.path.splitext(base)[0] if self.cfg.mock: meta = self._mock_movie_meta(info) else: found = self.api.search("movie", info["title"], info.get("year")) if not found: raise RuntimeError(f"TMDB 未匹配到电影: {info['title']}") details = self.api.get_details("movie", found["id"]) try: credits = self.api.get_credits("movie", found["id"]) except Exception: credits = {} meta = self._shape_movie_meta(details, credits) if self.dry_run: self.logi(f"[DRY] 电影 {base} 匹配 TMDB id={meta.get('tmdb_id')} " f"《{meta.get('title')}》({meta.get('year')})") return media_file = self._find_media_file(dst) out_dir = os.path.dirname(media_file) if media_file else dst nfo_name = os.path.splitext(os.path.basename(media_file))[0] if media_file else base_noext nfo_path = os.path.join(out_dir, nfo_name + ".nfo") with open(nfo_path, "w", encoding="utf-8") as f: f.write(build_movie_nfo(meta)) self.logi(f"生成 {nfo_path}") self._download_art(meta, out_dir, nfo_name, base_noext) # ---- 剧集刮削（剧集目录）---- def scrape_tv(self, src): info = parse_media_name(src) rel = self._relative(src) dst = self._copy_media(src, rel) if self.cfg.mock: meta = self._mock_tv_meta(info) else: found = self.api.search("tv", info["title"], info.get("year")) if not found: raise RuntimeError(f"TMDB 未匹配到剧集: {info['title']}") details = self.api.get_details("tv", found["id"]) try: credits = self.api.get_credits("tv", found["id"]) except Exception: credits = {} meta = self._shape_tv_meta(details, credits) if self.dry_run: self.logi(f"[DRY] 剧集 {os.path.basename(src)} 匹配 TMDB id={meta.get('tmdb_id')} " f"《{meta.get('name')}》") return # 剧集目录：在目录根生成 tvshow.nfo + poster；对每个视频文件生成 {文件}.nfo root_dir = dst if os.path.isdir(dst) else os.path.dirname(dst) with open(os.path.join(root_dir, "tvshow.nfo"), "w", encoding="utf-8") as f: f.write(build_tvshow_nfo(meta)) self.logi(f"生成 {root_dir}/tvshow.nfo") self._download_art(meta, root_dir, "poster", "poster") for vf in self._find_video_files(root_dir): if info.get("season") is not None and info.get("episode") is not None: ep_meta = self._episode_meta(meta, info["season"], info["episode"]) else: ep_meta = self._episode_meta_from_filename(vf) vbase = os.path.splitext(os.path.basename(vf))[0] with open(os.path.join(root_dir, vbase + ".nfo"), "w", encoding="utf-8") as f: f.write(build_episode_nfo(ep_meta)) self.logi(f"生成 {os.path.join(root_dir, vbase + '.nfo')}") # ---- 工具方法 ---- def _find_media_file(self, path): if os.path.isfile(path): return path for root, _, files in os.walk(path): for fn in sorted(files): if clean_ext(fn) in VIDEO_EXTS: return os.path.join(root, fn) return None def _find_video_files(self, root): out = [] for root2, _, files in os.walk(root): for fn in sorted(files): if clean_ext(fn) in VIDEO_EXTS: out.append(os.path.join(root2, fn)) return out def _download_art(self, meta, out_dir, prefix, poster_prefix=None): """下载海报/背景图（Kodi/Emby 兼容命名）。""" poster_prefix = poster_prefix or prefix poster = meta.get("poster_path") fanart = meta.get("fanart_path") if poster: url = TMDB_IMAGE_BASE + poster self._dl(url, os.path.join(out_dir, f"{poster_prefix}-poster.jpg")) if fanart: url = TMDB_IMAGE_ORIG + fanart self._dl(url, os.path.join(out_dir, f"{prefix}-fanart.jpg")) def _dl(self, url, out_path): if self.dry_run or self.cfg.mock: self.logi(f"[DRY/MOCK] 下载图片 {url} -> {out_path}") return try: r = requests.get(url, timeout=self.cfg.timeout) r.raise_for_status() with open(out_path, "wb") as f: f.write(r.content) self.logi(f"下载图片 {out_path}") except Exception as e: self.logi(f"下载图片失败 {url}: {e}") # ---- meta 整形 ---- def _shape_movie_meta(self, d, credits=None): credits = credits or {} return { "title": d.get("title") or d.get("original_title"), "original_title": d.get("original_title"), "year": (d.get("release_date") or "")[:4], "vote_average": d.get("vote_average"), "vote_count": d.get("vote_count"), "runtime": d.get("runtime"), "overview": d.get("overview"), "tagline": d.get("tagline"), "release_date": d.get("release_date"), "genres": [g.get("name") for g in (d.get("genres") or [])], "certification": None, "tmdb_id": d.get("id"), "imdb_id": d.get("imdb_id"), "poster_path": d.get("poster_path"), "fanart_path": (d.get("backdrop_path")), "directors": [c.get("name") for c in (credits.get("crew") or []) if c.get("job") == "Director"][:3], "studios": [c.get("name") for c in (d.get("production_companies") or [])], "cast": [{"name": c.get("name"), "role": c.get("character"), "thumb": c.get("profile_path")} for c in (credits.get("cast") or [])[:12]], } def _shape_tv_meta(self, d, credits=None): credits = credits or {} return { "name": d.get("name") or d.get("original_name"), "original_name": d.get("original_name"), "year": (d.get("first_air_date") or "")[:4], "vote_average": d.get("vote_average"), "vote_count": d.get("vote_count"), "overview": d.get("overview"), "first_air_date": d.get("first_air_date"), "status": d.get("status"), "genres": [g.get("name") for g in (d.get("genres") or [])], "tmdb_id": d.get("id"), "imdb_id": None, "poster_path": d.get("poster_path"), "fanart_path": (d.get("backdrop_path")), "studios": [c.get("name") for c in (d.get("production_companies") or [])], "cast": [{"name": c.get("name"), "role": c.get("character"), "thumb": c.get("profile_path")} for c in (credits.get("cast") or [])[:12]], } def _episode_meta(self, tv_meta, season, episode): return { "name": f"第 {episode} 集", "season_number": season, "episode_number": episode, "overview": None, "air_date": None, "vote_average": None, "still_path": None, } def _episode_meta_from_filename(self, vf): info = parse_media_name(vf) return self._episode_meta({}, info.get("season") or 1, info.get("episode") or 1) # ---- mock 元数据（无 Key 时验证管线用）---- def _mock_movie_meta(self, info): return { "title": info["title"], "original_title": info["title"], "year": info.get("year") or 2024, "vote_average": 8.5, "vote_count": 100, "runtime": 120, "overview": f"（MOCK）电影《{info['title']}》简介", "tagline": "", "release_date": f"{info.get('year') or 2024}-01-01", "genres": ["剧情"], "certification": None, "tmdb_id": 1000, "imdb_id": "tt0000000", "poster_path": None, "fanart_path": None, } def _mock_tv_meta(self, info): return { "name": info["title"], "original_name": info["title"], "year": info.get("year") or 2024, "vote_average": 8.8, "vote_count": 200, "overview": f"（MOCK）剧集《{info['title']}》简介", "first_air_date": f"{info.get('year') or 2024}-01-01", "status": "Returning Series", "genres": ["剧情"], "tmdb_id": 2000, "imdb_id": None, "poster_path": None, "fanart_path": None, } # --------------------------------------------------------------------------- # 命令行入口 # --------------------------------------------------------------------------- def scan_and_scrape(source, dest, scraper): """处理一个源路径：文件直接刮，目录则识别为电影/剧集目录。""" source = os.path.abspath(source) if os.path.isfile(source): info = parse_media_name(source) if info["kind"] == "tv": scraper.scrape_tv(source) else: scraper.scrape_movie(source) elif os.path.isdir(source): # 判断是剧集目录（含 SxxExx / 季结构）还是电影目录 vids = [f for f in _walk_files(source) if clean_ext(f) in VIDEO_EXTS] tv_mark = any(re.search(r"[Ss]\d{1,2}[Ee]\d{1,3}|第\s*\d{1,2}\s*[集話话]", f) for f in _walk_files(source)) subdir_seasons = any(re.match(r"[Ss]eason\s*\d+|第\s*\d+\s*季", d) for d in os.listdir(source)) if tv_mark or subdir_seasons or len(vids) > 1: scraper.scrape_tv(source) else: scraper.scrape_movie(source) else: raise RuntimeError(f"源路径不存在: {source}") def _walk_files(root): out = [] for r, _, files in os.walk(root): for fn in files: out.append(os.path.join(r, fn)) return out def main(): ap = argparse.ArgumentParser(description="自研媒体刮削器（TMDB）") ap.add_argument("--source", required=True, help="媒体文件/目录（可重复，或 --recurse 扫整目录）") ap.add_argument("--dest", required=True, help="成品落库目录") ap.add_argument("--config", default=None, help="config.json 路径") ap.add_argument("--mock", action="store_true", help="无 Key 时用模拟元数据跑通管线") ap.add_argument("--dry-run", action="store_true", help="只打印计划不落盘") ap.add_argument("--recurse", action="store_true", help="递归处理 --source 下所有新媒体单元") ap.add_argument("--log", default=None, help="日志文件路径（默认 stdout）") args = ap.parse_args() cfg = Config(args.config) if args.mock: cfg.mock = True log = (lambda msg: open(args.log, "a", encoding="utf-8").write(msg + "\n") and print(msg)) if args.log else print if args.recurse: source_root = os.path.abspath(args.source) else: source_root = os.path.abspath(os.path.dirname(args.source)) scraper = Scraper(cfg, args.dest, source_root=source_root, dry_run=args.dry_run, log=log) os.makedirs(args.dest, exist_ok=True) if args.recurse: targets = _walk_files(args.source) # 只处理视频文件，交给 parse 判定 targets = [t for t in targets if clean_ext(t) in VIDEO_EXTS] else: targets = [args.source] for t in targets: try: scan_and_scrape(t, args.dest, scraper) except Exception as e: scraper.logi(f"跳过 {t}: {e}") scraper.logi("全部处理完成") if __name__ == "__main__": main()