core/auto_deploy.py

# core/auto_deploy.py
"""
Auto-Deploy – Utilise l'API Hugging Face Hub pour uploader des fichiers.
Contourne les problèmes d'authentification Git.
"""
import os
import logging
from pathlib import Path
from typing import Optional

try:
    from huggingface_hub import upload_file, upload_folder, HfApi
    HF_HUB_AVAILABLE = True
except ImportError:
    HF_HUB_AVAILABLE = False
    logging.warning("⚠️ huggingface_hub non installé – auto-déploiement désactivé.")

logger = logging.getLogger("lucie.auto_deploy")

class AutoDeploy:
    def __init__(self, repo_dir: str = "/app", token_env: str = "HF_TOKEN"):
        self.repo_dir = Path(repo_dir)
        self.token = os.environ.get(token_env)
        self.repo_id = None
        self._detect_repo_id()
        if not self.token:
            logger.warning("⚠️ HF_TOKEN non défini – auto-déploiement désactivé.")

    def _detect_repo_id(self):
        """Détecte le nom du Space à partir du remote git ou des variables d'env."""
        try:
            import subprocess
            res = subprocess.run(
                ["git", "remote", "get-url", "origin"],
                cwd=self.repo_dir,
                capture_output=True,
                text=True,
                check=True
            )
            url = res.stdout.strip()
            if "spaces/" in url:
                parts = url.split("spaces/")[-1].split("/")
                if len(parts) >= 2:
                    self.repo_id = f"{parts[0]}/{parts[1].replace('.git', '')}"
            elif "huggingface.co/" in url:
                parts = url.split("huggingface.co/")[-1].split("/")
                if len(parts) >= 2:
                    self.repo_id = f"{parts[0]}/{parts[1].replace('.git', '')}"
        except Exception as e:
            logger.warning(f"⚠️ Impossible de détecter le repo : {e}")

        if not self.repo_id:
            space = os.environ.get("SPACE_NAME")
            user = os.environ.get("HF_USER")
            if space and user:
                self.repo_id = f"{user}/{space}"
        if not self.repo_id:
            logger.warning("⚠️ Repo ID non détecté – déploiement désactivé.")

    def is_configured(self) -> bool:
        return self.token is not None and self.repo_id is not None and HF_HUB_AVAILABLE

    def upload_file(self, local_path: Path, remote_path: str, commit_message: str = "Auto-update by LUCIE") -> bool:
        if not self.is_configured():
            logger.error("❌ Auto-déploiement non configuré.")
            return False
        try:
            api = HfApi(token=self.token)
            api.upload_file(
                path_or_fileobj=str(local_path),
                path_in_repo=remote_path,
                repo_id=self.repo_id,
                repo_type="space",
                commit_message=commit_message,
            )
            logger.info(f"✅ Fichier {remote_path} uploadé avec succès.")
            return True
        except Exception as e:
            logger.error(f"❌ Erreur upload : {e}")
            return False

    def push_file(self, file_path: str, commit_message: str = "Auto-update by LUCIE") -> bool:
        full_path = self.repo_dir / file_path
        if not full_path.exists():
            logger.error(f"❌ Fichier {file_path} introuvable.")
            return False
        return self.upload_file(full_path, file_path, commit_message)

    def push_lora(self, lora_dir: str) -> bool:
        if not self.is_configured():
            return False
        try:
            api = HfApi(token=self.token)
            api.upload_folder(
                folder_path=lora_dir,
                path_in_repo="lora",
                repo_id=self.repo_id,
                repo_type="space",
                commit_message="Auto-update LoRA by LUCIE",
                ignore_patterns=["*.pt", "*.pth", "*.bin"],
            )
            logger.info("✅ Dossier LoRA uploadé avec succès.")
            return True
        except Exception as e:
            logger.error(f"❌ Erreur upload LoRA : {e}")
            return False