288 lines
14 KiB
Python
288 lines
14 KiB
Python
import re
|
|
import uuid
|
|
import requests
|
|
import os
|
|
import time
|
|
import datetime
|
|
|
|
from bs4 import BeautifulSoup
|
|
from urllib.parse import urljoin, quote
|
|
from concurrent.futures import ThreadPoolExecutor, as_completed
|
|
from pathlib import Path
|
|
from requests.exceptions import ChunkedEncodingError
|
|
from tkinter import messagebox, simpledialog
|
|
|
|
class EromeDownloader:
|
|
def __init__(self, root, log_callback=None, enable_widgets_callback=None, update_progress_callback=None, update_global_progress_callback=None, download_images=True, download_videos=True, headers=None, language="en", is_profile_download=False, direct_download=False, tr=None, max_workers=5):
|
|
self.root = root
|
|
self.session = requests.Session()
|
|
self.headers = {k: str(v).encode('ascii', 'ignore').decode('ascii') for k, v in (headers or {
|
|
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/127.0.0.0 Safari/537.36',
|
|
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8'
|
|
}).items()}
|
|
self.log_messages = [] # Store log messages
|
|
self.log_callback = log_callback
|
|
self.enable_widgets_callback = enable_widgets_callback
|
|
self.update_progress_callback = update_progress_callback
|
|
self.update_global_progress_callback = update_global_progress_callback
|
|
self.download_images = download_images
|
|
self.download_videos = download_videos
|
|
self.cancel_requested = False
|
|
self.language = language
|
|
self.executor = ThreadPoolExecutor(max_workers=max_workers) # Thread pool for concurrent downloads
|
|
self.total_files = 0
|
|
self.completed_files = 0
|
|
self.is_profile_download = is_profile_download
|
|
self.direct_download = direct_download # Option for direct downloads without folder creation
|
|
self.tr = tr if tr else lambda x, **kwargs: x.format(**kwargs) # Translation function
|
|
|
|
def request_cancel(self):
|
|
self.cancel_requested = True
|
|
self.log(self.tr("Download cancelled"))
|
|
if self.is_profile_download:
|
|
self.enable_widgets_callback()
|
|
|
|
def log(self, message):
|
|
if self.log_callback is not None:
|
|
self.log_callback(message)
|
|
self.log_messages.append(message)
|
|
|
|
def shutdown_executor(self):
|
|
self.executor.shutdown(wait=False)
|
|
self.log(self.tr("Executor shut down."))
|
|
if self.is_profile_download:
|
|
self.enable_widgets_callback()
|
|
|
|
@staticmethod
|
|
def clean_filename(filename):
|
|
return re.sub(r'[<>:"/\\|?*]', '_', filename.split('?')[0])
|
|
|
|
def create_folder(self, folder_name):
|
|
try:
|
|
os.makedirs(folder_name, exist_ok=True)
|
|
except OSError as e:
|
|
self.log(self.tr("Error creating folder: {error}", error=e))
|
|
response = messagebox.askyesno(self.tr("Error"), self.tr("Couldn't create folder: {folder_name}\nWould you like to choose a new name?", folder_name=folder_name), parent=self.root)
|
|
if response:
|
|
new_folder_name = simpledialog.askstring(self.tr("New folder name"), self.tr("Enter new folder name:"), parent=self.root)
|
|
if new_folder_name:
|
|
folder_name = os.path.join(os.path.dirname(folder_name), self.clean_filename(new_folder_name))
|
|
try:
|
|
os.makedirs(folder_name, exist_ok=True)
|
|
except OSError as e:
|
|
messagebox.showerror(self.tr("Error"), self.tr("Could not create folder: {folder_name}\nError: {error}", folder_name=folder_name, error=e), parent=self.root)
|
|
return folder_name
|
|
|
|
def download_file(self, url, file_path, resource_type, file_id=None, max_retries=999999):
|
|
if self.cancel_requested:
|
|
return
|
|
|
|
# Evita sobreescrituras y descargas duplicadas
|
|
if os.path.exists(file_path):
|
|
self.log(self.tr("File already exists, skipping: {file_path}",
|
|
file_path=file_path))
|
|
return
|
|
|
|
os.makedirs(os.path.dirname(file_path), exist_ok=True)
|
|
self.log(self.tr("Start downloading {resource_type}: {file_path}",
|
|
resource_type=resource_type, file_path=file_path))
|
|
|
|
retries = 0
|
|
while retries <= max_retries:
|
|
try:
|
|
with requests.get(url, headers=self.headers,
|
|
stream=True, timeout=15) as response:
|
|
if response.status_code != 200:
|
|
self.log(self.tr("Error downloading {resource_type}, "
|
|
"status code: {status_code}",
|
|
resource_type=resource_type,
|
|
status_code=response.status_code))
|
|
break
|
|
|
|
total_size = int(response.headers.get("content-length", 0))
|
|
downloaded_size = 0
|
|
start_time = last_update = time.time()
|
|
|
|
with open(file_path, "wb") as f:
|
|
for chunk in response.iter_content(chunk_size=65536):
|
|
if self.cancel_requested:
|
|
return
|
|
f.write(chunk)
|
|
downloaded_size += len(chunk)
|
|
|
|
# Envía actualización cada 0.5 s máx.
|
|
now = time.time()
|
|
if now - last_update >= 0.5:
|
|
elapsed = now - start_time
|
|
speed = downloaded_size / elapsed if elapsed else 0
|
|
eta = ((total_size - downloaded_size) / speed
|
|
if speed else None)
|
|
self.update_progress_callback(
|
|
downloaded_size, total_size,
|
|
file_id=file_id,
|
|
file_path=file_path,
|
|
speed=speed,
|
|
eta=eta
|
|
)
|
|
last_update = now
|
|
|
|
# ─────── Fin de la descarga ───────
|
|
elapsed = time.time() - start_time
|
|
final_speed = total_size / elapsed if elapsed else 0
|
|
|
|
# 1· Fuerza la barra al 100 % (sin status)
|
|
self.update_progress_callback(
|
|
total_size, total_size,
|
|
file_id=file_id,
|
|
file_path=file_path,
|
|
speed=final_speed,
|
|
eta=0
|
|
)
|
|
|
|
# 2· Notifica “Completed” (no altera la barra)
|
|
self.update_progress_callback(
|
|
total_size, total_size,
|
|
file_id=file_id,
|
|
file_path=file_path,
|
|
status="Completed"
|
|
)
|
|
|
|
# Contabiliza y avanza la barra global
|
|
self.completed_files += 1
|
|
if self.update_global_progress_callback:
|
|
self.update_global_progress_callback(
|
|
self.completed_files, self.total_files
|
|
)
|
|
|
|
self.log(self.tr("Download successful: {resource_type}, "
|
|
"{file_path}",
|
|
resource_type=resource_type,
|
|
file_path=file_path))
|
|
break # Éxito ⇒ sal del bucle
|
|
|
|
except (requests.exceptions.ChunkedEncodingError,
|
|
requests.exceptions.ConnectionError,
|
|
requests.exceptions.Timeout) as e:
|
|
retries += 1
|
|
self.log(self.tr("Error downloading {resource_type}, "
|
|
"attempt {retries}/{max_retries}: {error}",
|
|
resource_type=resource_type,
|
|
retries=retries,
|
|
max_retries=max_retries,
|
|
error=e))
|
|
if retries == max_retries:
|
|
self.log(self.tr("Max retries reached. Failed to "
|
|
"download {resource_type}: {file_path}",
|
|
resource_type=resource_type,
|
|
file_path=file_path))
|
|
|
|
|
|
def process_album_page(self, page_url, base_folder, download_images=True, download_videos=True):
|
|
try:
|
|
if self.cancel_requested:
|
|
return
|
|
self.log(self.tr("Processing album URL: {page_url}", page_url=page_url))
|
|
response = requests.get(page_url, headers=self.headers)
|
|
if response.status_code == 200:
|
|
soup = BeautifulSoup(response.text, 'html.parser')
|
|
if not self.direct_download:
|
|
folder_name = self.clean_filename(soup.find('h1').text if soup.find('h1') else self.tr("Unknown Album"))
|
|
folder_path = self.create_folder(os.path.join(base_folder, folder_name))
|
|
else:
|
|
folder_path = base_folder # Use the base folder directly
|
|
|
|
media_urls = []
|
|
seen_urls = set()
|
|
|
|
# --- vídeos ---
|
|
if download_videos:
|
|
for video in soup.find_all('video'):
|
|
source = video.find('source')
|
|
if source:
|
|
abs_video_src = urljoin(page_url, source['src'])
|
|
if abs_video_src in seen_urls: # ya lo tenemos
|
|
continue
|
|
seen_urls.add(abs_video_src)
|
|
video_name = os.path.join(
|
|
folder_path,
|
|
self.clean_filename(os.path.basename(abs_video_src))
|
|
)
|
|
media_urls.append(
|
|
(abs_video_src, video_name, 'Video')
|
|
)
|
|
|
|
# --- imágenes ---
|
|
if download_images:
|
|
for div in soup.select('div.img'):
|
|
img = div.find('img', attrs={'data-src': True})
|
|
if img:
|
|
abs_img_src = urljoin(page_url, img['data-src'])
|
|
if abs_img_src in seen_urls:
|
|
continue
|
|
seen_urls.add(abs_img_src)
|
|
img_name = os.path.join(
|
|
folder_path,
|
|
self.clean_filename(os.path.basename(abs_img_src))
|
|
)
|
|
media_urls.append(
|
|
(abs_img_src, img_name, 'Image')
|
|
)
|
|
|
|
self.total_files += len(media_urls)
|
|
futures = [self.executor.submit(self.download_file, url, file_path, resource_type, str(uuid.uuid4())) for url, file_path, resource_type in media_urls]
|
|
for future in as_completed(futures):
|
|
if self.cancel_requested:
|
|
self.log(self.tr("Cancelling remaining downloads."))
|
|
break
|
|
future.result()
|
|
|
|
self.log(self.tr("Album download complete: {folder_name}", folder_name=folder_name) if not self.direct_download else self.tr("Album download complete"))
|
|
if not self.is_profile_download:
|
|
self.enable_widgets_callback()
|
|
else:
|
|
self.log(self.tr("Error accessing page: {page_url}, status code: {status_code}", page_url=page_url, status_code=response.status_code))
|
|
if not self.is_profile_download:
|
|
self.enable_widgets_callback()
|
|
finally:
|
|
if not self.is_profile_download:
|
|
self.enable_widgets_callback()
|
|
self.export_logs()
|
|
|
|
def process_profile_page(self, url, download_folder, download_images, download_videos):
|
|
try:
|
|
if self.cancel_requested:
|
|
return
|
|
self.log(self.tr("Processing profile URL: {url}", url=url))
|
|
response = requests.get(url, headers=self.headers)
|
|
if response.status_code == 200:
|
|
soup = BeautifulSoup(response.text, 'html.parser')
|
|
username = soup.find('h1', class_='username').text.strip() if soup.find('h1', class_='username') else self.tr("Unknown Profile")
|
|
base_folder = self.create_folder(os.path.join(download_folder, self.clean_filename(username)))
|
|
|
|
album_links = soup.find_all('a', class_='album-link')
|
|
for album_link in album_links:
|
|
album_href = album_link.get('href')
|
|
album_full_url = urljoin(url, album_href)
|
|
self.process_album_page(album_full_url, base_folder, download_images, download_videos)
|
|
|
|
self.log(self.tr("Profile download complete: {username}", username=username))
|
|
self.enable_widgets_callback()
|
|
else:
|
|
self.log(self.tr("Error accessing page: {url}, status code: {status_code}", url=url, status_code=response.status_code))
|
|
self.enable_widgets_callback()
|
|
finally:
|
|
if not self.is_profile_download:
|
|
self.enable_widgets_callback()
|
|
self.export_logs()
|
|
|
|
def export_logs(self):
|
|
log_folder = "resources/config/logs/"
|
|
Path(log_folder).mkdir(parents=True, exist_ok=True)
|
|
log_file_path = Path(log_folder) / f"log_{datetime.datetime.now().strftime('%Y%m%d_%H%M%S')}.txt"
|
|
try:
|
|
with open(log_file_path, 'w') as file:
|
|
file.write("\n".join(self.log_messages))
|
|
self.log(self.tr("Logs exported successfully to {path}", path=log_file_path))
|
|
except Exception as e:
|
|
self.log(self.tr(f"Failed to export logs: {e}"))
|