187 lines
6.1 KiB
Python
187 lines
6.1 KiB
Python
|
#!/usr/bin/env python3
|
|||
|
|
# -*- coding: utf-8 -*-
|
|||
|
|
"""
|
|||
|
|
watcher.py —— 自研媒体刮削「监听服务」
|
|||
|
|
|
|||
|
|
功能：
|
|||
|
|
1. 用 watchdog 监听「清理后媒体目录」
|
|||
|
|
2. 检测到新媒体文件/目录写入稳定后，自动调用 media_scraper 刮削
|
|||
|
|
3. 刮削结果（媒体 + nfo + 海报）复制落库到成品目录
|
|||
|
|
4. 源目录保持原样，不写入任何文件
|
|||
|
|
|
|||
|
|
用法：
|
|||
|
|
./watcher.py --config config.json # 常驻服务
|
|||
|
|
./watcher.py --config config.json --once # 只扫描一次现有内容后退出（用于初始化存量）
|
|||
|
|
|
|||
|
|
config.json 字段：
|
|||
|
|
{
|
|||
|
|
"source": "<清理后媒体目录>",
|
|||
|
|
"dest": "<成品落库目录>",
|
|||
|
|
"tmdb_api_key": "", # TMDB API Key；留空时用 --mock 跑通管线
|
|||
|
|
"mock": true, # true=模拟元数据(无Key验证用)；true后配好Key改为 false
|
|||
|
|
"settle": 6, # 写入稳定等待秒数
|
|||
|
|
"language": "zh-CN"
|
|||
|
|
}
|
|||
|
|
"""
|
|||
|
|
import argparse
|
|||
|
|
import json
|
|||
|
|
import os
|
|||
|
|
import sys
|
|||
|
|
import threading
|
|||
|
|
import time
|
|||
|
|
|
|||
|
|
from watchdog.events import FileSystemEventHandler
|
|||
|
|
from watchdog.observers import Observer
|
|||
|
|
|
|||
|
|
# 复用核心刮削器
|
|||
|
|
from media_scraper import (Config, Scraper, clean_ext, VIDEO_EXTS,
|
|||
|
|
scan_and_scrape)
|
|||
|
|
|
|||
|
|
|
|||
|
|
class Config2(Config):
|
|||
|
|
"""扩展 Config：加载 watcher 的 source/dest/settle。"""
|
|||
|
|
|
|||
|
|
def __init__(self, path):
|
|||
|
|
super().__init__(path)
|
|||
|
|
data = json.load(open(path, encoding="utf-8"))
|
|||
|
|
self.source = os.path.abspath(data.get("source", ""))
|
|||
|
|
self.dest = os.path.abspath(data.get("dest", ""))
|
|||
|
|
self.settle = int(data.get("settle", 6))
|
|||
|
|
self.mock = data.get("mock", False)
|
|||
|
|
if self.mock:
|
|||
|
|
self.tmdb_api_key = "" # mock 模式不需要 key
|
|||
|
|
|
|||
|
|
|
|||
|
|
class ScraperWorker:
|
|||
|
|
"""把新媒体单元交给刮削器，带去重与写入稳定判定。"""
|
|||
|
|
|
|||
|
|
def __init__(self, cfg, log):
|
|||
|
|
self.cfg = cfg
|
|||
|
|
self.log = log
|
|||
|
|
self.scraper = Scraper(cfg, cfg.dest, source_root=cfg.source, log=log)
|
|||
|
|
self.processed = set() # 已处理的绝对路径
|
|||
|
|
self.lock = threading.Lock()
|
|||
|
|
os.makedirs(cfg.dest, exist_ok=True)
|
|||
|
|
|
|||
|
|
def _is_stable(self, path):
|
|||
|
|
"""判断文件/目录是否写入稳定：两次采样大小一致，最多等 settle 秒。"""
|
|||
|
|
def size(p):
|
|||
|
|
if os.path.isfile(p):
|
|||
|
|
return os.path.getsize(p)
|
|||
|
|
# 目录：取所有文件大小之和
|
|||
|
|
try:
|
|||
|
|
return sum(os.path.getsize(os.path.join(r, f))
|
|||
|
|
for r, _, fs in os.walk(p) for f in fs)
|
|||
|
|
except OSError:
|
|||
|
|
return -1
|
|||
|
|
last = size(path)
|
|||
|
|
wait = self.cfg.settle
|
|||
|
|
while wait > 0:
|
|||
|
|
time.sleep(min(wait, 2))
|
|||
|
|
wait -= 2
|
|||
|
|
cur = size(path)
|
|||
|
|
if cur == last and cur >= 0:
|
|||
|
|
return True
|
|||
|
|
last = cur
|
|||
|
|
return True
|
|||
|
|
|
|||
|
|
def handle(self, path):
|
|||
|
|
path = os.path.abspath(path)
|
|||
|
|
if not os.path.exists(path):
|
|||
|
|
return
|
|||
|
|
key = path if os.path.isdir(path) else os.path.dirname(path)
|
|||
|
|
with self.lock:
|
|||
|
|
if key in self.processed:
|
|||
|
|
return
|
|||
|
|
self.processed.add(key)
|
|||
|
|
try:
|
|||
|
|
self.log(f"检测到新媒体: {path}，等待写入稳定…")
|
|||
|
|
self._is_stable(path)
|
|||
|
|
self.log(f"开始刮削: {path}")
|
|||
|
|
scan_and_scrape(path, self.cfg.dest, self.scraper)
|
|||
|
|
self.log(f"完成: {path}")
|
|||
|
|
except Exception as e:
|
|||
|
|
self.log(f"处理失败 {path}: {e}")
|
|||
|
|
|
|||
|
|
|
|||
|
|
class MediaHandler(FileSystemEventHandler):
|
|||
|
|
def __init__(self, worker):
|
|||
|
|
self.worker = worker
|
|||
|
|
self._threads = []
|
|||
|
|
self.lock = threading.Lock()
|
|||
|
|
|
|||
|
|
def _dispatch(self, path):
|
|||
|
|
if os.path.basename(path).startswith("."): # 隐藏/临时文件
|
|||
|
|
return
|
|||
|
|
ext = clean_ext(path)
|
|||
|
|
if ext in VIDEO_EXTS or os.path.isdir(path):
|
|||
|
|
t = threading.Thread(target=self.worker.handle, args=(path,),
|
|||
|
|
daemon=True)
|
|||
|
|
t.start()
|
|||
|
|
with self.lock:
|
|||
|
|
self._threads.append(t)
|
|||
|
|
self._threads = [x for x in self._threads if x.is_alive()]
|
|||
|
|
|
|||
|
|
def on_created(self, event):
|
|||
|
|
self._dispatch(event.src_path)
|
|||
|
|
|
|||
|
|
def on_moved(self, event):
|
|||
|
|
# 下载工具常见：先 .part 再改名成正式名，此时事件是 moved
|
|||
|
|
if not event.dest_path.endswith(".part"):
|
|||
|
|
self._dispatch(event.dest_path)
|
|||
|
|
|
|||
|
|
def on_closed(self, event):
|
|||
|
|
pass
|
|||
|
|
|
|||
|
|
|
|||
|
|
def main():
|
|||
|
|
ap = argparse.ArgumentParser(description="自研媒体刮削监听服务")
|
|||
|
|
ap.add_argument("--config", required=True, help="config.json 路径")
|
|||
|
|
ap.add_argument("--once", action="store_true",
|
|||
|
|
help="扫描现有内容一次后退出（初始化存量）")
|
|||
|
|
args = ap.parse_args()
|
|||
|
|
|
|||
|
|
cfg = Config2(args.config)
|
|||
|
|
if not cfg.source or not os.path.isdir(cfg.source):
|
|||
|
|
print(f"配置错误：source 不存在或未设置 -> {cfg.source}")
|
|||
|
|
sys.exit(1)
|
|||
|
|
|
|||
|
|
def log(msg):
|
|||
|
|
line = f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] {msg}"
|
|||
|
|
print(line)
|
|||
|
|
# 追加写日志文件
|
|||
|
|
logfile = os.path.join(cfg.dest, "..", "scraper.log")
|
|||
|
|
try:
|
|||
|
|
with open(logfile, "a", encoding="utf-8") as f:
|
|||
|
|
f.write(line + "\n")
|
|||
|
|
except OSError:
|
|||
|
|
pass
|
|||
|
|
|
|||
|
|
worker = ScraperWorker(cfg, log)
|
|||
|
|
|
|||
|
|
# 启动时先扫描一次现有内容（这样重启服务能补处理之前漏掉的媒体）
|
|||
|
|
for entry in os.listdir(cfg.source):
|
|||
|
|
p = os.path.join(cfg.source, entry)
|
|||
|
|
if os.path.isdir(p) or clean_ext(p) in VIDEO_EXTS:
|
|||
|
|
worker.handle(p)
|
|||
|
|
if args.once:
|
|||
|
|
log("存量扫描完成，退出")
|
|||
|
|
return
|
|||
|
|
|
|||
|
|
# 常驻监听
|
|||
|
|
handler = MediaHandler(worker)
|
|||
|
|
observer = Observer()
|
|||
|
|
observer.schedule(handler, cfg.source, recursive=False)
|
|||
|
|
observer.start()
|
|||
|
|
log(f"开始监听: {cfg.source} -> 成品目录: {cfg.dest}")
|
|||
|
|
try:
|
|||
|
|
while True:
|
|||
|
|
time.sleep(2)
|
|||
|
|
except KeyboardInterrupt:
|
|||
|
|
observer.stop()
|
|||
|
|
observer.join()
|
|||
|
|
|
|||
|
|
|
|||
|
|
if __name__ == "__main__":
|
|||
|
|
main()
|