azalea/MTfin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[+] Local mode

Browse Source
This commit is contained in:
Azalea Gui
2026-03-10 16:11:08 -04:00
parent 80ddcf3be2
commit 7be86b2d8f
1 changed files with 169 additions and 94 deletions
Download Patch File Download Diff File Expand all files Collapse all files
workflow.py
+114 -39
View File
@@ -83,55 +83,40 @@ def wait_for_download(qb, t_hash: str):
print("Download complete!")
def process_imdb_workflow(imdb_id: str, dl_dir: str = DEFAULT_DL_DIR, jellyfin_base_dir: str = DEFAULT_JELLYFIN_DIR):
def check_local_filesystem(dl_dir: str, imdb_id: str):
"""
Workflow to automatically find, download, and map torrents for an IMDb ID into a Jellyfin library.
Checks the local download directory for any existing files or folders
that match the given IMDb ID. Returns the path if found.
"""
print(f"=== [0] Fetching IMDB info for {imdb_id} ===")
imdb_info = mteam_imdb_info(imdb_id)
if 'data' not in imdb_info:
raise ValueError(f"Failed to get IMDB info from M-Team: {imdb_info}")
title = imdb_info['data'].get('title', 'Unknown_Title')
year = imdb_info['data'].get('year', '')
title_dir = f"{title} ({year})"
print(f"Found Title: {title_dir}")
print(f"\n=== [0.5] Checking if torrent already exists in qBittorrent ===")
qb = get_qb_client()
new_name = f"{year} {title} [{imdb_id}]".strip()
dl_path = Path(dl_dir)
if dl_path.exists():
for item in dl_path.iterdir():
if f"[{imdb_id}]" in item.name:
return item
return None
def check_qbittorrent(qb, imdb_id: str):
"""
Checks qBittorrent for any existing torrent that has the IMDb ID in its name.
Returns the torrent hash if found, or None.
"""
existing_torrents = qb.torrents_info()
existing_t_hash = None
for t in existing_torrents:
if f"[{imdb_id}]" in t.name:
existing_t_hash = t.hash
break
return t.hash
return None
hashes_to_process = []
if existing_t_hash:
print(f"Found existing torrent with hash {existing_t_hash}, skipping search and download.")
rename_torrent_and_folder(qb, existing_t_hash, new_name)
print(f"\n=== [0.6] Waiting for existing download to finish ===")
wait_for_download(qb, existing_t_hash)
hashes_to_process.append((existing_t_hash, "existing"))
else:
def search_and_download_mteam(qb, imdb_id: str, new_name: str, dl_dir: str) -> list:
"""
Searches M-Team for the IMDb ID, uses an LLM to select the best torrents,
downloads them, adds them to qBittorrent, and waits for them to complete.
Returns a list of tuples containing (torrent_hash, torrent_id).
"""
print(f"\n=== [1] Searching Torrents for {imdb_id} ===")
imdb_url = f"https://www.imdb.com/title/{imdb_id}/"
search_res = search_mteam_torrents(imdb_url)
# Extract the torrent list
if "data" in search_res and isinstance(search_res["data"], dict) and "data" in search_res["data"]:
torrents = search_res["data"]["data"]
elif "data" in search_res and isinstance(search_res["data"], list):
torrents = search_res["data"]
elif isinstance(search_res, list):
torrents = search_res
else:
torrents = []
torrents = search_mteam_torrents(imdb_url)["data"]["data"]
if not torrents:
raise ValueError(f"No torrents found on M-Team for IMDb ID: {imdb_id}")
@@ -151,6 +136,7 @@ def process_imdb_workflow(imdb_id: str, dl_dir: str = DEFAULT_DL_DIR, jellyfin_b
if not selected_ids:
raise ValueError(f"LLM did not select any torrents for IMDb ID: {imdb_id}")
hashes_to_process = []
for tid in selected_ids:
print(f"\n=== [3] Downloading .torrent for ID: {tid} ===")
torrent_bytes = generate_mteam_download_token(tid)
@@ -175,7 +161,13 @@ def process_imdb_workflow(imdb_id: str, dl_dir: str = DEFAULT_DL_DIR, jellyfin_b
hashes_to_process.append((t_hash, tid))
for t_hash, tid in hashes_to_process:
return hashes_to_process
def process_qb_torrent(qb, t_hash: str, tid: str, new_name: str, dl_dir: str, title_dir: str, imdb_id: str, jellyfin_base_dir: str):
"""
Uses an LLM to generate a rename mapping for the files inside a completed qBittorrent download.
Applies tags based on TV/Movie type and creates symbolic links to the Jellyfin library.
"""
print(f"\n=== [6] Generating rename mapping using LLM ===")
file_tree = get_torrent_file_tree(qb, t_hash)
@@ -208,6 +200,89 @@ def process_imdb_workflow(imdb_id: str, dl_dir: str = DEFAULT_DL_DIR, jellyfin_b
apply_rename_mapping(mapping, base_src_dir=src_dir_for_mapping, base_dst_dir=jellyfin_base)
print(f"Finished processing torrent: {tid}")
def process_local_file(fs_path: Path, dl_dir: str, title_dir: str, imdb_id: str, jellyfin_base_dir: str):
"""
Generates an LLM rename mapping for purely local files/folders (skipping qBittorrent)
and creates symbolic links to the Jellyfin library.
"""
print(f"\n=== [6] Generating rename mapping using LLM for local path ===")
# Mock file tree logic for local files
file_tree = []
if fs_path.is_file():
file_tree.append({"name": fs_path.name})
else:
for p in fs_path.rglob('*'):
if p.is_file():
# relative to fs_path's parent so it starts with fs_path.name
rel_path = p.relative_to(fs_path.parent)
file_tree.append({"name": str(rel_path.as_posix())})
src_dir_for_mapping = prepare_file_tree_paths(file_tree, fs_path.name, dl_dir)
file_tree_str = format_file_tree(file_tree)
prompt_text = f"Base directory: `{title_dir}`\n\n{file_tree_str}"
print(f"Sending paths to LLM...")
mapping = generate_rename_mapping(prompt_text)
print("Generated Mapping:")
is_tv = False
for src, dst in mapping.items():
print(f" {src} -->> {dst}")
if "Season " in dst or "Series " in dst:
is_tv = True
jellyfin_dir = f"{jellyfin_base_dir}/TV" if is_tv else f"{jellyfin_base_dir}/Movie"
jellyfin_base = Path(jellyfin_dir) / f"{title_dir} [{imdb_id}]"
print(f"\n=== [7] Creating symbolic links ===")
apply_rename_mapping(mapping, base_src_dir=src_dir_for_mapping, base_dst_dir=jellyfin_base)
print(f"Finished processing local file: {fs_path.name}")
def process_imdb_workflow(imdb_id: str, dl_dir: str = DEFAULT_DL_DIR, jellyfin_base_dir: str = DEFAULT_JELLYFIN_DIR):
"""
Workflow to automatically find, download, and map torrents for an IMDb ID into a Jellyfin library.
"""
print(f"=== [0] Fetching IMDB info for {imdb_id} ===")
imdb_info = mteam_imdb_info(imdb_id)
if 'data' not in imdb_info:
raise ValueError(f"Failed to get IMDB info from M-Team: {imdb_info}")
title = imdb_info['data'].get('title', 'Unknown_Title')
year = imdb_info['data'].get('year', '')
title_dir = f"{title} ({year})"
print(f"Found Title: {title_dir}")
print(f"\n=== [0.2] Checking if already exists in file system ===")
fs_match_dir = check_local_filesystem(dl_dir, imdb_id)
new_name = f"{year} {title} [{imdb_id}]".strip()
if fs_match_dir:
print(f"Found existing file/directory in file system: {fs_match_dir.name}, skipping qBit check, search, and download.")
process_local_file(fs_match_dir, dl_dir, title_dir, imdb_id, jellyfin_base_dir)
return
print(f"\n=== [0.5] Checking if torrent already exists in qBittorrent ===")
qb = get_qb_client()
existing_t_hash = check_qbittorrent(qb, imdb_id)
hashes_to_process = []
if existing_t_hash:
print(f"Found existing torrent with hash {existing_t_hash}, skipping search and download.")
rename_torrent_and_folder(qb, existing_t_hash, new_name)
print(f"\n=== [0.6] Waiting for existing download to finish ===")
wait_for_download(qb, existing_t_hash)
hashes_to_process.append((existing_t_hash, "existing"))
else:
hashes_to_process = search_and_download_mteam(qb, imdb_id, new_name, dl_dir)
# Process qB torrents
for t_hash, tid in hashes_to_process:
process_qb_torrent(qb, t_hash, tid, new_name, dl_dir, title_dir, imdb_id, jellyfin_base_dir)
if __name__ == "__main__":
import argparse