Implement per-file import for in-progress torrents
Files within a downloading torrent are now imported as soon as their individual progress reaches 100%, without waiting for the full torrent to complete. A new processed_files table (info_hash, file_name) tracks which files have been handled to avoid redundant re-hashing each poll. - db.py: add processed_files table, is_file_processed, record_processed_file - qbit_client.py: replace get_completed_torrents with get_torrents(filter, category) - stash_handler.py: extract process_torrent_files helper, daemon now polls both downloading and completed torrents each cycle Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -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()
|
||||
|
||||
+5
-3
@@ -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)
|
||||
|
||||
+86
-32
@@ -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)
|
||||
|
||||
|
||||
+14
-5
@@ -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=<QBIT_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)
|
||||
|
||||
Reference in New Issue
Block a user