from utils.data.models import Download, DownloadList from PySide6.QtCore import QObject, Signal from utils.data.state import state from dataclasses import asdict from datetime import datetime import json import time import os import re class LogSignal(QObject): new_log = Signal(str) log_emitter = LogSignal() def _downloads_file_path() -> str: return os.path.join(state.settings_path, "downloads.json") def _load_downloads() -> list[Download]: downloads_file = _downloads_file_path() if os.path.exists(downloads_file) and os.path.getsize(downloads_file) > 0: try: with open(downloads_file, "r") as file: existing_data = json.load(file) return [Download(**d) for d in existing_data.get("data", [])] except (json.JSONDecodeError, TypeError) as e: consoleLog(f"Error loading downloads.json: {e}") return [] def _save_downloads(downloads: list[Download]) -> DownloadList: download_list = DownloadList(data=downloads, count=len(downloads)) with open(_downloads_file_path(), "w") as file: json.dump(asdict(download_list), file, indent=4) return download_list def add_download_log(title, url, magnet_uri, completed) -> DownloadList: # wait for metadata outside the lock to avoid blocking other threads magnetdl = state.active_downloads.get(magnet_uri) if magnetdl: status = magnetdl.status() timeout = 60 start = time.time() while not status.has_metadata and (time.time() - start) < timeout: time.sleep(0.5) status = magnetdl.status() torrent_name = status.name save_path = status.save_path consoleLog(f"Torrent name (has_metadata={status.has_metadata}): {torrent_name}") path = os.path.join(save_path, torrent_name) else: path = os.path.join(state.download_path, title) with state.downloads_lock: return _add_download_log_inner(title, url, magnet_uri, completed, path) def _add_download_log_inner(title, url, magnet_uri, completed, path) -> DownloadList: downloads = _load_downloads() if any((magnet_uri and d.magnet_uri == magnet_uri) or (url and d.url == url) for d in downloads): if magnet_uri and magnet_uri in state.active_downloads: consoleLog("Skipping Logging, download already running...") return DownloadList(data=downloads, count=len(downloads)) consoleLog("File already in Log, updating Download State...") hash = extract_hash_from_magnet(magnet_uri) if hash: update_download_completed_by_hash(hash, False) return DownloadList(data=downloads, count=len(downloads)) downloads.append(Download( title=title, url=url, magnet_uri=magnet_uri, path=path, completed=completed )) consoleLog(f"Added {title} to Log File") return _save_downloads(downloads) def remove_download_log(magnet_uri) -> DownloadList: with state.downloads_lock: return _remove_download_log_inner(magnet_uri) def _remove_download_log_inner(magnet_uri) -> DownloadList: downloads = _load_downloads() magnet_link = (magnet_uri or "").strip() if not magnet_link: return DownloadList(data=downloads, count=len(downloads)) title = next((getattr(d, 'title', 'Unknown') for d in downloads if (getattr(d, 'magnet_uri', None) or '').strip() == magnet_link or (getattr(d, 'url', None) or '').strip() == magnet_link), 'Unknown') downloads = [d for d in downloads if (getattr(d, 'magnet_uri', None) or "").strip() != magnet_link and (getattr(d, 'url', None) or "").strip() != magnet_link] consoleLog(f"Removed {title} from Log File") return _save_downloads(downloads) def update_download_completed(magnet_uri, completed) -> DownloadList: with state.downloads_lock: return _update_download_completed_inner(magnet_uri, completed) def _update_download_completed_inner(magnet_uri, completed) -> DownloadList: downloads = _load_downloads() identifier = (magnet_uri or "").strip() if state.debug: consoleLog(f"Updating download completed for identifier: {identifier!r}") consoleLog(f"Total downloads in file: {len(downloads)}") for i, d in enumerate(downloads): try: mag = (getattr(d, 'magnet_uri', None) or '')[:50] url = (getattr(d, 'url', None) or '')[:50] consoleLog(f" [{i}] magnet: {mag}... url: {url}...") except Exception as e: consoleLog(f" [{i}] Error reading download: {e}") found = False for download in downloads: try: stored_magnet = (getattr(download, 'magnet_uri', None) or "").strip() stored_url = (getattr(download, 'url', None) or "").strip() if identifier and (stored_magnet == identifier or stored_url == identifier): download.completed = completed found = True except Exception as e: consoleLog(f"Error comparing download: {e}") if not found: consoleLog("No matching download found to update") return DownloadList(data=downloads, count=len(downloads)) consoleLog("Updated download log") return _save_downloads(downloads) def get_download_logs() -> DownloadList: with state.downloads_lock: return _get_download_logs_inner() def _get_download_logs_inner() -> DownloadList: downloads = _load_downloads() return DownloadList(data=downloads, count=len(downloads)) def extract_hash_from_magnet(magnet_uri): if not magnet_uri: return None try: import libtorrent as lt params = lt.parse_magnet_uri(magnet_uri) if hasattr(params, 'info_hashes'): # lt 2.0+ return str(params.info_hashes.v1).upper() return str(params.info_hash).upper() except Exception: match = re.search(r'urn:btih:([a-zA-Z0-9]+)', magnet_uri) if match: return match.group(1).upper() return None def update_download_completed_by_hash(info_hash, completed) -> DownloadList: with state.downloads_lock: return _update_download_completed_by_hash_inner(info_hash, completed) def _update_download_completed_by_hash_inner(info_hash, completed) -> DownloadList: downloads = _load_downloads() info_hash_upper = (info_hash or "").upper().strip() consoleLog(f"Updating download by hash: {info_hash_upper}") consoleLog(f"Total downloads in file: {len(downloads)}") found = False for download in downloads: try: magnet_uri = getattr(download, 'magnet_uri', None) if magnet_uri: stored_hash = extract_hash_from_magnet(magnet_uri) if stored_hash and stored_hash == info_hash_upper: download.completed = completed found = True except Exception as e: consoleLog(f"Error comparing download: {e}") if not found: consoleLog("No matching download found to update") return DownloadList(data=downloads, count=len(downloads)) consoleLog("Updated download log") return _save_downloads(downloads) def set_main_window(window): state.main_window = window def flush_log_buffer(): with state._log_lock: for text in state.log_buffer: log_emitter.new_log.emit(text) state.log_buffer.clear() def consoleLog(text, printAnyways = False): now = datetime.now() current_time = now.strftime("%H:%M:%S") formatted_text = f"[{current_time}] {text}" with state._log_lock: state.log_buffer.append(formatted_text) log_emitter.new_log.emit(formatted_text) if state.debug or printAnyways: print(formatted_text)