shuidrop_gui/.gitea/scripts/upload_installer_to_ks3.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Upload GUI installer to KS3 object storage
- Target directory: installers/
- Set proper HTTP headers for browser download
"""
import os
import sys
import logging
from pathlib import Path

try:
    from ks3.connection import Connection
except ImportError:
    print("ERROR: ks3sdk not installed. Please run: pip install ks3sdk")
    sys.exit(1)

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s [%(levelname)s] %(message)s'
)
logger = logging.getLogger(__name__)

# KS3 Configuration (using actual project config)
KS3_ACCESS_KEY = 'AKLT0Ey7Nq7ZSXykki0X0RGG'
KS3_SECRET_KEY = 'OONxkt9wwJa1FIco21vCbirs1HB6AGzDWdRyV0k2'
KS3_ENDPOINT = 'ks3-cn-guangzhou.ksyuncs.com'  # Guangzhou region
KS3_BUCKET = 'shuidrop-chat-server'
KS3_IS_SECURE = True  # Use HTTPS
KS3_PREFIX = 'installers/'


def get_ks3_connection():
    """Get KS3 connection"""
    try:
        connection = Connection(
            KS3_ACCESS_KEY,
            KS3_SECRET_KEY,
            host=KS3_ENDPOINT,
            is_secure=KS3_IS_SECURE,
            timeout=1800,  # Increase timeout to 30 minutes for large files
        )
        logger.info(f"KS3 connection established: {KS3_ENDPOINT}")
        return connection
    except Exception as e:
        logger.error(f"KS3 connection failed: {e}")
        return None


def find_latest_installer():
    """Find the latest generated installer"""
    project_root = Path(__file__).parent.parent.parent
    installer_dir = project_root / 'installer' / 'output'
    
    if not installer_dir.exists():
        logger.error(f"Installer directory not found: {installer_dir}")
        return None
    
    installers = list(installer_dir.glob('*.exe'))
    if not installers:
        logger.error(f"No installer files found in: {installer_dir}")
        return None
    
    latest_installer = max(installers, key=lambda p: p.stat().st_mtime)
    file_size_mb = latest_installer.stat().st_size / 1024 / 1024
    
    logger.info(f"Found installer: {latest_installer.name}")
    logger.info(f"File size: {file_size_mb:.2f} MB")
    
    return latest_installer


def progress_callback(uploaded_bytes, total_bytes):
    """Upload progress callback"""
    if total_bytes > 0:
        percentage = (uploaded_bytes / total_bytes) * 100
        uploaded_mb = uploaded_bytes / 1024 / 1024
        total_mb = total_bytes / 1024 / 1024
        logger.info(f"Upload progress: {percentage:.1f}% ({uploaded_mb:.1f}/{total_mb:.1f} MB)")


def upload_installer(connection, installer_path):
    """Upload installer to KS3 with progress tracking"""
    try:
        bucket = connection.get_bucket(KS3_BUCKET)
        ks3_key = f"{KS3_PREFIX}{installer_path.name}"
        
        file_size = installer_path.stat().st_size
        file_size_mb = file_size / 1024 / 1024
        
        logger.info(f"Starting upload to KS3...")
        logger.info(f"Target path: {ks3_key}")
        logger.info(f"File size: {file_size_mb:.2f} MB")
        logger.info(f"Estimated time: ~{int(file_size_mb / 2)} seconds (assuming 2MB/s)")
        logger.info("Please wait, this may take several minutes...")
        
        key = bucket.new_key(ks3_key)
        
        # Upload file with public read permission and progress tracking
        # Use cb parameter for progress callback (called every 5% or every 10MB)
        key.set_contents_from_filename(
            str(installer_path),
            headers={
                'Content-Type': 'application/octet-stream',
                'Content-Disposition': f'attachment; filename="{installer_path.name}"',
                'Cache-Control': 'public, max-age=3600',
                'x-kss-storage-class': 'STANDARD',
                'x-kss-acl': 'public-read'  # Set public read permission in headers
            },
            policy='public-read',  # Set ACL policy
            cb=progress_callback,  # Progress callback function
            num_cb=10  # Call callback 10 times during upload (every 10%)
        )
        
        # Generate download URL (using KS3 third-level domain format)
        # Format: https://{bucket}.{endpoint}/{key}
        protocol = 'https' if KS3_IS_SECURE else 'http'
        download_url = f"{protocol}://{KS3_BUCKET}.{KS3_ENDPOINT}/{ks3_key}"
        
        logger.info("")
        logger.info("=" * 70)
        logger.info("Upload successful!")
        logger.info(f"Download URL: {download_url}")
        logger.info("=" * 70)
        
        return download_url
    except Exception as e:
        logger.error("")
        logger.error("=" * 70)
        logger.error(f"Upload failed: {e}")
        logger.error("=" * 70)
        import traceback
        logger.error(traceback.format_exc())
        return None


def main():
    """Main function"""
    logger.info("=" * 70)
    logger.info("Starting GUI installer upload to KS3")
    logger.info(f"KS3 Endpoint: {KS3_ENDPOINT}")
    logger.info(f"Bucket: {KS3_BUCKET}")
    logger.info(f"Target directory: {KS3_PREFIX}")
    logger.info("=" * 70)
    
    installer_path = find_latest_installer()
    if not installer_path:
        return 1
    
    connection = get_ks3_connection()
    if not connection:
        return 1
    
    download_url = upload_installer(connection, installer_path)
    if not download_url:
        return 1
    
    logger.info("=" * 70)
    logger.info("Task completed successfully!")
    logger.info(f"Installer: {installer_path.name}")
    logger.info(f"Download URL: {download_url}")
    logger.info("=" * 70)
    
    return 0


if __name__ == '__main__':
    sys.exit(main())