diff --git a/db.py b/db.py index e1154a2..34aa45e 100644 --- a/db.py +++ b/db.py @@ -23,6 +23,14 @@ def init_db(db_path): processed_at TEXT NOT NULL ) """) + conn.execute(""" + CREATE TABLE IF NOT EXISTS processed_files ( + info_hash TEXT NOT NULL, + file_name TEXT NOT NULL, + processed_at TEXT NOT NULL, + PRIMARY KEY (info_hash, file_name) + ) + """) conn.commit() return conn @@ -50,16 +58,35 @@ def increment_seen(conn, file_hash): def is_torrent_processed(conn, info_hash): - """Check if a torrent has already been processed.""" + """Check if a torrent has already been fully processed.""" row = conn.execute("SELECT 1 FROM processed_torrents WHERE info_hash = ?", (info_hash,)).fetchone() return row is not None def record_torrent(conn, info_hash, torrent_name): - """Record a torrent as processed.""" + """Record a torrent as fully processed.""" now = datetime.now(timezone.utc).isoformat() conn.execute( "INSERT OR IGNORE INTO processed_torrents (info_hash, torrent_name, processed_at) VALUES (?, ?, ?)", (info_hash, torrent_name, now), ) conn.commit() + + +def is_file_processed(conn, info_hash, file_name): + """Check if an individual file within a torrent has already been processed.""" + row = conn.execute( + "SELECT 1 FROM processed_files WHERE info_hash = ? AND file_name = ?", + (info_hash, file_name), + ).fetchone() + return row is not None + + +def record_processed_file(conn, info_hash, file_name): + """Record an individual file within a torrent as processed.""" + now = datetime.now(timezone.utc).isoformat() + conn.execute( + "INSERT OR IGNORE INTO processed_files (info_hash, file_name, processed_at) VALUES (?, ?, ?)", + (info_hash, file_name, now), + ) + conn.commit() diff --git a/qbit_client.py b/qbit_client.py index d3f1177..574a41d 100644 --- a/qbit_client.py +++ b/qbit_client.py @@ -37,9 +37,11 @@ class QBitClient: resp.raise_for_status() return resp - def get_completed_torrents(self, category=None): - """Return a list of completed torrent dicts from qBittorrent.""" - params = {"filter": "completed"} + def get_torrents(self, filter=None, category=None): + """Return a list of torrent dicts, optionally filtered by state and/or category.""" + params = {} + if filter: + params["filter"] = filter if category: params["category"] = category resp = self._request("GET", "/api/v2/torrents/info", params=params) diff --git a/stash_handler.py b/stash_handler.py index cd95070..09d3e76 100644 --- a/stash_handler.py +++ b/stash_handler.py @@ -75,6 +75,64 @@ def transfer_file(src, dst, mode): # --- Daemon mode --- +def process_torrent_files(conn, qbit, t, staging, import_dir, extensions, algorithm, move_mode): + """ + Process all complete files for a torrent using the qBit files API. + + Files with progress < 1.0 are skipped (still downloading). + Files already in processed_files are skipped (already handled in a prior poll). + Returns (new_count, dup_count, incomplete_count). + """ + info_hash = t["hash"] + name = t["name"] + new_count = 0 + dup_count = 0 + incomplete_count = 0 + + try: + files = qbit.get_torrent_files(info_hash) + except Exception: + log.exception("Failed to get file list for torrent: %s", name) + return 0, 0, 0 + + for f in files: + file_name = f["name"] # relative to qBit's save_path (= our staging root) + + if f["progress"] < 1.0: + incomplete_count += 1 + continue + + if db.is_file_processed(conn, info_hash, file_name): + continue + + fpath = staging / file_name + if not fpath.exists(): + log.warning("File reported complete by qBit but not found on disk: %s", fpath) + continue + + if not is_media(fpath, extensions): + # Not a media file — mark processed so we don't check it again + db.record_processed_file(conn, info_hash, file_name) + continue + + file_hash = hash_file(fpath, algorithm) + if db.is_known(conn, file_hash): + db.increment_seen(conn, file_hash) + dup_count += 1 + log.debug("Duplicate skipped: %s", fpath.name) + else: + rel = Path(file_name) + dst = import_dir / rel + transfer_file(fpath, dst, move_mode) + db.record_file(conn, file_hash, str(rel), "torrent", info_hash) + new_count += 1 + log.info("Imported: %s", rel) + + db.record_processed_file(conn, info_hash, file_name) + + return new_count, dup_count, incomplete_count + + def daemon(cfg): qbit = QBitClient(cfg["qbit_url"], cfg["qbit_username"], cfg["qbit_password"]) conn = db.init_db(cfg["db_path"]) @@ -84,6 +142,7 @@ def daemon(cfg): algorithm = cfg["hash_algorithm"] move_mode = cfg["move_mode"] poll_interval = cfg["poll_interval"] + category = cfg["qbit_category"] running = True @@ -95,49 +154,44 @@ def daemon(cfg): signal.signal(signal.SIGINT, stop) signal.signal(signal.SIGTERM, stop) - log.info("Daemon started — polling every %ds", poll_interval) + log.info("Daemon started — polling every %ds (category: %s)", poll_interval, category) while running: try: - torrents = qbit.get_completed_torrents(cfg["qbit_category"]) + downloading = qbit.get_torrents(filter="downloading", category=category) + completed = qbit.get_torrents(filter="completed", category=category) except Exception: - log.exception("Failed to fetch torrents") + log.exception("Failed to fetch torrents from qBittorrent") _sleep(poll_interval, lambda: running) continue - for t in torrents: + # Process in-progress torrents — import any individually complete files + for t in downloading: if not running: break - info_hash = t["hash"] - name = t["name"] - - if db.is_torrent_processed(conn, info_hash): + if db.is_torrent_processed(conn, t["hash"]): continue + new, dup, incomplete = process_torrent_files( + conn, qbit, t, staging, import_dir, extensions, algorithm, move_mode + ) + if new or dup: + log.info( + "In-progress '%s': %d imported, %d duplicates, %d still downloading", + t["name"], new, dup, incomplete, + ) - log.info("Processing torrent: %s (%s)", name, info_hash) - torrent_path = staging / t.get("content_path", name).lstrip("/") - if not torrent_path.exists(): - # Fall back: content_path from qBit may be absolute or relative - torrent_path = staging / name - - new_count = 0 - dup_count = 0 - for fpath in walk_media(torrent_path if torrent_path.is_dir() else torrent_path.parent, extensions): - file_hash = hash_file(fpath, algorithm) - if db.is_known(conn, file_hash): - db.increment_seen(conn, file_hash) - dup_count += 1 - log.debug("Duplicate skipped: %s", fpath.name) - else: - rel = fpath.relative_to(staging) - dst = import_dir / rel - transfer_file(fpath, dst, move_mode) - db.record_file(conn, file_hash, str(rel), "torrent", info_hash) - new_count += 1 - log.info("Imported: %s", rel) - - db.record_torrent(conn, info_hash, name) - log.info("Torrent done: %s — %d new, %d duplicates", name, new_count, dup_count) + # Process completed torrents — import remaining files and mark torrent done + for t in completed: + if not running: + break + if db.is_torrent_processed(conn, t["hash"]): + continue + log.info("Completed torrent: %s (%s)", t["name"], t["hash"]) + new, dup, _ = process_torrent_files( + conn, qbit, t, staging, import_dir, extensions, algorithm, move_mode + ) + db.record_torrent(conn, t["hash"], t["name"]) + log.info("Torrent done: '%s' — %d imported, %d duplicates", t["name"], new, dup) _sleep(poll_interval, lambda: running) diff --git a/summary.md b/summary.md index e8b3401..a90c53b 100644 --- a/summary.md +++ b/summary.md @@ -4,7 +4,7 @@ ## Purpose -StashHandler sits between qBittorrent and Stash (both running in Docker). It polls qBit's API for completed torrents filtered by category, hashes media files, deduplicates them via a SQLite database, and copies new files into Stash's import directory (preserving originals for continued seeding). +StashHandler sits between qBittorrent and Stash (both running in Docker). It polls qBit's API for torrents filtered by category, hashes media files, deduplicates them via a SQLite database, and copies new files into Stash's import directory (preserving originals for continued seeding). Individual files within in-progress torrents are imported as soon as they complete — the daemon does not wait for the full torrent to finish. **Problems solved:** 1. Deleted files in Stash don't get reimported (hash DB remembers them) @@ -29,7 +29,7 @@ Designed for Docker Swarm — no host-local config files, everything via env var | File | Purpose | |---|---| | `stash_handler.py` | Entry point — `daemon` and `scan` CLI commands, config loading from env vars | -| `db.py` | SQLite operations — `init_db`, `is_known`, `record_file`, `is_torrent_processed`, `record_torrent`, `increment_seen` | +| `db.py` | SQLite operations — `init_db`, `is_known`, `record_file`, `is_torrent_processed`, `record_torrent`, `increment_seen`, `is_file_processed`, `record_processed_file` | | `hasher.py` | Streaming file hashing (sha256 default, blake3 optional) in 64KB chunks | | `qbit_client.py` | qBittorrent API client — login, session cookies, auto re-auth on 403 | | `Dockerfile` | python:3.12-slim, entrypoint is `stash_handler.py`, default command `daemon` | @@ -45,9 +45,11 @@ docker run stash-handler # or: python stash_handler.py daemon ``` - Polls qBit API at `POLL_INTERVAL` seconds -- Fetches completed torrents via `/api/v2/torrents/info?filter=completed&category=` -- Skips already-processed torrents (by info_hash in `processed_torrents` table) -- For each new torrent: walks media files, hashes, deduplicates, transfers new files to import dir +- Each poll fetches both `downloading` and `completed` torrents filtered by `QBIT_CATEGORY` +- Uses `/api/v2/torrents/files` to get per-file progress (0.0–1.0) for each torrent +- Files with `progress < 1.0` are skipped — imported as soon as they reach 1.0 +- Per-file state tracked in `processed_files` table to avoid re-hashing on subsequent polls +- Completed torrents are marked in `processed_torrents` and never checked again - Clean shutdown on SIGINT/SIGTERM (1-second sleep granularity) ### Scan mode @@ -95,6 +97,13 @@ CREATE TABLE processed_torrents ( torrent_name TEXT, processed_at TEXT NOT NULL ); + +CREATE TABLE processed_files ( + info_hash TEXT NOT NULL, + file_name TEXT NOT NULL, -- relative to qBit save_path (= staging root) + processed_at TEXT NOT NULL, + PRIMARY KEY (info_hash, file_name) +); ``` ## Volume Layout (Docker)