classroom_topic_export.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

"""
Export Google Classroom (UN SEUL COURS) par Thème (Topic) :
- Récupère les Topics du cours
- Récupère les Devoirs (courseWork) et Ressources (courseWorkMaterials)
- Télécharge les pièces jointes Drive (get_media + export Google Docs/Sheets/Slides)
- Produit :
  /Archive/<Cours>/<Topic>/fichiers/...
  /Archive/<Cours>/index.csv (global)
  /Archive/<Cours>/<Topic>/index_topic.csv (par thème)
  /Archive/<Cours>/scenario_pedagogique.md (chronologie, regroupé par thème)

Exécution :
  python classroom_topic_export.py --course-id <ID>    # recommandé si tu connais l'ID
  python classroom_topic_export.py --course-name "Nom du cours exact"
  python classroom_topic_export.py                     # affichera la liste et demandera à choisir

Prérequis :
- credentials.json (OAuth client desktop) dans le même dossier
- pip install -r requirements.txt (voir plus bas)
"""

import os
import io
import re
import csv
import sys
import json
import time
import argparse
import logging
import datetime
from typing import Dict, List, Any, Optional, Tuple

from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
from googleapiclient.http import MediaIoBaseDownload
from google_auth_oauthlib.flow import InstalledAppFlow
from google.auth.transport.requests import Request
from google.oauth2.credentials import Credentials

# ----------------- CONFIG -----------------
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
ARCHIVE_ROOT = "Archive"
TOKEN_FILE = os.path.join(SCRIPT_DIR, "token.json")


# Scopes minimaux (readonly)
SCOPES = [
    "https://www.googleapis.com/auth/classroom.courses.readonly",
    "https://www.googleapis.com/auth/classroom.topics.readonly",
    "https://www.googleapis.com/auth/classroom.courseworkmaterials.readonly",
    "https://www.googleapis.com/auth/classroom.coursework.students.readonly",
    "https://www.googleapis.com/auth/classroom.coursework.me.readonly",  # <-- AJOUT
    "https://www.googleapis.com/auth/drive.readonly",
]


# Export par défaut pour fichiers Google Workspace
EXPORT_MAP = {
    "application/vnd.google-apps.document": ("application/pdf", ".pdf"),
    "application/vnd.google-apps.spreadsheet": ("application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", ".xlsx"),
    "application/vnd.google-apps.presentation": ("application/pdf", ".pdf"),
    "application/vnd.google-apps.drawing": ("application/pdf", ".pdf"),
}

RETRY_MAX = 6
RETRY_BASE_SLEEP = 1.0

# ------------------------------------------


def log_setup():
    logging.basicConfig(
        level=logging.INFO,
        format="%(message)s",
        handlers=[logging.StreamHandler(sys.stdout)]
    )


def ensure_dir(p: str):
    os.makedirs(p, exist_ok=True)


def sanitize_filename(name: str, max_len: int = 150) -> str:
    cleaned = re.sub(r'[\/\\\:\*\?\"\<\>\|]+', "_", name).strip()
    if not cleaned:
        cleaned = "untitled"
    if len(cleaned) > max_len:
        root, ext = os.path.splitext(cleaned)
        cleaned = root[: max_len - len(ext)] + ext
    return cleaned


def with_retries(fn, *args, **kwargs):
    delay = RETRY_BASE_SLEEP
    for attempt in range(1, RETRY_MAX + 1):
        try:
            return fn(*args, **kwargs)
        except HttpError as e:
            status = getattr(e, "status_code", None) or (e.resp.status if hasattr(e, "resp") else None)
            if status in (429, 500, 502, 503, 504):
                logging.warning(f"HTTP {status} - tentative {attempt}/{RETRY_MAX}, retry dans {delay:.1f}s…")
                time.sleep(delay)
                delay *= 2
                continue
            raise
        except Exception:
            if attempt == RETRY_MAX:
                raise
            time.sleep(delay)
            delay *= 2


def get_credentials() -> Credentials:
    creds = None
    if os.path.exists(TOKEN_FILE):
        try:
            creds = Credentials.from_authorized_user_file(TOKEN_FILE, SCOPES)
        except Exception as e:
            print(f"[auth] token.json illisible → on repart propre ({e})")
            creds = None

    if not creds or not creds.valid:
        if creds and creds.expired and creds.refresh_token:
            with_retries(creds.refresh, Request())
        else:
            # Choix du mode d'auth selon args globaux
            use_console = getattr(sys.modules["__main__"], "ARGS_CONSOLE_AUTH", False)
            flow = InstalledAppFlow.from_client_secrets_file("credentials.json", SCOPES)

            if use_console:
                # Fallback sans serveur local (copie/colle du code)
                creds = flow.run_console()
            else:
                # Serveur local (ouvre le navigateur)
                creds = flow.run_local_server(
                    port=0,
                    authorization_prompt_message="Veuillez autoriser l'accès dans votre navigateur…",
                    success_message="Autorisation validée, vous pouvez fermer cette fenêtre.",
                    open_browser=True,
                )

        # Écrit toujours le token s’il est valide
        with open(TOKEN_FILE, "w", encoding="utf-8") as f:
            f.write(creds.to_json())
        print(f"[auth] token écrit ici : {TOKEN_FILE}")

    return creds

def build_services(creds: Credentials):
    classroom = with_retries(build, "classroom", "v1", credentials=creds)
    drive = with_retries(build, "drive", "v3", credentials=creds)
    return classroom, drive


def list_courses(classroom) -> List[Dict[str, Any]]:
    courses = []
    page_token = None
    while True:
        req = classroom.courses().list(pageToken=page_token)
        resp = with_retries(req.execute)
        courses.extend(resp.get("courses", []))
        page_token = resp.get("nextPageToken")
        if not page_token:
            break
    return courses


def choose_course(classroom, course_id: Optional[str], course_name: Optional[str]) -> Dict[str, Any]:
    courses = list_courses(classroom)
    if not courses:
        raise SystemExit("Aucun cours trouvé sur ce compte.")
    if course_id:
        for c in courses:
            if c.get("id") == course_id:
                return c
        raise SystemExit(f"Course ID introuvable: {course_id}")
    if course_name:
        for c in courses:
            if c.get("name") == course_name:
                return c
        raise SystemExit(f'Course name introuvable: "{course_name}" (respecter la casse).')

    # Sinon, affichage menu
    logging.info("\nCours disponibles :")
    for i, c in enumerate(courses, 1):
        logging.info(f"{i:2d}. {c.get('name')}  (id={c.get('id')})")
    choice = input("\nNuméro du cours à exporter : ").strip()
    try:
        idx = int(choice) - 1
        assert 0 <= idx < len(courses)
    except Exception:
        raise SystemExit("Choix invalide.")
    return courses[idx]


def paginate_classroom_list(service, resource: str, course_id: str, list_key: str, extra_params: Optional[Dict[str, Any]] = None):
    if extra_params is None:
        extra_params = {}
    page_token = None
    while True:
        params = dict(extra_params)
        if page_token:
            params["pageToken"] = page_token
        if resource == "topics":
            req = service.courses().topics().list(courseId=course_id, **params)
        elif resource == "courseWork":
            req = service.courses().courseWork().list(courseId=course_id, **params)
        elif resource == "courseWorkMaterials":
            req = service.courses().courseWorkMaterials().list(courseId=course_id, **params)
        else:
            raise ValueError("Ressource inconnue")
        resp = with_retries(req.execute)
        for item in resp.get(list_key, []):
            return_item = dict(item)
            yield return_item
        page_token = resp.get("nextPageToken")
        if not page_token:
            break


def drive_get_metadata(drive, file_id: str) -> Dict[str, Any]:
    req = drive.files().get(
        fileId=file_id,
        fields="id,name,mimeType,modifiedTime,size,webViewLink,webContentLink"
    )
    return with_retries(req.execute)


def drive_download(drive, file_id: str, target_path: str, mime_type: Optional[str]):
    # Si fichier Google -> export, sinon get_media
    if mime_type and mime_type.startswith("application/vnd.google-apps"):
        export_mime, ext = EXPORT_MAP.get(mime_type, ("application/pdf", ".pdf"))
        if not target_path.lower().endswith(ext):
            target_path += ext
        req = drive.files().export(fileId=file_id, mimeType=export_mime)
    else:
        req = drive.files().get_media(fileId=file_id)

    os.makedirs(os.path.dirname(target_path), exist_ok=True)
    fh = io.BytesIO()
    downloader = MediaIoBaseDownload(fh, req)
    done = False
    while not done:
        status, done = with_retries(downloader.next_chunk)
    with open(target_path, "wb") as f:
        f.write(fh.getvalue())


def material_iter(materials: Optional[List[Dict[str, Any]]]):
    if not materials:
        return
    for m in materials:
        for k in ("driveFile", "link", "youtubeVideo", "form"):
            if k in m and m[k]:
                yield (k, m[k])


def extract_drive_file_info(payload: Dict[str, Any]) -> Dict[str, Any]:
    """Normalize Classroom driveFile payloads into a dict with id/title/link.

    Depending on the API version, `payload` can be:
    - {"driveFile": {"driveFile": { ... }, "shareMode": "VIEW"}}
    - {"driveFile": { ... , "shareMode": "VIEW"}}
    - Directly {"id": ..., "title": ...}
    We look through those patterns and return the innermost dict that has an `id`.
    """

    if not isinstance(payload, dict):
        raise ValueError(f"Unexpected driveFile payload type: {type(payload)}")

    # Candidates we will inspect (outer -> inner)
    candidates: List[Dict[str, Any]] = []
    direct = payload
    if isinstance(direct, dict):
        candidates.append(direct)
        inner = direct.get("driveFile")
        if isinstance(inner, dict):
            candidates.append(inner)
            deeper = inner.get("driveFile")
            if isinstance(deeper, dict):
                candidates.append(deeper)

    for candidate in candidates:
        file_id = candidate.get("id") if isinstance(candidate, dict) else None
        if file_id:
            return {
                "id": file_id,
                "title": candidate.get("title") or candidate.get("name"),
                "alternateLink": candidate.get("alternateLink"),
                "thumbnailUrl": candidate.get("thumbnailUrl"),
            }

    raise KeyError("driveFile")

def parse_iso(iso: Optional[str]) -> Optional[datetime.datetime]:
    if not iso:
        return None
    try:
        return datetime.datetime.fromisoformat(iso.replace("Z", "+00:00")).astimezone()
    except Exception:
        return None

def iso_to_dtstr(iso: Optional[str]) -> str:
    if not iso:
        return ""
    try:
        dt = datetime.datetime.fromisoformat(iso.replace("Z", "+00:00"))
        return dt.astimezone().strftime("%Y-%m-%d %H:%M")
    except Exception:
        return iso


def main():
    log_setup()
    parser = argparse.ArgumentParser(description="Export Classroom (un cours) par thème.")
    parser.add_argument("--course-id", type=str, help="ID du cours Classroom")
    parser.add_argument("--course-name", type=str, help="Nom EXACT du cours Classroom")
    parser.add_argument("--no-download", action="store_true", help="Ne pas télécharger les fichiers, seulement indexer")
    parser.add_argument("--reset-auth", action="store_true", help="Supprime token.json et relance l'auth")
    parser.add_argument("--console-auth", action="store_true", help="Utiliser le flux console (pas de serveur local/balai de navigateur)")
    parser.add_argument("--order", choices=["classroom", "created-desc", "created-asc"],
                    default="created-desc",
                    help="Ordre dans le scénario: 'classroom' (ordre renvoyé par l'API), "
                         "'created-desc' (création du + récent au + ancien), "
                         "'created-asc' (création du + ancien au + récent)")

    args = parser.parse_args()


    creds = get_credentials()
    classroom, drive = build_services(creds)

    course = choose_course(classroom, args.course_id, args.course_name)
    course_id = course["id"]
    course_name = course["name"]
    logging.info(f"\n== Cours sélectionné : {course_name} (id={course_id}) ==")

    # Dossiers
    course_dir = os.path.join(ARCHIVE_ROOT, sanitize_filename(course_name))
    files_root = os.path.join(course_dir)  # un dossier par topic en dessous
    ensure_dir(course_dir)

    # 1) Récup Topics
    topics = list(paginate_classroom_list(classroom, "topics", course_id, "topic"))
    topic_map = {t["topicId"]: t.get("name", f"Topic_{t['topicId']}") for t in topics}
    if not topic_map:
        logging.info("Aucun Thème (Topic) trouvé : on placera tout dans _Sans_Theme.")
    topic_map[None] = "_Sans_Theme"

    # 2) Récup courseWork (devoirs) & courseWorkMaterials (ressources)
    all_coursework = list(paginate_classroom_list(classroom, "courseWork", course_id, "courseWork"))
    all_materials = list(paginate_classroom_list(classroom, "courseWorkMaterials", course_id, "courseWorkMaterial"))
    logging.info(f"- Devoirs trouvés : {len(all_coursework)}")
    logging.info(f"- Ressources trouvées : {len(all_materials)}")

    # 3) Index global
    global_rows: List[Dict[str, Any]] = []
    scenario_items: List[Dict[str, Any]] = []

    def add_row(topic_name, item_type, title, description, created_at, due_at, attachment_type, attachment_name, attachment_id, url, local_path):
        global_rows.append({
            "topic": topic_name,
            "type": item_type,
            "title": title or "",
            "description": (description or "").replace("\n", " ").strip(),
            "created_at": created_at or "",
            "due_at": due_at or "",
            "attachment_type": attachment_type or "",
            "attachment_name": attachment_name or "",
            "attachment_id": attachment_id or "",
            "url": url or "",
            "local_path": local_path or "",
        })

    # 4) Parcours utilitaires
    def topic_name_of(item):
        tid = item.get("topicId")
        return topic_map.get(tid) or "_Sans_Theme"

    def due_to_str(dueDate, dueTime):
        if not dueDate:
            return ""
        y = dueDate.get("year", 0)
        m = dueDate.get("month", 0)
        d = dueDate.get("day", 0)
        hh = (dueTime or {}).get("hours", 0)
        mm = (dueTime or {}).get("minutes", 0)
        try:
            return datetime.datetime(y, m, d, hh, mm).strftime("%Y-%m-%d %H:%M")
        except Exception:
            return f"{y:04d}-{m:02d}-{d:02d}"

    # 5) Traiter Ressources (courseWorkMaterials)
    for mat in all_materials:
        tname = topic_name_of(mat)
        title = mat.get("title", "")
        desc = (mat.get("description") or "")
        created_iso = mat.get("creationTime")
        created = iso_to_dtstr(created_iso)
        item_folder = os.path.join(files_root, sanitize_filename(tname), "fichiers")
        ensure_dir(item_folder)

        attachments_for_scenario = []
        for kind, obj in material_iter(mat.get("materials")):
            local_path = ""
            url = ""
            att_name = ""
            att_id = ""

            if kind == "driveFile":
                try:
                    info = extract_drive_file_info(obj)
                    att_id = info["id"]
                    meta = drive_get_metadata(drive, att_id)
                    att_name_raw = meta.get("name") or info.get("title") or att_id
                    att_name = sanitize_filename(att_name_raw)
                    mime = meta.get("mimeType")
                    target_base = os.path.join(item_folder, att_name)
                    candidate_paths = [target_base]
                    for _, ext in EXPORT_MAP.values():
                        if not target_base.lower().endswith(ext):
                            candidate_paths.append(target_base + ext)

                    if not args.no_download:
                        if not any(os.path.exists(p) for p in candidate_paths):
                            drive_download(drive, att_id, target_base, mime)

                    for candidate in candidate_paths:
                        if os.path.exists(candidate):
                            local_path = os.path.relpath(candidate, start=course_dir)
                            break
                    if not local_path:
                        local_path = os.path.relpath(target_base, start=course_dir)

                    url_candidates = [
                        meta.get("webViewLink"),
                        meta.get("webContentLink"),
                        info.get("alternateLink"),
                    ]
                    url = next((u for u in url_candidates if u), url)
                except Exception as e:
                    logging.warning(f"Échec téléchargement Drive (resource): {e}")
            elif kind == "link":
                att_name = obj.get("title") or "Lien"
                url = obj.get("url", "")
            elif kind == "youtubeVideo":
                att_name = obj.get("title") or "YouTube"
                url = obj.get("alternateLink", "")
            elif kind == "form":
                att_name = obj.get("title") or "Formulaire"
                url = obj.get("formUrl", "") or obj.get("responseUrl", "")

            add_row(tname, "ressource", title, desc, created, "", kind, att_name, att_id, url, local_path)
            attachments_for_scenario.append({"name": att_name, "url": url, "local_path": local_path})

        scenario_items.append({
            "type": "Ressource",
            "topic": tname,
            "title": title,
            "description": desc,
            "createdAt": created,
            "createdAtISO": created_iso, 
            "due": "",
            "attachments": attachments_for_scenario
        })

    # 6) Traiter Devoirs (courseWork)
    for cw in all_coursework:
        tname = topic_name_of(cw)
        title = cw.get("title", "")
        desc = cw.get("description") or ""
        created_iso = cw.get("creationTime")
        created = iso_to_dtstr(created_iso)
        due = due_to_str(cw.get("dueDate"), cw.get("dueTime"))
        item_folder = os.path.join(files_root, sanitize_filename(tname), "fichiers")
        ensure_dir(item_folder)

        attachments_for_scenario = []
        # Matériaux du devoir (pas les remises)
        for kind, obj in material_iter(cw.get("materials")):
            local_path = ""
            url = ""
            att_name = ""
            att_id = ""

            if kind == "driveFile":
                try:
                    info = extract_drive_file_info(obj)
                    att_id = info["id"]
                    meta = drive_get_metadata(drive, att_id)
                    att_name_raw = meta.get("name") or info.get("title") or att_id
                    att_name = sanitize_filename(att_name_raw)
                    mime = meta.get("mimeType")
                    target_base = os.path.join(item_folder, att_name)
                    candidate_paths = [target_base]
                    for _, ext in EXPORT_MAP.values():
                        if not target_base.lower().endswith(ext):
                            candidate_paths.append(target_base + ext)

                    if not args.no_download:
                        if not any(os.path.exists(p) for p in candidate_paths):
                            drive_download(drive, att_id, target_base, mime)

                    for candidate in candidate_paths:
                        if os.path.exists(candidate):
                            local_path = os.path.relpath(candidate, start=course_dir)
                            break
                    if not local_path:
                        local_path = os.path.relpath(target_base, start=course_dir)

                    url_candidates = [
                        meta.get("webViewLink"),
                        meta.get("webContentLink"),
                        info.get("alternateLink"),
                    ]
                    url = next((u for u in url_candidates if u), url)
                except Exception as e:
                    logging.warning(f"Échec téléchargement Drive (devoir): {e}")
            elif kind == "link":
                att_name = obj.get("title") or "Lien"
                url = obj.get("url", "")
            elif kind == "youtubeVideo":
                att_name = obj.get("title") or "YouTube"
                url = obj.get("alternateLink", "")
            elif kind == "form":
                att_name = obj.get("title") or "Formulaire"
                url = obj.get("formUrl", "") or obj.get("responseUrl", "")

            add_row(tname, "devoir", title, desc, created, due, kind, att_name, att_id, url, local_path)
            attachments_for_scenario.append({"name": att_name, "url": url, "local_path": local_path})

        scenario_items.append({
            "type": "Devoir",
            "topic": tname,
            "title": title,
            "description": desc,
            "createdAt": created,
            "createdAtISO": created_iso,
            "due": due,
            "attachments": attachments_for_scenario
        })

    # 7) CSV global
    ensure_dir(course_dir)
    csv_global = os.path.join(course_dir, "index.csv")
    with open(csv_global, "w", encoding="utf-8-sig", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=[
            "topic","type","title","description","created_at","due_at",
            "attachment_type","attachment_name","attachment_id","url","local_path"
        ])
        writer.writeheader()
        writer.writerows(global_rows)
    logging.info(f"\n✔ index global : {csv_global}")

    # 8) CSV par topic
    by_topic: Dict[str, List[Dict[str, Any]]] = {}
    for r in global_rows:
        by_topic.setdefault(r["topic"], []).append(r)
    for tname, rows in by_topic.items():
        tdir = os.path.join(files_root, sanitize_filename(tname))
        ensure_dir(tdir)
        path = os.path.join(tdir, "index_topic.csv")
        with open(path, "w", encoding="utf-8-sig", newline="") as f:
            writer = csv.DictWriter(f, fieldnames=list(global_rows[0].keys()))
            writer.writeheader()
            writer.writerows(rows)

    # 9) Scénario pédagogique (md)
    scen_path = os.path.join(course_dir, "scenario_pedagogique.md")
    with open(scen_path, "w", encoding="utf-8") as md:
        md.write(f"# Scénario pédagogique — {course_name}\n\n")
        md.write(f"_Généré le {datetime.datetime.now().astimezone().isoformat()}_\n\n")

        # On regroupe par thème, en conservant l'ordre d'apparition (dict insertion order)
        by_topic_scen: Dict[str, List[Dict[str, Any]]] = {}
        for it in scenario_items:
            by_topic_scen.setdefault(it.get("topic") or "_Sans_Theme", []).append(it)

        # Écriture par thème, et tri interne selon --order
        for topic_name, items in by_topic_scen.items():
            md.write(f"\n## Thème : {topic_name}\n")

            if args.order == "classroom":
                ordered = items  # on respecte l'ordre de l'API (proche de Classroom)
            else:
                # Tri par date de création (pas update), desc ou asc
                reverse = (args.order == "created-desc")
                ordered = sorted(items, key=lambda x: (parse_iso(x.get("createdAtISO")) or datetime.datetime.min.replace(tzinfo=datetime.timezone.utc),
                                                    x.get("title") or ""),
                                reverse=reverse)

            for it in ordered:
                md.write(f"\n### [{it['type']}] {it['title'] or '(sans titre)'}\n")
                if it.get("createdAt"):
                    md.write(f"- Créé le : {it['createdAt']}\n")
                if it.get("due"):
                    md.write(f"- Échéance : {it['due']}\n")
                if it.get("description"):
                    md.write(f"\n{it['description']}\n")
                if it.get("attachments"):
                    md.write("\n**Pièces/ressources :**\n")
                    for att in it["attachments"]:
                        label = att.get("name") or "pièce"
                        if att.get("local_path"):
                            md.write(f"- {label} → `{att['local_path']}`\n")
                        elif att.get("url"):
                            md.write(f"- {label} → {att['url']}\n")

    logging.info(f"✔ scénario : {scen_path}")


    logging.info("\nTerminé ✅")


if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        print("\nInterrompu.")