mm_api_web.py

"""
Cliente para a API interna da aplicação web do MindMeister
(https://www.mindmeister.com/app/...), usada pelo próprio editor de mapas no
navegador.

Essa é a ÚNICA credencial de que este projeto precisa: o **cookie de sessão
do navegador** (`_mind_session`). Não é uma API pública/documentada, e o
cookie expira periodicamente (é preciso capturar um novo quando parar de
funcionar) — mas cobre tudo que a exportação precisa:

- `mm.maps`/`mm.folders` via GraphQL (`/app/api/graphql`) — lista todos os
  mapas (cada um já com seu `folderId`) e todas as pastas da conta. É a
  única fonte confirmada pra isso: a API v2 (OAuth2/Personal Access Token)
  não tem endpoint de listagem (testado exaustivamente — 404 em qualquer
  variação de path), e a API v1 (REST clássica, deprecated) tinha um fluxo
  de autenticação totalmente separado (api_key + secret + auth_token) só
  pra conseguir a mesma coisa que o cookie já dá de graça.
- `content.json` (`/app/api/maps/content.json`) — árvore completa de um
  mapa (id, hierarquia via parentId/rank, título, nota, imagem, vídeo e os
  links "anexados" a cada nó). Nem a API v1 nem o `.docx` exportado pela v2
  preservam os links anexados/vídeos, nem evitam distorção de proporção
  nas imagens — só essa fonte tem isso (confirmado testando exaustivamente
  contra a conta real).

Como obter o cookie (`MINDMEISTER_SESSION_COOKIE` no .env):
1. Abra https://www.mindmeister.com/app/ logado no navegador.
2. Abra o DevTools (F12) > aba Network > filtre por Fetch/XHR > abra
   qualquer mapa.
3. Ache uma requisição para `content.json` > aba Headers > Request Headers
   > copie só o valor do cookie `_mind_session` (não precisa dos outros
   cookies de analytics).
"""
from __future__ import annotations

import logging
import time
from typing import Any

import requests

from organizer import MMFolder, MMMap

log = logging.getLogger("mm_api_web")

CONTENT_URL = "https://www.mindmeister.com/app/api/maps/content.json"
GRAPHQL_URL = "https://www.mindmeister.com/app/api/graphql"


class MindMeisterWebError(RuntimeError):
    def __init__(self, status_code: int, url: str, body: str):
        super().__init__(f"HTTP {status_code} em {url}: {body[:300]}")
        self.status_code = status_code
        self.url = url
        self.body = body


class MindMeisterWebClient:
    def __init__(self, session_cookie: str, timeout: int = 60, max_retries: int = 3):
        self.timeout = timeout
        self.max_retries = max_retries
        self.session = requests.Session()
        cookie_value = session_cookie.split("=", 1)[1] if session_cookie.startswith("_mind_session=") else session_cookie
        self.session.cookies.set("_mind_session", cookie_value, domain="www.mindmeister.com")

    def _request(self, method: str, url: str, **kwargs) -> requests.Response:
        last_exc = None
        for attempt in range(1, self.max_retries + 1):
            try:
                resp = self.session.request(method, url, timeout=self.timeout, **kwargs)
            except requests.RequestException as exc:
                last_exc = exc
                log.warning("Falha de rede (%s/%s) em %s: %s", attempt, self.max_retries, url, exc)
                time.sleep(2 * attempt)
                continue

            if resp.status_code >= 500:
                log.warning("Erro %s do servidor (%s/%s) em %s", resp.status_code, attempt, self.max_retries, url)
                time.sleep(2 * attempt)
                continue

            if resp.status_code >= 400:
                raise MindMeisterWebError(resp.status_code, url, resp.text)

            return resp

        raise MindMeisterWebError(-1, url, str(last_exc) if last_exc else "esgotou tentativas")

    def _get(self, url: str, **kwargs) -> requests.Response:
        return self._request("GET", url, **kwargs)

    def _graphql(self, operation_name: str, query: str) -> dict[str, Any]:
        resp = self._request(
            "POST",
            GRAPHQL_URL,
            json={"operationName": operation_name, "query": query, "variables": {}},
            headers={"Content-Type": "application/json", "Accept": "application/json"},
        )
        data = resp.json()
        if "errors" in data:
            raise MindMeisterWebError(resp.status_code, resp.url, str(data["errors"])[:300])
        return data["data"]

    def get_map_content(self, map_id: str, *, _migration_wait_retries: int = 5) -> dict[str, Any]:
        """GET content.json?idea_id={map_id} — árvore completa de ideas do
        mapa (título, hierarquia, notas, imagens, vídeos e links anexados).

        Mapas antigos que ainda não foram migrados para o editor "panda"
        (o motor atual do MindMeister) disparam essa migração sob demanda
        no primeiro acesso. Enquanto ela não termina, a resposta vem sem
        `ideas`, com um campo `pandaChangesMigration` — nesse caso esperamos
        e tentamos de novo. Quando termina, o id antigo passa a responder
        com um redirect (contendo `newIdeaId`) para o novo id — nesse caso
        refazemos a chamada com o id novo.
        """
        params = {
            "idea_id": map_id,
            "isPublicView": "false",
            "isEmbeddedMapEditor": "false",
            "isPrintView": "false",
        }
        resp = self._get(CONTENT_URL, params=params)
        data = resp.json()

        if "ideas" in data:
            return data

        new_idea_id = data.get("newIdeaId")
        if new_idea_id:
            log.info("Mapa %s migrado para o editor panda; usando novo id %s.", map_id, new_idea_id)
            return self.get_map_content(str(new_idea_id), _migration_wait_retries=_migration_wait_retries)

        if "pandaChangesMigration" in data and _migration_wait_retries > 0:
            log.info("Mapa %s ainda migrando para o editor panda; aguardando e tentando de novo.", map_id)
            time.sleep(5)
            return self.get_map_content(map_id, _migration_wait_retries=_migration_wait_retries - 1)

        # Resposta 200 mas sem o formato esperado (nem ideas, nem
        # newIdeaId, nem migração em andamento) — acontece quando a sessão
        # expira/degrada no meio de uma execução longa. Trata como falha
        # (em vez de deixar o KeyError subir e travar o processamento dos
        # mapas seguintes).
        raise MindMeisterWebError(resp.status_code, resp.url, str(data)[:300])

    def download_file(self, url: str) -> tuple[bytes, str]:
        """Baixa um arquivo autenticado com o mesmo cookie de sessão (usado
        para as imagens dos nós — `imageURL`). Retorna (bytes, content_type).
        """
        resp = self._get(url)
        return resp.content, resp.headers.get("Content-Type", "")

    def list_folders(self) -> list[MMFolder]:
        """GraphQL `query Folders` — todas as pastas da conta."""
        data = self._graphql("Folders", "query Folders { folders { id name parentId } }")
        return [
            MMFolder(id=str(f["id"]), name=f["name"], parent_id=str(f["parentId"]) if f.get("parentId") else None)
            for f in data["folders"]
        ]

    def list_maps(self) -> list[MMMap]:
        """GraphQL `query Maps` — todos os mapas da conta, cada um já com
        seu `folderId` (sem precisar filtrar por pasta, um problema real
        que a listagem da API v1 tinha).
        """
        data = self._graphql("Maps", "query Maps { maps { id title folderId } }")
        return [
            MMMap(id=str(m["id"]), title=m["title"], folder_id=str(m["folderId"]) if m.get("folderId") else None)
            for m in data["maps"]
        ]

    def list_all(self) -> tuple[list[MMFolder], list[MMMap]]:
        return self.list_folders(), self.list_maps()