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()
|