direct_upload.py

#!/usr/bin/env python
"""
Direct uploader for Hugging Face Spaces - simpler approach
"""
import os
import sys
import requests
from getpass import getpass

def upload_file(file_path, space_id, token):
    """Upload a single file to Hugging Face Space using API"""
    # Read the file content
    with open(file_path, 'rb') as f:
        file_content = f.read()
    
    # Construct the API URL
    api_url = f"https://huggingface.co/api/spaces/{space_id}/upload/{file_path}"
    
    # Set headers
    headers = {
        "Authorization": f"Bearer {token}"
    }
    
    # Upload the file
    response = requests.post(
        api_url,
        headers=headers,
        files={"file": (os.path.basename(file_path), file_content)}
    )
    
    # Check response
    if response.status_code == 200:
        return True
    else:
        print(f"Error: {response.status_code} - {response.text}")
        return False

def main():
    print("=" * 60)
    print("  Direct File Upload to Hugging Face Space")
    print("=" * 60)
    
    # Get required information
    print("\nPlease enter your Hugging Face details:")
    username = input("Username: ")
    token = getpass("Access Token: ")
    
    # Space name with validation
    while True:
        space_name = input("Space Name (without username prefix): ")
        if "/" in space_name:
            print("Error: Please enter just the Space name without username or slashes")
        else:
            break
    
    # Construct full space ID
    space_id = f"{username}/{space_name}"
    
    # Validate the space exists
    print(f"\nValidating Space: {space_id}")
    validation_url = f"https://huggingface.co/api/spaces/{space_id}"
    headers = {"Authorization": f"Bearer {token}"}
    
    try:
        validation_response = requests.get(validation_url, headers=headers)
        if validation_response.status_code != 200:
            print(f"Error: Space '{space_id}' not found or not accessible.")
            print(f"Please check the Space name and your permissions.")
            print(f"Space URL would be: https://huggingface.co/spaces/{space_id}")
            return False
        else:
            print(f"✅ Space found! URL: https://huggingface.co/spaces/{space_id}")
    except Exception as e:
        print(f"Error validating space: {e}")
        return False
    
    # Files to upload
    files_to_upload = [
        "app/core/memory.py",
        "app/core/ingestion.py",
        "app/ui/streamlit_app.py"
    ]
    
    # Verify files exist locally
    missing_files = [f for f in files_to_upload if not os.path.exists(f)]
    if missing_files:
        print(f"Error: The following files don't exist locally: {missing_files}")
        return False
    
    # Upload each file
    print("\nUploading files:")
    success_count = 0
    
    for file_path in files_to_upload:
        print(f"📤 Uploading {file_path}... ", end="", flush=True)
        if upload_file(file_path, space_id, token):
            print("✅ Success!")
            success_count += 1
        else:
            print("❌ Failed!")
    
    # Print summary
    print(f"\nUpload summary: {success_count}/{len(files_to_upload)} files uploaded successfully.")
    
    if success_count == len(files_to_upload):
        print("\n✅ All files uploaded successfully!")
        print(f"View your Space at: https://huggingface.co/spaces/{space_id}")
        print("Note: It may take a few minutes for your Space to rebuild with the new changes.")
        return True
    else:
        print("\n⚠️ Some files failed to upload. Please check the errors above.")
        return False

if __name__ == "__main__":
    try:
        # Check for required packages
        try:
            import requests
        except ImportError:
            print("Installing required package: requests")
            import subprocess
            subprocess.check_call([sys.executable, "-m", "pip", "install", "requests"])
            import requests
        
        # Run the main function
        success = main()
        if success:
            sys.exit(0)
        else:
            sys.exit(1)
    except KeyboardInterrupt:
        print("\nUpload cancelled by user.")
        sys.exit(1)
    except Exception as e:
        print(f"\nUnexpected error: {e}")
        sys.exit(1)