wikimedia commons download files in folder.py

main.py

import sys
import os
import hashlib
import json
import requests
from PyQt6.QtWidgets import (QApplication, QWidget, QVBoxLayout, QLabel, 
                             QLineEdit, QPushButton, QTextEdit, QProgressBar, 
                             QCheckBox, QMessageBox)
from PyQt6.QtCore import QThread, pyqtSignal, Qt

'''
pip install PyQt6 requests
'''
class DownloadWorker(QThread):
    log_signal = pyqtSignal(str)
    progress_signal = pyqtSignal(int)
    finished_signal = pyqtSignal()

    def __init__(self, category, username, user_agent):
        super().__init__()
        self.category = category.strip()
        self.username = username.strip()
        self.user_agent = user_agent
        self.is_running = True

    def run(self):
        try:
            self.log_signal.emit(f"🚀 Starting processing for category: {self.category}")
            
            # Create Session
            session = requests.Session()
            session.headers.update({"User-Agent": self.user_agent})

            # 1. Fetch file list from Category
            self.log_signal.emit("🔍 Fetching file list...")
            files = self.get_category_members(session)
            print(files)
            
            if not files:
                self.log_signal.emit("❌ No files found in this category.")
                self.finished_signal.emit()
                return

            self.log_signal.emit(f"📄 Found {len(files)} files. Starting download...")

            # Create directory
            # Sanitize category name for folder
            safe_folder = "".join(c for c in self.category if c.isalnum() or c in (' ', '_', '-')).strip()
            if not os.path.exists(safe_folder):
                os.makedirs(safe_folder)

            # 2. Process files
            total = len(files)
            for i, file_data in enumerate(files):
                if not self.is_running:
                    break

                title = file_data['title']
                
                # Check user filter if specified
                uploader = file_data.get('imageinfo', [{}])[0].get('user', '')
                if self.username and uploader != self.username:
                    self.log_signal.emit(f"⏭️ Skipping {title} (Uploader: {uploader})")
                    self.progress_signal.emit(int((i + 1) / total * 100))
                    continue

                # Get URL and Extension
                image_info = file_data.get('imageinfo', [{}])[0]
                url = image_info.get('url')
                if not url:
                    self.log_signal.emit(f"⚠️ No URL for {title}")
                    continue
                
                ext = os.path.splitext(title)[1]
                
                # Generate Short Unique Filename (SHA1 Hash)
                hash_name = hashlib.sha1(title.encode('utf-8')).hexdigest()[:12]
                file_name = f"{title[5:50]}{hash_name}{ext}"
                file_path = os.path.join(safe_folder, file_name)
                json_path = os.path.join(safe_folder, f"{title[5:50]}{hash_name}.json")

                # Download File
                if not os.path.exists(file_path):
                    self.log_signal.emit(f"⬇️ Downloading: {title} -> {file_name}")
                    try:
                        self.download_file(session, url, file_path)
                    except Exception as e:
                        self.log_signal.emit(f"❌ Error downloading {title}: {str(e)}")
                        continue
                else:
                    self.log_signal.emit(f"✅ File exists: {file_name}")

                # Fetch and Save SDC (Structured Data)
                self.log_signal.emit(f"📋 Saving SDC for {title}...")
                sdc_data = self.get_sdc_data(session, title)
                if sdc_data:
                    with open(json_path, 'w', encoding='utf-8') as f:
                        json.dump(sdc_data, f, ensure_ascii=False, indent=2)

                self.progress_signal.emit(int((i + 1) / total * 100))

            self.log_signal.emit("🎉 Job Completed!")

        except Exception as e:
            self.log_signal.emit(f"❌ Critical Error: {str(e)}")
        finally:
            self.finished_signal.emit()

    def get_category_members(self, session):
        """Fetch all file members from the category using a generator."""
        files = []
        url = "https://commons.wikimedia.org/w/api.php"
        params = {
            "action": "query",
            "generator": "categorymembers",
            "gcmtitle": self.category if self.category.startswith("Category:") else f"Category:{self.category}",
            "gcmtype": "file",
            "gcmlimit": "50",  # Max 500 for normal users
            "prop": "imageinfo",
            "iiprop": "url|user", # We need URL and User for filtering
            "format": "json"
        }

        while True:
            response = session.get(url, params=params)
            data = response.json()
            
            if 'query' in data and 'pages' in data['query']:
                files.extend(data['query']['pages'].values())
            
            if 'continue' in data:
                params.update(data['continue'])
            else:
                break
        return files

    def get_sdc_data(self, session, title):
        """Fetch Structured Data on Commons (Wikibase Entities) for a file."""
        url = "https://commons.wikimedia.org/w/api.php"
        params = {
            "action": "wbgetentities",
            "sites": "commonswiki",
            "titles": title,
            "format": "json"
        }
        try:
            resp = session.get(url, params=params)
            data = resp.json()
            # The API returns entities keyed by ID (e.g., M12345). We return the whole dict.
            return data.get('entities', {})
        except Exception:
            return None

    def download_file(self, session, url, path):
        with session.get(url, stream=True) as r:
            r.raise_for_status()
            with open(path, 'wb') as f:
                for chunk in r.iter_content(chunk_size=8192):
                    f.write(chunk)

    def stop(self):
        self.is_running = False

class App(QWidget):
    def __init__(self):
        super().__init__()
        self.setWindowTitle("Wikimedia Commons Downloader")
        self.resize(600, 500)
        self.init_ui()

    def init_ui(self):
        layout = QVBoxLayout()

        # Category Input
        layout.addWidget(QLabel("Category Name (e.g., 'Featured_pictures_of_Mars'):"))
        self.cat_input = QLineEdit()
        layout.addWidget(self.cat_input)

        # Username Input
        layout.addWidget(QLabel("Filter by Username (Optional - leave empty for all):"))
        self.user_input = QLineEdit()
        self.user_input.setPlaceholderText("Only download files uploaded by this user")
        layout.addWidget(self.user_input)

        # Start Button
        self.btn_start = QPushButton("Start Download")
        self.btn_start.clicked.connect(self.start_download)
        layout.addWidget(self.btn_start)

        # Stop Button
        self.btn_stop = QPushButton("Stop")
        self.btn_stop.clicked.connect(self.stop_download)
        self.btn_stop.setEnabled(False)
        layout.addWidget(self.btn_stop)

        # Progress Bar
        self.progress = QProgressBar()
        layout.addWidget(self.progress)

        # Logs
        layout.addWidget(QLabel("Logs:"))
        self.log_view = QTextEdit()
        self.log_view.setReadOnly(True)
        layout.addWidget(self.log_view)

        self.setLayout(layout)

    def log(self, message):
        self.log_view.append(message)

    def start_download(self):
        category = self.cat_input.text()
        if not category:
            QMessageBox.warning(self, "Error", "Please enter a category name.")
            return

        username = self.user_input.text()
        # Polite User-Agent is required by Wikimedia API
        user_agent = "PyQt6CommonsDownloader/1.0 (trolleway@yandex.ru)"

        self.worker = DownloadWorker(category, username, user_agent)
        self.worker.log_signal.connect(self.log)
        self.worker.progress_signal.connect(self.progress.setValue)
        self.worker.finished_signal.connect(self.on_finished)
        
        self.btn_start.setEnabled(False)
        self.btn_stop.setEnabled(True)
        self.worker.start()

    def stop_download(self):
        if hasattr(self, 'worker'):
            self.worker.stop()
            self.log("🛑 Stopping... please wait for current file to finish.")

    def on_finished(self):
        self.btn_start.setEnabled(True)
        self.btn_stop.setEnabled(False)
        self.log("--- Thread Finished ---")

if __name__ == "__main__":
    app = QApplication(sys.argv)
    window = App()
    window.show()
    sys.exit(app.exec())