"""Google Workspace operations behind the Task Designer's Google nodes —
Sheets, Docs, Slides, Drive, Calendar, Gmail and Forms, all through one
"google" Connector (sign-in lives in app/crud/google_auth.py).

A node names an operation and carries string settings; run() dispatches to the
function registered for (app, operation). Each returns (detail, result): the
words for the run's trace and the value its output params read from.

Writing to a sheet defaults to "RAW" so a value that came from a form can't
become a live formula (=IMPORTXML(...) would send data anywhere). The author
can opt in to "as typed" per node.

Ids are checked before they go in a URL, so a value from a form can't reach a
different API path.
"""

import asyncio
import base64
import json
import re
import uuid
from dataclasses import dataclass, field
from email.message import EmailMessage
from typing import Any, Awaitable, Callable
from urllib.parse import quote

import httpx

from app.crud.google_auth import GoogleError, account_label, error_reason, request

SHEETS = "https://sheets.googleapis.com/v4/spreadsheets"
DOCS = "https://docs.googleapis.com/v1/documents"
SLIDES = "https://slides.googleapis.com/v1/presentations"
DRIVE_API = "https://www.googleapis.com/drive/v3/files"
DRIVE_ABOUT = "https://www.googleapis.com/drive/v3/about"
DRIVE_UPLOAD = "https://www.googleapis.com/upload/drive/v3/files"
CALENDAR = "https://www.googleapis.com/calendar/v3/calendars"
GMAIL = "https://gmail.googleapis.com/gmail/v1/users/me"
FORMS = "https://forms.googleapis.com/v1/forms"

_AUTH = "https://www.googleapis.com/auth/"
S_SHEETS = (_AUTH + "spreadsheets",)
S_DOCS = (_AUTH + "documents",)
S_SLIDES = (_AUTH + "presentations",)
S_DRIVE = (_AUTH + "drive",)
S_DRIVE_META = (_AUTH + "drive.metadata.readonly",)
S_CALENDAR = (_AUTH + "calendar",)
S_GMAIL_SEND = (_AUTH + "gmail.send",)
S_GMAIL_READ = (_AUTH + "gmail.readonly",)
S_FORMS = (_AUTH + "forms.body.readonly", _AUTH + "forms.responses.readonly")

NATIVE_DOC = "application/vnd.google-apps.document"
NATIVE_SHEET = "application/vnd.google-apps.spreadsheet"
NATIVE_SLIDES = "application/vnd.google-apps.presentation"

_EXPORTS = {
    "pdf": ("application/pdf", "pdf"),
    "docx": ("application/vnd.openxmlformats-officedocument.wordprocessingml.document", "docx"),
    "xlsx": ("application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", "xlsx"),
    "pptx": ("application/vnd.openxmlformats-officedocument.presentationml.presentation", "pptx"),
    "txt": ("text/plain", "txt"),
    "csv": ("text/csv", "csv"),
}
_NATIVE_DEFAULT_EXPORT = {NATIVE_DOC: "docx", NATIVE_SHEET: "xlsx", NATIVE_SLIDES: "pptx"}

MAX_FILE_BYTES = 50 * 1024 * 1024
MAX_EMAIL_BYTES = 25 * 1024 * 1024
MAX_ROWS = 50_000
MAX_LIST = 1000

_ID = re.compile(r"^[A-Za-z0-9_@.\-]{1,256}$")
_ID_IN_URL = re.compile(r"/(?:d|e|folders)/(?:e/)?([A-Za-z0-9_\-]{10,})")
_ID_PARAM = re.compile(r"[?&]id=([A-Za-z0-9_\-]{10,})")

LoadFile = Callable[[str], Awaitable[tuple[str, str, bytes] | None]]
SaveFile = Callable[[str, str, bytes], Awaitable[str]]


@dataclass
class Call:
    config: dict
    settings: dict
    inputs: dict = field(default_factory=dict)
    load_file: LoadFile | None = None  # file id -> (name, content type, bytes)
    save_file: SaveFile | None = None  # (name, content type, bytes) -> file id

    def text(self, key: str, default: str = "") -> str:
        value = self.settings.get(key)
        return default if value is None or value == "" else str(value).strip()

    def need(self, key: str, label: str) -> str:
        value = self.text(key)
        if not value:
            raise GoogleError(f"set the {label}")
        return value

    def yes(self, key: str, default: bool = False) -> bool:
        value = self.settings.get(key)
        if value is None or value == "":
            return default
        if isinstance(value, bool):
            return value
        return str(value).strip().lower() in ("yes", "true", "1", "on")

    def number(self, key: str, default: int, low: int = 1, high: int = MAX_LIST) -> int:
        try:
            return max(low, min(high, int(float(self.settings.get(key) or default))))
        except (TypeError, ValueError) as exc:
            raise GoogleError(f"{key.replace('_', ' ')} must be a number") from exc

    def file_ref(self, key: str, label: str) -> str:
        """A file id, or the file's Google URL pasted whole."""
        return clean_id(self.need(key, label), label)

    async def api(self, scopes: tuple[str, ...], method: str, url: str, **kwargs: Any) -> httpx.Response:
        resp = await request(self.config, scopes, method, url, **kwargs)
        if resp.status_code >= 400:
            raise GoogleError(_explain(resp, self.config))
        return resp

    async def json(self, scopes: tuple[str, ...], method: str, url: str, **kwargs: Any) -> dict:
        resp = await self.api(scopes, method, url, **kwargs)
        if not resp.content:
            return {}
        try:
            body = resp.json()
        except ValueError as exc:
            raise GoogleError("Google sent back something that wasn't JSON") from exc
        return body if isinstance(body, dict) else {}


def _explain(resp: httpx.Response, config: dict) -> str:
    code, message = error_reason(resp)
    text = f"Google said {resp.status_code} {code}: {message}"
    if resp.status_code == 404:
        text += f" — check the id, and that the file is shared with {account_label(config)}"
    elif resp.status_code == 403 and "has not been used" in message:
        text += " — enable this API in the Google Cloud project"
    elif resp.status_code == 403 and "quota" in message.lower() and "storage" in message.lower():
        text += " — service accounts have no storage of their own; choose a Shared Drive folder, or connect with an OAuth user"
    elif resp.status_code == 403:
        text += f" — check that {account_label(config)} has access"
    return text


def clean_id(value: str, label: str = "id") -> str:
    value = str(value or "").strip()
    match = _ID_IN_URL.search(value) or _ID_PARAM.search(value)
    if match:
        return match.group(1)
    if not _ID.match(value):
        raise GoogleError(f"the {label} doesn't look right: paste the id or the file's URL")
    return value


def _q(value: str) -> str:
    return quote(str(value), safe="")


def _cell(value: Any) -> Any:
    if value is None:
        return ""
    if isinstance(value, (str, int, float, bool)):
        return value
    return json.dumps(value, default=str)


def _plain(value: Any) -> str:
    if value is None:
        return ""
    return value if isinstance(value, str) else json.dumps(value, default=str)


def parse_json(value: Any, label: str) -> Any:
    if not isinstance(value, str):
        return value
    text = value.strip()
    if not text:
        return None
    try:
        return json.loads(text)
    except ValueError as exc:
        raise GoogleError(f"{label} isn't valid JSON") from exc


def _rows(value: Any) -> list:
    """Rows from a list of lists, a list of objects, one object or one list —
    as a value from an earlier node, or as JSON text."""
    data = parse_json(value, "the rows")
    if data is None or data == []:
        raise GoogleError("no rows to write — pass a list from an earlier node, or JSON such as [[\"a\", 1]]")
    if isinstance(data, dict):
        return [data]
    if not isinstance(data, list):
        return [[data]]
    if all(isinstance(r, (dict, list)) for r in data):
        return data
    return [data]  # one flat list is one row


def _tab(a1_range: str) -> str:
    tab = a1_range.split("!")[0].strip()
    if tab.startswith("'"):
        return tab
    return "'" + tab.replace("'", "''") + "'"


def _input_option(call: Call) -> str:
    return "USER_ENTERED" if call.text("value_input") == "as_typed" else "RAW"


def _doc_url(kind: str, file_id: str) -> str:
    return f"https://docs.google.com/{kind}/d/{file_id}/edit"


def _replacements(call: Call) -> dict[str, str]:
    """{{name}} -> value: every declared input, plus an optional JSON object."""
    pairs = {f"{{{{{k}}}}}": _plain(v) for k, v in call.inputs.items()}
    extra = parse_json(call.settings.get("replacements"), "the replacements")
    if extra:
        if not isinstance(extra, dict):
            raise GoogleError('the replacements must be a JSON object, like {"{{name}}": "Asha"}')
        pairs.update({str(k): _plain(v) for k, v in extra.items()})
    return pairs


def _replace_requests(pairs: dict[str, str]) -> list[dict]:
    return [{"replaceAllText": {"containsText": {"text": k, "matchCase": True}, "replaceText": v}} for k, v in pairs.items()]


def _occurrences(body: dict) -> int:
    return sum((r.get("replaceAllText") or {}).get("occurrencesChanged") or 0 for r in body.get("replies") or [])


async def _create_in_folder(call: Call, name: str, mime: str, folder: str) -> str:
    """Creating through Drive is what can place a file in a chosen folder —
    and a service account needs a Shared Drive folder to store anything."""
    body = await call.json(S_DRIVE, "POST", DRIVE_API, params={"supportsAllDrives": "true", "fields": "id"}, json={"name": name, "mimeType": mime, "parents": [clean_id(folder, "folder")]})
    return body["id"]


async def _copy(call: Call, file_id: str, name: str, folder: str) -> str:
    payload: dict[str, Any] = {}
    if name:
        payload["name"] = name
    if folder:
        payload["parents"] = [clean_id(folder, "folder")]
    body = await call.json(S_DRIVE, "POST", f"{DRIVE_API}/{_q(file_id)}/copy", params={"supportsAllDrives": "true", "fields": "id,name"}, json=payload)
    return body["id"]


async def _export(call: Call, file_id: str, fmt: str) -> tuple[str, dict]:
    if fmt not in _EXPORTS:
        raise GoogleError(f"can't export as {fmt}")
    if call.save_file is None:
        raise GoogleError("exporting isn't available here")
    meta = await call.json(S_DRIVE, "GET", f"{DRIVE_API}/{_q(file_id)}", params={"supportsAllDrives": "true", "fields": "name"})
    mime, extension = _EXPORTS[fmt]
    resp = await call.api(S_DRIVE, "GET", f"{DRIVE_API}/{_q(file_id)}/export", params={"mimeType": mime}, timeout=120.0)
    name = f"{meta.get('name') or file_id}.{extension}"
    saved = await call.save_file(name, mime, resp.content)
    return f"exported {name}", {"file_id": saved, "file_name": name, "size": len(resp.content)}


# ---- Sheets ----------------------------------------------------------------------------------


async def _sheet_values(call: Call, sheet_id: str, a1_range: str) -> list[list]:
    body = await call.json(S_SHEETS, "GET", f"{SHEETS}/{_q(sheet_id)}/values/{_q(a1_range)}", params={"valueRenderOption": call.text("value_render", "FORMATTED_VALUE")})
    return body.get("values") or []


def _headers_of(first: list) -> list[str]:
    return [str(h) if str(h).strip() else f"col_{i}" for i, h in enumerate(first)]


def _as_records(values: list[list]) -> tuple[list[str], list[dict]]:
    if not values:
        return [], []
    headers = _headers_of(values[0])
    return headers, [{h: (row[i] if i < len(row) else "") for i, h in enumerate(headers)} for row in values[1:]]


async def _sheets_read(call: Call) -> tuple[str, Any]:
    sheet_id = call.file_ref("spreadsheet_id", "spreadsheet")
    values = await _sheet_values(call, sheet_id, call.need("range", "range"))
    limit = call.number("limit", 10_000, high=MAX_ROWS)
    if call.yes("first_row_headers", True):
        headers, rows = _as_records(values)
    else:
        headers, rows = [], values
    truncated = len(rows) > limit
    rows = rows[:limit]
    return f"{len(rows)} row(s)", {"rows": rows, "headers": headers, "row_count": len(rows), "truncated": "yes" if truncated else "no", "first_row": rows[0] if rows else None}


async def _sheets_find(call: Call) -> tuple[str, Any]:
    sheet_id = call.file_ref("spreadsheet_id", "spreadsheet")
    a1_range = call.need("range", "range")
    column = call.need("column", "column to search")
    wanted = call.text("value")
    headers, rows = _as_records(await _sheet_values(call, sheet_id, a1_range))
    if column not in headers:
        raise GoogleError(f"no column \"{column}\" — this sheet's first row has: {', '.join(headers[:30]) or 'nothing'}")
    found, numbers = [], []
    for i, row in enumerate(rows):
        if str(row.get(column, "")).strip() == wanted:
            found.append(row)
            numbers.append(i + 2)  # the header is row 1
    return f"{len(found)} matching row(s)", {"rows": found, "row_numbers": numbers, "row_count": len(found), "first_row": found[0] if found else None, "first_row_number": numbers[0] if numbers else None}


async def _to_grid(call: Call, sheet_id: str, a1_range: str, rows: list) -> tuple[list[list], list[str]]:
    """Objects are laid out by the sheet's header row (an empty sheet gets one
    from their keys). Keys with no column are reported, not silently lost."""
    if not any(isinstance(r, dict) for r in rows):
        return [[_cell(c) for c in r] for r in rows], []
    if not all(isinstance(r, dict) for r in rows):
        raise GoogleError("the rows mix lists and objects — use one or the other")
    tab = _tab(a1_range)
    existing = await _sheet_values(call, sheet_id, f"{tab}!1:1")
    grid: list[list] = []
    if existing and existing[0]:
        headers = [str(h) for h in existing[0]]
    else:
        headers = list(dict.fromkeys(k for r in rows for k in r))
        grid.append(list(headers))
    known = set(headers)
    skipped = sorted({k for r in rows for k in r if k not in known})
    grid.extend([_cell(r.get(h)) for h in headers] for r in rows)
    return grid, skipped


async def _sheets_append(call: Call) -> tuple[str, Any]:
    sheet_id = call.file_ref("spreadsheet_id", "spreadsheet")
    a1_range = call.need("range", "sheet or range")
    grid, skipped = await _to_grid(call, sheet_id, a1_range, _rows(call.settings.get("rows")))
    body = await call.json(S_SHEETS, "POST", f"{SHEETS}/{_q(sheet_id)}/values/{_q(a1_range)}:append", params={"valueInputOption": _input_option(call), "insertDataOption": "INSERT_ROWS"}, json={"values": grid})
    updates = body.get("updates") or {}
    detail = f"added {updates.get('updatedRows', len(grid))} row(s)" + (f" — no column for: {', '.join(skipped)}" if skipped else "")
    return detail, {"updated_rows": updates.get("updatedRows", len(grid)), "updated_range": updates.get("updatedRange", ""), "skipped_columns": skipped}


async def _sheets_update(call: Call) -> tuple[str, Any]:
    sheet_id = call.file_ref("spreadsheet_id", "spreadsheet")
    a1_range = call.need("range", "range")
    grid, skipped = await _to_grid(call, sheet_id, a1_range, _rows(call.settings.get("rows")))
    body = await call.json(S_SHEETS, "PUT", f"{SHEETS}/{_q(sheet_id)}/values/{_q(a1_range)}", params={"valueInputOption": _input_option(call)}, json={"values": grid})
    return f"updated {body.get('updatedCells', 0)} cell(s)", {"updated_cells": body.get("updatedCells", 0), "updated_rows": body.get("updatedRows", 0), "updated_range": body.get("updatedRange", ""), "skipped_columns": skipped}


async def _sheets_clear(call: Call) -> tuple[str, Any]:
    sheet_id = call.file_ref("spreadsheet_id", "spreadsheet")
    body = await call.json(S_SHEETS, "POST", f"{SHEETS}/{_q(sheet_id)}/values/{_q(call.need('range', 'range'))}:clear", json={})
    return f"cleared {body.get('clearedRange', '')}", {"cleared_range": body.get("clearedRange", "")}


async def _sheets_create(call: Call) -> tuple[str, Any]:
    title = call.need("title", "title")
    folder = call.text("folder_id")
    tabs = [t.strip() for t in call.text("sheet_names").split(",") if t.strip()]
    if folder:
        sheet_id = await _create_in_folder(call, title, NATIVE_SHEET, folder)
        url = _doc_url("spreadsheets", sheet_id)
        if tabs:
            await call.json(S_SHEETS, "POST", f"{SHEETS}/{_q(sheet_id)}:batchUpdate", json={"requests": [{"addSheet": {"properties": {"title": t}}} for t in tabs]})
    else:
        payload: dict[str, Any] = {"properties": {"title": title}}
        if tabs:
            payload["sheets"] = [{"properties": {"title": t}} for t in tabs]
        body = await call.json(S_SHEETS, "POST", SHEETS, json=payload)
        sheet_id, url = body["spreadsheetId"], body.get("spreadsheetUrl") or _doc_url("spreadsheets", body["spreadsheetId"])
    return f"created spreadsheet {title}", {"spreadsheet_id": sheet_id, "url": url}


async def _sheet_tabs(call: Call, sheet_id: str) -> list[dict]:
    body = await call.json(S_SHEETS, "GET", f"{SHEETS}/{_q(sheet_id)}", params={"fields": "sheets.properties"})
    return [s.get("properties") or {} for s in body.get("sheets") or []]


async def _sheets_list_tabs(call: Call) -> tuple[str, Any]:
    tabs = await _sheet_tabs(call, call.file_ref("spreadsheet_id", "spreadsheet"))
    return f"{len(tabs)} sheet(s)", {"sheets": [{"title": t.get("title"), "sheet_id": t.get("sheetId"), "index": t.get("index")} for t in tabs], "names": [t.get("title") for t in tabs], "count": len(tabs)}


async def _sheets_add_tab(call: Call) -> tuple[str, Any]:
    sheet_id = call.file_ref("spreadsheet_id", "spreadsheet")
    title = call.need("title", "sheet name")
    body = await call.json(S_SHEETS, "POST", f"{SHEETS}/{_q(sheet_id)}:batchUpdate", json={"requests": [{"addSheet": {"properties": {"title": title}}}]})
    props = ((body.get("replies") or [{}])[0].get("addSheet") or {}).get("properties") or {}
    return f"added sheet {title}", {"sheet_id": props.get("sheetId"), "title": title}


async def _sheets_delete_tab(call: Call) -> tuple[str, Any]:
    sheet_id = call.file_ref("spreadsheet_id", "spreadsheet")
    title = call.need("title", "sheet name")
    match = next((t for t in await _sheet_tabs(call, sheet_id) if t.get("title") == title), None)
    if match is None:
        raise GoogleError(f"no sheet named \"{title}\"")
    await call.json(S_SHEETS, "POST", f"{SHEETS}/{_q(sheet_id)}:batchUpdate", json={"requests": [{"deleteSheet": {"sheetId": match.get("sheetId")}}]})
    return f"deleted sheet {title}", {"title": title}


async def _sheets_export(call: Call) -> tuple[str, Any]:
    return await _export(call, call.file_ref("spreadsheet_id", "spreadsheet"), call.text("format", "xlsx"))


# ---- Docs ------------------------------------------------------------------------------------


def _doc_text(content: list) -> str:
    parts: list[str] = []
    for element in content or []:
        paragraph = element.get("paragraph")
        if paragraph:
            parts.extend((e.get("textRun") or {}).get("content", "") for e in paragraph.get("elements") or [])
        table = element.get("table")
        if table:
            for row in table.get("tableRows") or []:
                for cell in row.get("tableCells") or []:
                    parts.append(_doc_text(cell.get("content") or []))
    return "".join(parts)


async def _docs_batch(call: Call, doc_id: str, requests_: list[dict]) -> dict:
    return await call.json(S_DOCS, "POST", f"{DOCS}/{_q(doc_id)}:batchUpdate", json={"requests": requests_})


async def _docs_create(call: Call) -> tuple[str, Any]:
    title = call.need("title", "title")
    folder = call.text("folder_id")
    if folder:
        doc_id = await _create_in_folder(call, title, NATIVE_DOC, folder)
    else:
        doc_id = (await call.json(S_DOCS, "POST", DOCS, json={"title": title}))["documentId"]
    text = _plain(call.settings.get("text"))
    if text:
        await _docs_batch(call, doc_id, [{"insertText": {"location": {"index": 1}, "text": text}}])
    return f"created document {title}", {"document_id": doc_id, "url": _doc_url("document", doc_id)}


async def _docs_read(call: Call) -> tuple[str, Any]:
    doc_id = call.file_ref("document_id", "document")
    body = await call.json(S_DOCS, "GET", f"{DOCS}/{_q(doc_id)}")
    text = _doc_text((body.get("body") or {}).get("content") or [])
    return f"read {len(text)} character(s)", {"document_id": doc_id, "title": body.get("title", ""), "text": text, "url": _doc_url("document", doc_id)}


async def _docs_append(call: Call) -> tuple[str, Any]:
    doc_id = call.file_ref("document_id", "document")
    text = _plain(call.settings.get("text"))
    if not text:
        raise GoogleError("set the text to add")
    await _docs_batch(call, doc_id, [{"insertText": {"endOfSegmentLocation": {}, "text": text}}])
    return f"added {len(text)} character(s)", {"document_id": doc_id, "url": _doc_url("document", doc_id)}


async def _docs_fill(call: Call) -> tuple[str, Any]:
    doc_id = call.file_ref("document_id", "document")
    pairs = _replacements(call)
    if not pairs:
        raise GoogleError("nothing to fill in — declare inputs (each replaces {{its name}}) or set the replacements")
    body = await _docs_batch(call, doc_id, _replace_requests(pairs))
    count = _occurrences(body)
    return f"replaced {count} placeholder(s)", {"document_id": doc_id, "replaced": count, "url": _doc_url("document", doc_id)}


async def _docs_from_template(call: Call) -> tuple[str, Any]:
    template = call.file_ref("template_id", "template document")
    doc_id = await _copy(call, template, call.text("title"), call.text("folder_id"))
    pairs = _replacements(call)
    count = _occurrences(await _docs_batch(call, doc_id, _replace_requests(pairs))) if pairs else 0
    return f"created a document from the template, replaced {count} placeholder(s)", {"document_id": doc_id, "replaced": count, "url": _doc_url("document", doc_id)}


async def _docs_export(call: Call) -> tuple[str, Any]:
    return await _export(call, call.file_ref("document_id", "document"), call.text("format", "pdf"))


# ---- Slides ----------------------------------------------------------------------------------


def _shape_text(element: dict) -> str:
    text = ((element.get("shape") or {}).get("text") or {}).get("textElements") or []
    return "".join((t.get("textRun") or {}).get("content", "") for t in text)


async def _slides_batch(call: Call, deck_id: str, requests_: list[dict]) -> dict:
    return await call.json(S_SLIDES, "POST", f"{SLIDES}/{_q(deck_id)}:batchUpdate", json={"requests": requests_})


async def _slides_create(call: Call) -> tuple[str, Any]:
    title = call.need("title", "title")
    folder = call.text("folder_id")
    if folder:
        deck_id = await _create_in_folder(call, title, NATIVE_SLIDES, folder)
    else:
        deck_id = (await call.json(S_SLIDES, "POST", SLIDES, json={"title": title}))["presentationId"]
    return f"created presentation {title}", {"presentation_id": deck_id, "url": _doc_url("presentation", deck_id)}


async def _slides_read(call: Call) -> tuple[str, Any]:
    deck_id = call.file_ref("presentation_id", "presentation")
    body = await call.json(S_SLIDES, "GET", f"{SLIDES}/{_q(deck_id)}")
    slides = [
        {"slide_id": s.get("objectId"), "number": i + 1, "text": "\n".join(t.strip() for t in (_shape_text(e) for e in s.get("pageElements") or []) if t.strip())}
        for i, s in enumerate(body.get("slides") or [])
    ]
    return f"{len(slides)} slide(s)", {"presentation_id": deck_id, "title": body.get("title", ""), "slides": slides, "slide_count": len(slides), "url": _doc_url("presentation", deck_id)}


_SLIDE_LAYOUTS = {"TITLE_AND_BODY": ("TITLE", "BODY"), "TITLE_ONLY": ("TITLE",), "BLANK": ()}


async def _slides_add(call: Call) -> tuple[str, Any]:
    deck_id = call.file_ref("presentation_id", "presentation")
    layout = call.text("layout", "TITLE_AND_BODY")
    if layout not in _SLIDE_LAYOUTS:
        raise GoogleError(f"unknown layout {layout}")
    suffix = uuid.uuid4().hex[:12]
    slide_id = f"ff_slide_{suffix}"
    texts = {"TITLE": _plain(call.settings.get("title")), "BODY": _plain(call.settings.get("text"))}
    mappings, inserts = [], []
    for kind in _SLIDE_LAYOUTS[layout]:
        object_id = f"ff_{kind.lower()}_{suffix}"
        mappings.append({"layoutPlaceholder": {"type": kind, "index": 0}, "objectId": object_id})
        if texts[kind]:
            inserts.append({"insertText": {"objectId": object_id, "text": texts[kind]}})
    create: dict[str, Any] = {"objectId": slide_id, "slideLayoutReference": {"predefinedLayout": layout}, "placeholderIdMappings": mappings}
    if call.text("position"):
        create["insertionIndex"] = call.number("position", 1, low=0, high=10_000)
    await _slides_batch(call, deck_id, [{"createSlide": create}, *inserts])
    return "added a slide", {"presentation_id": deck_id, "slide_id": slide_id, "url": _doc_url("presentation", deck_id)}


async def _slides_delete(call: Call) -> tuple[str, Any]:
    deck_id = call.file_ref("presentation_id", "presentation")
    slide = call.need("slide_id", "slide id")
    await _slides_batch(call, deck_id, [{"deleteObject": {"objectId": slide}}])
    return f"deleted slide {slide}", {"slide_id": slide}


async def _slides_fill(call: Call) -> tuple[str, Any]:
    deck_id = call.file_ref("presentation_id", "presentation")
    pairs = _replacements(call)
    if not pairs:
        raise GoogleError("nothing to fill in — declare inputs (each replaces {{its name}}) or set the replacements")
    count = _occurrences(await _slides_batch(call, deck_id, _replace_requests(pairs)))
    return f"replaced {count} placeholder(s)", {"presentation_id": deck_id, "replaced": count, "url": _doc_url("presentation", deck_id)}


async def _slides_from_template(call: Call) -> tuple[str, Any]:
    template = call.file_ref("template_id", "template presentation")
    deck_id = await _copy(call, template, call.text("title"), call.text("folder_id"))
    pairs = _replacements(call)
    count = _occurrences(await _slides_batch(call, deck_id, _replace_requests(pairs))) if pairs else 0
    return f"created a presentation from the template, replaced {count} placeholder(s)", {"presentation_id": deck_id, "replaced": count, "url": _doc_url("presentation", deck_id)}


async def _slides_export(call: Call) -> tuple[str, Any]:
    return await _export(call, call.file_ref("presentation_id", "presentation"), call.text("format", "pptx"))


# ---- Drive -----------------------------------------------------------------------------------

_DRIVE_FIELDS = "id,name,mimeType,size,modifiedTime,webViewLink,parents"


def _drive_row(f: dict) -> dict:
    return {"id": f.get("id"), "name": f.get("name"), "mime_type": f.get("mimeType"), "size": int(f["size"]) if f.get("size") else None, "modified": f.get("modifiedTime"), "url": f.get("webViewLink"), "is_folder": f.get("mimeType") == "application/vnd.google-apps.folder"}


def _drive_literal(text: str) -> str:
    return text.replace("\\", "\\\\").replace("'", "\\'")


async def _drive_list(call: Call) -> tuple[str, Any]:
    clauses = ["trashed = false"]
    if call.text("folder_id"):
        clauses.append(f"'{_drive_literal(clean_id(call.text('folder_id'), 'folder'))}' in parents")
    if call.text("name_contains"):
        clauses.append(f"name contains '{_drive_literal(call.text('name_contains'))}'")
    limit = call.number("limit", 100)
    files: list[dict] = []
    page: str | None = None
    while len(files) < limit:
        params: dict[str, Any] = {"q": " and ".join(clauses), "pageSize": min(limit - len(files), 200), "fields": f"nextPageToken,files({_DRIVE_FIELDS})", "supportsAllDrives": "true", "includeItemsFromAllDrives": "true", "orderBy": "modifiedTime desc"}
        if page:
            params["pageToken"] = page
        body = await call.json(S_DRIVE, "GET", DRIVE_API, params=params)
        files.extend(_drive_row(f) for f in body.get("files") or [])
        page = body.get("nextPageToken")
        if not page:
            break
    return f"{len(files)} file(s)", {"files": files, "count": len(files), "file_ids": [f["id"] for f in files], "names": [f["name"] for f in files]}


async def _drive_upload(call: Call) -> tuple[str, Any]:
    if call.load_file is None:
        raise GoogleError("uploading isn't available here")
    source = call.need("upload_file_id", "file to upload")
    loaded = await call.load_file(source)
    if loaded is None:
        raise GoogleError("the file to upload wasn't found — pass an uploaded or generated file's id")
    name, content_type, data = loaded
    if len(data) > MAX_FILE_BYTES:
        raise GoogleError("that file is over the 50 MB limit for this node")
    metadata: dict[str, Any] = {"name": call.text("name", name)}
    if call.text("folder_id"):
        metadata["parents"] = [clean_id(call.text("folder_id"), "folder")]
    boundary = f"ff{uuid.uuid4().hex}"
    body = (
        f"--{boundary}\r\nContent-Type: application/json; charset=UTF-8\r\n\r\n{json.dumps(metadata)}\r\n"
        f"--{boundary}\r\nContent-Type: {content_type or 'application/octet-stream'}\r\n\r\n"
    ).encode() + data + f"\r\n--{boundary}--\r\n".encode()
    result = await call.json(S_DRIVE, "POST", DRIVE_UPLOAD, params={"uploadType": "multipart", "supportsAllDrives": "true", "fields": _DRIVE_FIELDS}, content=body, headers={"Content-Type": f"multipart/related; boundary={boundary}"}, timeout=120.0)
    row = _drive_row(result)
    return f"uploaded {row['name']}", {**row, "drive_file_id": row["id"]}


async def _drive_download(call: Call) -> tuple[str, Any]:
    if call.save_file is None:
        raise GoogleError("downloading isn't available here")
    file_id = call.file_ref("file_id", "file")
    meta = await call.json(S_DRIVE, "GET", f"{DRIVE_API}/{_q(file_id)}", params={"supportsAllDrives": "true", "fields": _DRIVE_FIELDS})
    mime = meta.get("mimeType") or ""
    if mime.startswith("application/vnd.google-apps."):
        return await _export(call, file_id, call.text("format") or _NATIVE_DEFAULT_EXPORT.get(mime, "pdf"))
    if int(meta.get("size") or 0) > MAX_FILE_BYTES:
        raise GoogleError("that file is over the 50 MB limit for this node")
    resp = await call.api(S_DRIVE, "GET", f"{DRIVE_API}/{_q(file_id)}", params={"alt": "media", "supportsAllDrives": "true"}, timeout=120.0)
    name = meta.get("name") or file_id
    saved = await call.save_file(name, mime or "application/octet-stream", resp.content)
    return f"downloaded {name}", {"file_id": saved, "file_name": name, "size": len(resp.content)}


async def _drive_create_folder(call: Call) -> tuple[str, Any]:
    name = call.need("name", "folder name")
    metadata: dict[str, Any] = {"name": name, "mimeType": "application/vnd.google-apps.folder"}
    if call.text("folder_id"):
        metadata["parents"] = [clean_id(call.text("folder_id"), "parent folder")]
    body = await call.json(S_DRIVE, "POST", DRIVE_API, params={"supportsAllDrives": "true", "fields": _DRIVE_FIELDS}, json=metadata)
    row = _drive_row(body)
    return f"created folder {name}", {**row, "folder_id": row["id"]}


async def _drive_copy(call: Call) -> tuple[str, Any]:
    new_id = await _copy(call, call.file_ref("file_id", "file"), call.text("name"), call.text("folder_id"))
    return f"copied to {new_id}", {"id": new_id, "drive_file_id": new_id}


async def _drive_move(call: Call) -> tuple[str, Any]:
    file_id = call.file_ref("file_id", "file")
    target = clean_id(call.need("folder_id", "folder to move to"), "folder")
    meta = await call.json(S_DRIVE, "GET", f"{DRIVE_API}/{_q(file_id)}", params={"supportsAllDrives": "true", "fields": "parents"})
    params = {"addParents": target, "removeParents": ",".join(meta.get("parents") or []), "supportsAllDrives": "true", "fields": _DRIVE_FIELDS}
    body = await call.json(S_DRIVE, "PATCH", f"{DRIVE_API}/{_q(file_id)}", params=params, json={})
    return "moved the file", _drive_row(body)


async def _drive_rename(call: Call) -> tuple[str, Any]:
    body = await call.json(S_DRIVE, "PATCH", f"{DRIVE_API}/{_q(call.file_ref('file_id', 'file'))}", params={"supportsAllDrives": "true", "fields": _DRIVE_FIELDS}, json={"name": call.need("name", "new name")})
    return "renamed the file", _drive_row(body)


async def _drive_trash(call: Call) -> tuple[str, Any]:
    # To the bin, where it can still be restored — never a permanent delete.
    file_id = call.file_ref("file_id", "file")
    await call.json(S_DRIVE, "PATCH", f"{DRIVE_API}/{_q(file_id)}", params={"supportsAllDrives": "true", "fields": "id"}, json={"trashed": True})
    return "moved the file to the bin", {"id": file_id}


async def _drive_share(call: Call) -> tuple[str, Any]:
    file_id = call.file_ref("file_id", "file")
    kind = call.text("share_with", "user")
    role = call.text("role", "reader")
    if role not in ("reader", "commenter", "writer"):
        raise GoogleError(f"unknown role {role}")
    permission: dict[str, Any] = {"type": kind, "role": role}
    if kind in ("user", "group"):
        permission["emailAddress"] = call.need("email", "email address")
    elif kind == "domain":
        permission["domain"] = call.need("domain", "domain")
    elif kind != "anyone":
        raise GoogleError(f"can't share with {kind}")
    params = {"supportsAllDrives": "true", "sendNotificationEmail": "true" if call.yes("notify") else "false", "fields": "id"} if kind in ("user", "group") else {"supportsAllDrives": "true", "fields": "id"}
    body = await call.json(S_DRIVE, "POST", f"{DRIVE_API}/{_q(file_id)}/permissions", params=params, json=permission)
    meta = await call.json(S_DRIVE, "GET", f"{DRIVE_API}/{_q(file_id)}", params={"supportsAllDrives": "true", "fields": "webViewLink"})
    return f"shared as {role}", {"permission_id": body.get("id"), "url": meta.get("webViewLink")}


# ---- Calendar --------------------------------------------------------------------------------


def _rfc3339(value: str) -> str:
    text = value.strip()
    if re.fullmatch(r"\d{4}-\d{2}-\d{2}", text):
        return f"{text}T00:00:00Z"
    return text if re.search(r"(Z|[+-]\d{2}:?\d{2})$", text) else f"{text}Z"


def _when(value: str, zone: str) -> dict:
    text = value.strip()
    if re.fullmatch(r"\d{4}-\d{2}-\d{2}", text):
        return {"date": text}
    return {"dateTime": text, "timeZone": zone} if zone else {"dateTime": _rfc3339(text)}


def _event_row(e: dict) -> dict:
    start, end = e.get("start") or {}, e.get("end") or {}
    return {
        "id": e.get("id"), "summary": e.get("summary", ""), "description": e.get("description", ""), "location": e.get("location", ""),
        "start": start.get("dateTime") or start.get("date"), "end": end.get("dateTime") or end.get("date"), "status": e.get("status"),
        "url": e.get("htmlLink"), "meet_link": e.get("hangoutLink"), "attendees": [a.get("email") for a in e.get("attendees") or []],
    }


def _calendar_url(call: Call, event: str | None = None) -> str:
    base = f"{CALENDAR}/{_q(call.text('calendar_id', 'primary'))}/events"
    return f"{base}/{_q(event)}" if event else base


def _event_body(call: Call) -> dict:
    zone = call.text("time_zone")
    body: dict[str, Any] = {}
    for key, name in (("summary", "summary"), ("description", "description"), ("location", "location")):
        if call.text(key):
            body[name] = call.text(key)
    if call.text("start"):
        body["start"] = _when(call.text("start"), zone)
    if call.text("end"):
        body["end"] = _when(call.text("end"), zone)
    attendees = [a.strip() for a in re.split(r"[,;\s]+", call.text("attendees")) if a.strip()]
    if attendees:
        body["attendees"] = [{"email": a} for a in attendees]
    return body


async def _calendar_list(call: Call) -> tuple[str, Any]:
    params: dict[str, Any] = {"singleEvents": "true", "orderBy": "startTime", "maxResults": call.number("limit", 50, high=500)}
    if call.text("time_min"):
        params["timeMin"] = _rfc3339(call.text("time_min"))
    if call.text("time_max"):
        params["timeMax"] = _rfc3339(call.text("time_max"))
    if call.text("query"):
        params["q"] = call.text("query")
    body = await call.json(S_CALENDAR, "GET", _calendar_url(call), params=params)
    events = [_event_row(e) for e in body.get("items") or []]
    return f"{len(events)} event(s)", {"events": events, "count": len(events)}


def _send_updates(call: Call) -> str:
    return "all" if call.yes("notify") else "none"


async def _calendar_create(call: Call) -> tuple[str, Any]:
    body = _event_body(call)
    if not body.get("summary"):
        raise GoogleError("set the event's title")
    if "start" not in body or "end" not in body:
        raise GoogleError("set the event's start and end")
    params: dict[str, Any] = {"sendUpdates": _send_updates(call)}
    if call.yes("add_meet"):
        params["conferenceDataVersion"] = 1
        body["conferenceData"] = {"createRequest": {"requestId": uuid.uuid4().hex, "conferenceSolutionKey": {"type": "hangoutsMeet"}}}
    event = await call.json(S_CALENDAR, "POST", _calendar_url(call), params=params, json=body)
    row = _event_row(event)
    return f"created event {row['summary']}", {**row, "event_id": row["id"]}


async def _calendar_update(call: Call) -> tuple[str, Any]:
    event_id = call.need("event_id", "event id")
    body = _event_body(call)
    if not body:
        raise GoogleError("set at least one thing to change")
    event = await call.json(S_CALENDAR, "PATCH", _calendar_url(call, event_id), params={"sendUpdates": _send_updates(call)}, json=body)
    row = _event_row(event)
    return f"updated event {row['summary']}", {**row, "event_id": row["id"]}


async def _calendar_delete(call: Call) -> tuple[str, Any]:
    event_id = call.need("event_id", "event id")
    await call.api(S_CALENDAR, "DELETE", _calendar_url(call, event_id), params={"sendUpdates": _send_updates(call)})
    return f"deleted event {event_id}", {"event_id": event_id}


# ---- Gmail -----------------------------------------------------------------------------------


async def _gmail_send(call: Call) -> tuple[str, Any]:
    to = call.need("to", "recipient")
    message = EmailMessage()
    body = _plain(call.settings.get("body"))
    if call.text("body_format") == "html":
        message.set_content("This message needs an HTML-capable mail reader.")
        message.add_alternative(body, subtype="html")
    else:
        message.set_content(body)
    try:
        message["To"] = to
        if call.text("cc"):
            message["Cc"] = call.text("cc")
        if call.text("bcc"):
            message["Bcc"] = call.text("bcc")
        if call.text("reply_to"):
            message["Reply-To"] = call.text("reply_to")
        message["Subject"] = call.text("subject")
    except ValueError as exc:
        raise GoogleError(f"an email header isn't valid: {exc}") from exc
    attachment = call.text("attachment_file_id")
    if attachment:
        if call.load_file is None:
            raise GoogleError("attachments aren't available here")
        loaded = await call.load_file(attachment)
        if loaded is None:
            raise GoogleError("the attachment wasn't found — pass an uploaded or generated file's id")
        name, content_type, data = loaded
        if len(data) > MAX_EMAIL_BYTES:
            raise GoogleError("that attachment is over Gmail's 25 MB limit")
        main, _, sub = (content_type or "application/octet-stream").partition("/")
        message.add_attachment(data, maintype=main, subtype=sub or "octet-stream", filename=name)
    raw = base64.urlsafe_b64encode(message.as_bytes()).decode()
    sent = await call.json(S_GMAIL_SEND, "POST", f"{GMAIL}/messages/send", json={"raw": raw})
    return f"sent to {to}", {"message_id": sent.get("id"), "thread_id": sent.get("threadId")}


def _gmail_headers(payload: dict) -> dict[str, str]:
    return {h.get("name", "").lower(): h.get("value", "") for h in payload.get("headers") or []}


def _gmail_row(message: dict) -> dict:
    headers = _gmail_headers(message.get("payload") or {})
    return {"id": message.get("id"), "thread_id": message.get("threadId"), "snippet": message.get("snippet", ""), "from": headers.get("from", ""), "to": headers.get("to", ""), "subject": headers.get("subject", ""), "date": headers.get("date", "")}


async def _gmail_search(call: Call) -> tuple[str, Any]:
    limit = call.number("limit", 10, high=100)
    params: dict[str, Any] = {"maxResults": limit}
    if call.text("query"):
        params["q"] = call.text("query")
    found = await call.json(S_GMAIL_READ, "GET", f"{GMAIL}/messages", params=params)
    gate = asyncio.Semaphore(5)

    async def one(message_id: str) -> dict:
        async with gate:
            return await call.json(S_GMAIL_READ, "GET", f"{GMAIL}/messages/{_q(message_id)}", params=[("format", "metadata"), ("metadataHeaders", "From"), ("metadataHeaders", "To"), ("metadataHeaders", "Subject"), ("metadataHeaders", "Date")])

    details = await asyncio.gather(*(one(m["id"]) for m in found.get("messages") or []))
    rows = [_gmail_row(m) for m in details]
    return f"{len(rows)} message(s)", {"messages": rows, "count": len(rows)}


def _gmail_part_text(part: dict, wanted: str) -> str:
    if part.get("mimeType") == wanted and (part.get("body") or {}).get("data"):
        data = part["body"]["data"]
        return base64.urlsafe_b64decode(data + "=" * (-len(data) % 4)).decode("utf-8", errors="replace")
    for child in part.get("parts") or []:
        text = _gmail_part_text(child, wanted)
        if text:
            return text
    return ""


async def _gmail_read(call: Call) -> tuple[str, Any]:
    message_id = clean_id(call.need("message_id", "message id"), "message id")
    message = await call.json(S_GMAIL_READ, "GET", f"{GMAIL}/messages/{_q(message_id)}", params={"format": "full"})
    payload = message.get("payload") or {}
    body = _gmail_part_text(payload, "text/plain") or _gmail_part_text(payload, "text/html")
    row = _gmail_row(message)
    return f"read message from {row['from'] or 'unknown'}", {**row, "body": body[:100_000]}


# ---- Forms -----------------------------------------------------------------------------------


def _form_questions(form: dict) -> list[dict]:
    questions: list[dict] = []
    for item in form.get("items") or []:
        question = (item.get("questionItem") or {}).get("question")
        if question:
            kind = next((k for k in question if k not in ("questionId", "required", "grading")), "")
            questions.append({"id": question.get("questionId"), "title": item.get("title", ""), "kind": kind, "required": bool(question.get("required"))})
    return questions


async def _forms_get(call: Call) -> tuple[str, Any]:
    form_id = call.file_ref("form_id", "form")
    form = await call.json(S_FORMS, "GET", f"{FORMS}/{_q(form_id)}")
    info = form.get("info") or {}
    questions = _form_questions(form)
    return f"{len(questions)} question(s)", {"form_id": form_id, "title": info.get("title", ""), "description": info.get("description", ""), "questions": questions, "responder_url": form.get("responderUri", "")}


async def _forms_responses(call: Call) -> tuple[str, Any]:
    form_id = call.file_ref("form_id", "form")
    form = await call.json(S_FORMS, "GET", f"{FORMS}/{_q(form_id)}")
    titles = {q["id"]: q["title"] for q in _form_questions(form)}
    limit = call.number("limit", 500, high=5000)
    rows: list[dict] = []
    page: str | None = None
    while len(rows) < limit:
        params: dict[str, Any] = {"pageSize": min(limit - len(rows), 500)}
        if page:
            params["pageToken"] = page
        body = await call.json(S_FORMS, "GET", f"{FORMS}/{_q(form_id)}/responses", params=params)
        for response in body.get("responses") or []:
            row = {"response_id": response.get("responseId"), "submitted_at": response.get("lastSubmittedTime"), "email": response.get("respondentEmail", "")}
            for qid, answer in (response.get("answers") or {}).items():
                values = [a.get("value", "") for a in (answer.get("textAnswers") or {}).get("answers") or []]
                row[titles.get(qid, qid)] = ", ".join(values)
            rows.append(row)
        page = body.get("nextPageToken")
        if not page:
            break
    return f"{len(rows)} response(s)", {"responses": rows, "count": len(rows)}


_Op = Callable[[Call], Awaitable[tuple[str, Any]]]

OPERATIONS: dict[str, dict[str, _Op]] = {
    "google_sheets": {
        "read": _sheets_read, "find": _sheets_find, "append": _sheets_append, "update": _sheets_update, "clear": _sheets_clear,
        "create": _sheets_create, "list_tabs": _sheets_list_tabs, "add_tab": _sheets_add_tab, "delete_tab": _sheets_delete_tab, "export": _sheets_export,
    },
    "google_docs": {"create": _docs_create, "read": _docs_read, "append": _docs_append, "fill": _docs_fill, "from_template": _docs_from_template, "export": _docs_export},
    "google_slides": {
        "create": _slides_create, "read": _slides_read, "add_slide": _slides_add, "delete_slide": _slides_delete,
        "fill": _slides_fill, "from_template": _slides_from_template, "export": _slides_export,
    },
    "google_drive": {
        "list": _drive_list, "upload": _drive_upload, "download": _drive_download, "create_folder": _drive_create_folder,
        "copy": _drive_copy, "move": _drive_move, "rename": _drive_rename, "trash": _drive_trash, "share": _drive_share,
    },
    "google_calendar": {"list_events": _calendar_list, "create_event": _calendar_create, "update_event": _calendar_update, "delete_event": _calendar_delete},
    "google_gmail": {"send": _gmail_send, "search": _gmail_search, "read": _gmail_read},
    "google_forms": {"get": _forms_get, "responses": _forms_responses},
}
NODE_TYPES = frozenset(OPERATIONS)
DEFAULT_OPERATION = {"google_sheets": "read", "google_docs": "read", "google_slides": "read", "google_drive": "list", "google_calendar": "list_events", "google_gmail": "send", "google_forms": "responses"}


async def run(node_type: str, operation: str, call: Call) -> tuple[str, Any]:
    ops = OPERATIONS[node_type]
    op = ops.get(operation or DEFAULT_OPERATION[node_type])
    if op is None:
        raise GoogleError(f"unknown operation \"{operation}\"")
    return await op(call)


async def test_credentials(config: dict) -> tuple[bool, str]:
    """Signs in and asks Drive who this is — reads nothing else."""
    try:
        resp = await request(config, S_DRIVE_META, "GET", DRIVE_ABOUT, params={"fields": "user(emailAddress,displayName)"})
    except GoogleError as exc:
        return False, str(exc)
    except httpx.HTTPError as exc:
        return False, f"couldn't reach Google: {exc}"
    if resp.status_code >= 400:
        code, message = error_reason(resp)
        return False, f"signed in, but Drive said {code}: {message}"
    user = resp.json().get("user") or {}
    return True, f"Connected as {user.get('emailAddress') or user.get('displayName') or account_label(config)}"


def _f(key: str, label: str, *, secret: bool = False, required: bool = True, placeholder: str = "", help: str = "", options: list | None = None, advanced: bool = False, multiline: bool = False) -> dict:
    return {"key": key, "label": label, "secret": secret, "required": required, "placeholder": placeholder, "help": help, "options": options, "advanced": advanced, "multiline": multiline}


PROVIDERS: list[dict] = [
    {
        "id": "service_account", "name": "Service account", "tagline": "A robot account — best for shared files and server work", "environments": [],
        "docs_url": "https://cloud.google.com/iam/docs/service-accounts-create", "capabilities": [],
        "notes": "Share each file or folder with the service account's email. Service accounts have no storage of their own: to create files, pick a Shared Drive folder on the node, or use the OAuth option.",
        "fields": [
            _f("service_account_json", "Service account key (JSON)", secret=True, placeholder='{ "type": "service_account", … }', help="Google Cloud › IAM › Service accounts › Keys › Add key › JSON. Paste the whole file, or save it under Settings › Secrets and pick it here. Enable the Sheets, Docs, Slides, Drive, Calendar, Gmail and Forms APIs you need on the project."),
            _f("delegated_user", "Act as user", required=False, advanced=True, placeholder="someone@yourcompany.com", help="Google Workspace only, with domain-wide delegation allowed by your admin. Required for Gmail, and for a person's own calendar and files."),
        ],
    },
    {
        "id": "oauth", "name": "OAuth user", "tagline": "Act as one Google account — works with personal Gmail too", "environments": [],
        "docs_url": "https://developers.google.com/oauthplayground", "capabilities": [],
        "notes": "Tasks act as the account that granted the refresh token, with the permissions it granted. While the OAuth app is in Testing, Google expires refresh tokens after 7 days — publish it for long-term use.",
        "fields": [
            _f("client_id", "OAuth client ID", placeholder="…apps.googleusercontent.com"),
            _f("client_secret", "OAuth client secret", secret=True),
            _f("refresh_token", "Refresh token", secret=True, help="Create one for your own client in the OAuth 2.0 Playground (gear icon › Use your own OAuth credentials), granting the Sheets, Docs, Slides, Drive, Calendar, Gmail or Forms scopes the tasks need."),
        ],
    },
]