Malin/Kemono-Downloader
1
0
Fork 0
mirror of https://github.com/Yuvi9587/Kemono-Downloader.git synced 2025-12-29 16:14:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

commit

Browse Source
This commit is contained in:
Yuvi9587
2025-07-27 06:32:15 -07:00
parent 9db89cfad0
commit e3dd0e70b6
9 changed files with 508 additions and 254 deletions
Download Patch File Download Diff File Expand all files Collapse all files

168
src/core/manager.py
View File

@@ -5,11 +5,10 @@ import json
import traceback
from concurrent.futures import ThreadPoolExecutor, as_completed, Future
from .api_client import download_from_api
from .workers import PostProcessorWorker, DownloadThread
from .workers import PostProcessorWorker
from ..config.constants import (
STYLE_DATE_BASED, STYLE_POST_TITLE_GLOBAL_NUMBERING,
MAX_THREADS, POST_WORKER_BATCH_THRESHOLD, POST_WORKER_NUM_BATCHES,
POST_WORKER_BATCH_DELAY_SECONDS
MAX_THREADS
)
from ..utils.file_utils import clean_folder_name
@@ -44,6 +43,7 @@ class DownloadManager:
self.creator_profiles_dir = None
self.current_creator_name_for_profile = None
self.current_creator_profile_path = None
self.session_file_path = None
def _log(self, message):
"""Puts a progress message into the queue for the UI."""
@@ -61,12 +61,16 @@ class DownloadManager:
if self.is_running:
self._log("❌ Cannot start a new session: A session is already in progress.")
return
self.session_file_path = config.get('session_file_path')
creator_profile_data = self._setup_creator_profile(config)
creator_profile_data['settings'] = config
creator_profile_data.setdefault('processed_post_ids', [])
self._save_creator_profile(creator_profile_data)
self._log(f"✅ Loaded/created profile for '{self.current_creator_name_for_profile}'. Settings saved.")
# Save settings to profile at the start of the session
if self.current_creator_profile_path:
creator_profile_data['settings'] = config
creator_profile_data.setdefault('processed_post_ids', [])
self._save_creator_profile(creator_profile_data)
self._log(f"✅ Loaded/created profile for '{self.current_creator_name_for_profile}'. Settings saved.")
self.is_running = True
self.cancellation_event.clear()
@@ -77,6 +81,7 @@ class DownloadManager:
self.total_downloads = 0
self.total_skips = 0
self.all_kept_original_filenames = []
is_single_post = bool(config.get('target_post_id_from_initial_url'))
use_multithreading = config.get('use_multithreading', True)
is_manga_sequential = config.get('manga_mode_active') and config.get('manga_filename_style') in [STYLE_DATE_BASED, STYLE_POST_TITLE_GLOBAL_NUMBERING]
@@ -86,72 +91,99 @@ class DownloadManager:
if should_use_multithreading_for_posts:
fetcher_thread = threading.Thread(
target=self._fetch_and_queue_posts_for_pool,
args=(config, restore_data, creator_profile_data), # Add argument here
args=(config, restore_data, creator_profile_data),
daemon=True
)
fetcher_thread.start()
else:
self._start_single_threaded_session(config)
# Single-threaded mode does not use the manager's complex logic
self._log("ℹ️ Manager is handing off to a single-threaded worker...")
# The single-threaded worker will manage its own lifecycle and signals.
# The manager's role for this session is effectively over.
self.is_running = False # Allow another session to start if needed
self.progress_queue.put({'type': 'handoff_to_single_thread', 'payload': (config,)})
def _start_single_threaded_session(self, config):
"""Handles downloads that are best processed by a single worker thread."""
self._log("ℹ️ Initializing single-threaded download process...")
self.worker_thread = threading.Thread(
target=self._run_single_worker,
args=(config,),
daemon=True
)
self.worker_thread.start()
def _run_single_worker(self, config):
"""Target function for the single-worker thread."""
try:
worker = DownloadThread(config, self.progress_queue)
worker.run() # This is the main blocking call for this thread
except Exception as e:
self._log(f"❌ CRITICAL ERROR in single-worker thread: {e}")
self._log(traceback.format_exc())
finally:
self.is_running = False
def _fetch_and_queue_posts_for_pool(self, config, restore_data):
def _fetch_and_queue_posts_for_pool(self, config, restore_data, creator_profile_data):
"""
Fetches all posts from the API and submits them as tasks to a thread pool.
This method runs in its own dedicated thread to avoid blocking.
Fetches posts from the API in batches and submits them as tasks to a thread pool.
This method runs in its own dedicated thread to avoid blocking the UI.
It provides immediate feedback as soon as the first batch of posts is found.
"""
try:
num_workers = min(config.get('num_threads', 4), MAX_THREADS)
self.thread_pool = ThreadPoolExecutor(max_workers=num_workers, thread_name_prefix='PostWorker_')
session_processed_ids = set(restore_data['processed_post_ids']) if restore_data else set()
session_processed_ids = set(restore_data.get('processed_post_ids', [])) if restore_data else set()
profile_processed_ids = set(creator_profile_data.get('processed_post_ids', []))
processed_ids = session_processed_ids.union(profile_processed_ids)
if restore_data:
if restore_data and 'all_posts_data' in restore_data:
# This logic for session restore remains as it relies on a pre-fetched list
all_posts = restore_data['all_posts_data']
processed_ids = set(restore_data['processed_post_ids'])
posts_to_process = [p for p in all_posts if p.get('id') not in processed_ids]
self.total_posts = len(all_posts)
self.processed_posts = len(processed_ids)
self._log(f"🔄 Restoring session. {len(posts_to_process)} posts remaining.")
self.progress_queue.put({'type': 'overall_progress', 'payload': (self.total_posts, self.processed_posts)})
if not posts_to_process:
self._log("✅ No new posts to process from restored session.")
return
for post_data in posts_to_process:
if self.cancellation_event.is_set(): break
worker = PostProcessorWorker(post_data, config, self.progress_queue)
future = self.thread_pool.submit(worker.process)
future.add_done_callback(self._handle_future_result)
self.active_futures.append(future)
else:
posts_to_process = self._get_all_posts(config)
self.total_posts = len(posts_to_process)
# --- START: REFACTORED STREAMING LOGIC ---
post_generator = download_from_api(
api_url_input=config['api_url'],
logger=self._log,
start_page=config.get('start_page'),
end_page=config.get('end_page'),
manga_mode=config.get('manga_mode_active', False),
cancellation_event=self.cancellation_event,
pause_event=self.pause_event,
use_cookie=config.get('use_cookie', False),
cookie_text=config.get('cookie_text', ''),
selected_cookie_file=config.get('selected_cookie_file'),
app_base_dir=config.get('app_base_dir'),
manga_filename_style_for_sort_check=config.get('manga_filename_style'),
processed_post_ids=list(processed_ids)
)
self.total_posts = 0
self.processed_posts = 0
self.progress_queue.put({'type': 'overall_progress', 'payload': (self.total_posts, self.processed_posts)})
if not posts_to_process:
self._log("✅ No new posts to process.")
return
for post_data in posts_to_process:
if self.cancellation_event.is_set():
break
worker = PostProcessorWorker(post_data, config, self.progress_queue)
future = self.thread_pool.submit(worker.process)
future.add_done_callback(self._handle_future_result)
self.active_futures.append(future)
# Process posts in batches as they are yielded by the API client
for batch in post_generator:
if self.cancellation_event.is_set():
self._log(" Post fetching cancelled.")
break
# Filter out any posts that might have been processed since the start
posts_in_batch_to_process = [p for p in batch if p.get('id') not in processed_ids]
if not posts_in_batch_to_process:
continue
# Update total count and immediately inform the UI
self.total_posts += len(posts_in_batch_to_process)
self.progress_queue.put({'type': 'overall_progress', 'payload': (self.total_posts, self.processed_posts)})
for post_data in posts_in_batch_to_process:
if self.cancellation_event.is_set(): break
worker = PostProcessorWorker(post_data, config, self.progress_queue)
future = self.thread_pool.submit(worker.process)
future.add_done_callback(self._handle_future_result)
self.active_futures.append(future)
if self.total_posts == 0 and not self.cancellation_event.is_set():
self._log("✅ No new posts found to process.")
except Exception as e:
self._log(f"❌ CRITICAL ERROR in post fetcher thread: {e}")
self._log(traceback.format_exc())
@@ -164,28 +196,6 @@ class DownloadManager:
'type': 'finished',
'payload': (self.total_downloads, self.total_skips, self.cancellation_event.is_set(), self.all_kept_original_filenames)
})
def _get_all_posts(self, config):
"""Helper to fetch all posts using the API client."""
all_posts = []
post_generator = download_from_api(
api_url_input=config['api_url'],
logger=self._log,
start_page=config.get('start_page'),
end_page=config.get('end_page'),
manga_mode=config.get('manga_mode_active', False),
cancellation_event=self.cancellation_event,
pause_event=self.pause_event,
use_cookie=config.get('use_cookie', False),
cookie_text=config.get('cookie_text', ''),
selected_cookie_file=config.get('selected_cookie_file'),
app_base_dir=config.get('app_base_dir'),
manga_filename_style_for_sort_check=config.get('manga_filename_style'),
processed_post_ids=config.get('processed_post_ids', [])
)
for batch in post_generator:
all_posts.extend(batch)
return all_posts
def _handle_future_result(self, future: Future):
"""Callback executed when a worker task completes."""
@@ -261,9 +271,15 @@ class DownloadManager:
"""Cancels the current running session."""
if not self.is_running:
return
if self.cancellation_event.is_set():
self._log("ℹ️ Cancellation already in progress.")
return
self._log("⚠️ Cancellation requested by user...")
self.cancellation_event.set()
if self.thread_pool:
self.thread_pool.shutdown(wait=False, cancel_futures=True)
self.is_running = False
self._log(" Signaling all worker threads to stop and shutting down pool...")
self.thread_pool.shutdown(wait=False)