biblioteca_conocimiento_laboratorio/scripts/tareas_hijas/drive.py

# -*- coding: utf-8 -*-
"""drive.py — Helpers para el sistema de tareas de las hijas (Google Drive).

Autentica con el token personal (proyecto Cloud "tareasnenas") y ofrece
utilidades para listar, leer, descargar, crear y mover ficheros en Drive.

Uso rapido:
    python scripts/tareas_hijas/drive.py list "TAREAS/CAROLINA/LENGUA"
    python scripts/tareas_hijas/drive.py read <DOC_ID>          # exporta texto de un Google Doc
    python scripts/tareas_hijas/drive.py pdf <FILE_ID> <out.pdf>
    python scripts/tareas_hijas/drive.py make "Nombre doc" <FOLDER_ID> fichero.html
    python scripts/tareas_hijas/drive.py move <FILE_ID> <FOLDER_ID>

Notas:
- Ejecutar con `python -X utf8` en Windows (evita errores de acentos).
- El token se refresca solo y se re-guarda en disco.
"""
from __future__ import annotations

import io
import sys
from pathlib import Path

from google.oauth2.credentials import Credentials
from google.auth.transport.requests import Request
from googleapiclient.discovery import build
from googleapiclient.http import MediaIoBaseDownload, MediaIoBaseUpload

# Ruta por defecto al token personal dentro del repo de negocio.
DEFAULT_TOKEN = Path(
    r"C:\Users\juanm\Documents\GitHub\biblioteca_negocio_prolongo\google\token_personal.json"
)


def _creds(token_path: Path = DEFAULT_TOKEN):
    creds = Credentials.from_authorized_user_file(str(token_path))
    if creds.expired and creds.refresh_token:
        creds.refresh(Request())
        token_path.write_text(creds.to_json(), encoding="utf-8")
    return creds


def drive(token_path: Path = DEFAULT_TOKEN):
    return build("drive", "v3", credentials=_creds(token_path), static_discovery=False)


def docs(token_path: Path = DEFAULT_TOKEN):
    return build("docs", "v1", credentials=_creds(token_path), static_discovery=False)


def find_folder(name: str, parent: str | None = None, svc=None) -> str:
    """Devuelve el id de la primera carpeta con ese nombre (opcionalmente bajo `parent`)."""
    svc = svc or drive()
    q = f"name='{name}' and mimeType='application/vnd.google-apps.folder' and trashed=false"
    if parent:
        q += f" and '{parent}' in parents"
    r = svc.files().list(q=q, fields="files(id,name)").execute()
    files = r.get("files", [])
    if not files:
        raise SystemExit(f"[ERROR] Carpeta no encontrada: {name}")
    return files[0]["id"]


def list_folder(fid: str, svc=None):
    svc = svc or drive()
    r = svc.files().list(
        q=f"'{fid}' in parents and trashed=false",
        fields="files(id,name,mimeType,size,modifiedTime)", orderBy="folder,name",
    ).execute()
    for f in r.get("files", []):
        kind = "DIR " if f["mimeType"] == "application/vnd.google-apps.folder" else "file"
        print(f"  {kind} {f['name']}  id={f['id']}")


def read_doc(fid: str, svc=None) -> str:
    svc = svc or drive()
    return svc.files().export(fileId=fid, mimeType="text/plain").execute().decode("utf-8")


def download(fid: str, out: Path, svc=None):
    svc = svc or drive()
    req = svc.files().get_media(fileId=fid)
    with io.FileIO(out, "wb") as fh:
        dl = MediaIoBaseDownload(fh, req)
        done = False
        while not done:
            _, done = dl.next_chunk()


def create_doc_from_html(name: str, parent: str, html: str, svc=None) -> dict:
    svc = svc or drive()
    media = MediaIoBaseUpload(io.BytesIO(html.encode("utf-8")), mimetype="text/html", resumable=False)
    return svc.files().create(
        body={"name": name, "parents": [parent], "mimeType": "application/vnd.google-apps.document"},
        media_body=media, fields="id,name,webViewLink",
    ).execute()


def move(fid: str, new_parent: str, old_parent: str | None = None, svc=None):
    svc = svc or drive()
    return svc.files().update(
        fileId=fid, addParents=new_parent, removeParents=old_parent, fields="id,parents"
    ).execute()


def trash(fid: str, svc=None):
    svc = svc or drive()
    return svc.files().update(fileId=fid, body={"trashed": True}).execute()


def strip_solutions(doc_id: str, marker: str = "SOLUCIONES", svc=None):
    """Borra todo el contenido de un Google Doc desde la cabecera `marker` hasta el final."""
    svc = svc or docs()
    d = svc.documents().get(documentId=doc_id).execute()
    start = None
    for el in d["body"]["content"]:
        para = el.get("paragraph")
        if not para:
            continue
        text = "".join(r.get("textRun", {}).get("content", "") for r in para.get("elements", []))
        if marker in text:
            start = el["startIndex"]
            break
    if start is None:
        print("[aviso] No se encontró el marcador:", marker)
        return
    end = d["body"]["content"][-1]["endIndex"] - 1
    if end > start:
        svc.documents().batchUpdate(documentId=doc_id, body={"requests": [
            {"deleteContentRange": {"range": {"startIndex": start, "endIndex": end}}}]}).execute()
        print("[ok] Contenido desde '%s' eliminado" % marker)


def _main(argv):
    if not argv:
        print(__doc__)
        return
    cmd = argv[0]
    if cmd == "list":
        list_folder(argv[1] if len(argv) > 1 else find_folder("TAREAS"))
    elif cmd == "find":
        print(find_folder(argv[1], argv[2] if len(argv) > 2 else None))
    elif cmd == "read":
        print(read_doc(argv[1]))
    elif cmd == "pdf":
        download(argv[1], Path(argv[2]))
        print("descargado", argv[2])
    elif cmd == "make":
        html = Path(argv[3]).read_text(encoding="utf-8")
        print(create_doc_from_html(argv[1], argv[2], html)["webViewLink"])
    elif cmd == "move":
        move(argv[1], argv[2], argv[3] if len(argv) > 3 else None)
        print("movido")
    elif cmd == "strip":
        strip_solutions(argv[1])
    else:
        print(__doc__)


if __name__ == "__main__":
    _main(sys.argv[1:])