Transfert récursif de dossiers Google Drive vers Whaller (via API)

gdrive_to_whaller.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
gdrive_to_whaller.py

Transfert récursif de dossiers Google Drive vers Whaller (via API)
- Recrée l'arborescence
- Gère les fichiers
- Supporte un mode dry-run
- Log JSON et vérification de l'existence des dossiers

📦 Modules à installer :
google-api-python-client google-auth google-auth-oauthlib requests

Auteur : Sébastien L.
Date : 2025-03-26
Version : 0.0.1
Licence : MIT (voir https://opensource.org/licenses/MIT)
"""

__author__ = "Sébastien L."
__contributors__ = ["Edouard A.", "Kim Jong Un"]
__version__ = "0.0.1"
__license__ = "MIT"

import io
import os
import json
from typing import Optional, List, Dict, Any
import requests
from google.oauth2.credentials import Credentials
from google_auth_oauthlib.flow import InstalledAppFlow
from googleapiclient.discovery import build
from googleapiclient.http import MediaIoBaseDownload

# ========================== CONFIGURATION ========================== #
WHALLER_TOKEN: str = "TON_TOKEN_ICI"
WHALLER_SPHERE_ID: str = "ID_DE_LA_SPHERE"
GOOGLE_DRIVE_FOLDER_ID: str = "ID_DU_DOSSIER_DRIVE"
SCOPES: List[str] = ['https://www.googleapis.com/auth/drive.readonly']
DRY_RUN: bool = True
LOG_PATH: str = "gdrive_to_whaller.log.json"
# =================================================================== #

log: Dict[str, List[Dict[str, Any]]] = {"folders": [], "files": []}


def get_drive_service():
    """Authentifie l'utilisateur et retourne un client Google Drive API."""
    if DRY_RUN:
        print("[DRY-RUN] Pas besoin de credentials.json, l'accès à Google Drive est simulé.")
        return None
    if not os.path.exists("credentials.json"):
        raise FileNotFoundError("credentials.json introuvable. Nécessaire pour l'accès à Google Drive.")
    flow = InstalledAppFlow.from_client_secrets_file("credentials.json", SCOPES)
    creds: Credentials = flow.run_local_server(port=0)
    return build('drive', 'v3', credentials=creds)


def list_drive_files(service, folder_id: str) -> List[Dict[str, Any]]:
    """Liste les fichiers et dossiers contenus dans un dossier Google Drive."""
    if DRY_RUN:
        # Retourne une simulation vide ou fictive
        print(f"[DRY-RUN] list_drive_files ignoré pour le dossier {folder_id}")
        return []
    query = f"'{folder_id}' in parents and trashed=false"
    result = service.files().list(q=query, fields="files(id, name, mimeType)").execute()
    return result.get('files', [])


def download_file(service, file_id: str, filename: str) -> io.BytesIO:
    """Télécharge un fichier de Google Drive et retourne son contenu en BytesIO."""
    request = service.files().get_media(fileId=file_id)
    buffer = io.BytesIO()
    downloader = MediaIoBaseDownload(buffer, request)
    done = False
    while not done:
        _, done = downloader.next_chunk()
    buffer.seek(0)
    return buffer


def whaller_folder_exists(name: str, parent_id: Optional[str] = None) -> Optional[str]:
    """Vérifie si un dossier Whaller existe déjà (par nom et parent)."""
    url = f"https://api.whaller.com/spheres/{WHALLER_SPHERE_ID}/boxes"
    headers = {"Authorization": f"Bearer {WHALLER_TOKEN}"}
    params = {"parent_id": parent_id} if parent_id else {}
    response = requests.get(url, headers=headers, params=params)
    if response.status_code != 200:
        return None
    for box in response.json():
        if box.get("is_directory") and box.get("name") == name:
            return box["id"]
    return None


def upload_to_whaller(file_obj: Optional[io.BytesIO], filename: str, parent_id: Optional[str] = None) -> Dict[str, Any]:
    """
    Upload un fichier dans Whaller.

    Args:
        file_obj: Contenu du fichier (ou None en dry-run).
        filename: Nom du fichier.
        parent_id: ID du dossier parent.

    Returns:
        Dictionnaire JSON de la réponse (ou vide en dry-run).
    """
    if DRY_RUN:
        print(f"[DRY-RUN] UPLOAD FILE: {filename} → dans dossier ID {parent_id}")
        log["files"].append({"name": filename, "parent": parent_id})
        return {}
    url = "https://api.whaller.com/boxes"
    headers = {"Authorization": f"Bearer {WHALLER_TOKEN}"}
    files = {"uploaded_file": (filename, file_obj)} if file_obj else None
    data = {"sphere_id": WHALLER_SPHERE_ID}
    if parent_id:
        data["parent_id"] = parent_id
    response = requests.post(url, headers=headers, data=data, files=files)
    result = response.json()
    log["files"].append({"name": filename, "parent": parent_id, "id": result.get("id")})
    return result


def create_folder_in_whaller(name: str, parent_id: Optional[str] = None) -> str:
    """
    Crée un dossier dans Whaller (sauf s'il existe déjà).

    Args:
        name: Nom du dossier.
        parent_id: ID du dossier parent.

    Returns:
        ID du dossier créé (ou existant).
    """
    existing_id = whaller_folder_exists(name, parent_id)
    if existing_id:
        print(f"[SKIP] Dossier déjà existant: {name}")
        return existing_id
    if DRY_RUN:
        fake_id = f"DRYFOLDER-{name}"
        print(f"[DRY-RUN] CRÉATION DOSSIER: {name} → {parent_id} (fictif: {fake_id})")
        log["folders"].append({"name": name, "parent": parent_id, "id": fake_id})
        return fake_id
    url = "https://api.whaller.com/boxes"
    headers = {"Authorization": f"Bearer {WHALLER_TOKEN}"}
    data = {
        "sphere_id": WHALLER_SPHERE_ID,
        "name": name,
        "is_directory": True
    }
    if parent_id:
        data["parent_id"] = parent_id
    response = requests.post(url, headers=headers, data=data)
    result = response.json()
    log["folders"].append({"name": name, "parent": parent_id, "id": result.get("id")})
    return result["id"]


def transfer_folder(service, folder_id: str, parent_whaller_id: Optional[str] = None, level: int = 0) -> None:
    """
    Transfère récursivement le contenu d'un dossier Google Drive vers Whaller.

    Args:
        service: Instance Google Drive API.
        folder_id: ID du dossier Drive source.
        parent_whaller_id: ID du dossier Whaller parent.
        level: Niveau d'indentation pour l'affichage.
    """
    indent = "  " * level
    items = list_drive_files(service, folder_id)
    for item in items:
        name = item["name"]
        if item["mimeType"] == "application/vnd.google-apps.folder":
            print(f"{indent}📁 {name}")
            whaller_folder_id = create_folder_in_whaller(name, parent_whaller_id)
            transfer_folder(service, item["id"], whaller_folder_id, level + 1)
        else:
            print(f"{indent}📄 {name}")
            file_content = None if DRY_RUN else download_file(service, item["id"], name)
            upload_to_whaller(file_content, name, parent_whaller_id)


def save_log() -> None:
    """Sauvegarde le journal d'activité dans un fichier JSON."""
    with open(LOG_PATH, "w", encoding="utf-8") as f:
        json.dump(log, f, ensure_ascii=False, indent=2)
    print(f"\n✅ Log sauvegardé dans {LOG_PATH}")


def main() -> None:
    """Point d'entrée principal du script."""
    try:
        drive_service = get_drive_service()
        if not DRY_RUN:
            transfer_folder(drive_service, GOOGLE_DRIVE_FOLDER_ID)
        else:
            print("[DRY-RUN] Simulation de transfert (aucun appel à Google Drive).")
            transfer_folder(None, GOOGLE_DRIVE_FOLDER_ID)
    finally:
        save_log()

if __name__ == "__main__":
    main()