ttshop/sync.py

#!/usr/bin/env python3
"""Sync a public Google Sheets workbook into data/merged.html."""

from __future__ import annotations

import html
import json
import os
import subprocess
import sys
import tempfile
import zipfile
from io import BytesIO
from pathlib import Path
from urllib.parse import urlencode
import xml.etree.ElementTree as ET

import requests


ROOT = Path(__file__).resolve().parent
WORKBOOK_ID = "1DPbdVsX3SG_kgDXs2j34mDAgplRSfcAWKKG9yqalFmw"
PUBLISH_PATH = Path("data/merged.html")
GVIZ_PREFIX = "google.visualization.Query.setResponse("


def sheet_names(xlsx: bytes) -> list[str]:
    """Return worksheet names in the order stored by the workbook."""
    with zipfile.ZipFile(BytesIO(xlsx)) as package:
        root = ET.fromstring(package.read("xl/workbook.xml"))
    names = [
        node.attrib["name"]
        for node in root.iter()
        if node.tag.rsplit("}", 1)[-1] == "sheet" and node.attrib.get("name")
    ]
    if not names:
        raise ValueError("workbook contains no worksheets")
    return names


def read_gviz(sheet: str) -> list[list[str]]:
    params = urlencode({"sheet": sheet})
    url = f"https://docs.google.com/spreadsheets/d/{WORKBOOK_ID}/gviz/tq?{params}"
    response = requests.get(url, timeout=30)
    response.raise_for_status()
    text = response.text.strip()
    if text.startswith("/*O_o*/"):
        text = text[len("/*O_o*/") :].strip()
    if text.endswith(";"):
        text = text[:-1].rstrip()
    if not text.startswith(GVIZ_PREFIX) or not text.endswith(")"):
        raise ValueError(f"invalid gviz response for worksheet {sheet!r}")
    try:
        payload = json.loads(text[len(GVIZ_PREFIX) : -1])
    except json.JSONDecodeError as error:
        raise ValueError(f"invalid gviz response for worksheet {sheet!r}") from error
    if not isinstance(payload, dict) or payload.get("status") != "ok":
        raise ValueError(f"gviz query failed for worksheet {sheet!r}")
    table = payload.get("table")
    if not isinstance(table, dict):
        raise ValueError(f"invalid gviz table for worksheet {sheet!r}")
    columns = table.get("cols")
    rows = table.get("rows")
    if not isinstance(columns, list) or not isinstance(rows, list):
        raise ValueError(f"invalid gviz table for worksheet {sheet!r}")

    headers: list[str] = []
    for column in columns:
        if not isinstance(column, dict):
            raise ValueError(f"invalid gviz column for worksheet {sheet!r}")
        label = column.get("label")
        if label is not None and not isinstance(label, str):
            raise ValueError(f"invalid gviz column label for worksheet {sheet!r}")
        headers.append(label or "")

    while headers and headers[-1] == "":
        headers.pop()
    if not headers:
        return []

    def cell_value(cell: dict | None) -> str:
        if cell is None:
            return ""
        if not isinstance(cell, dict):
            raise ValueError(f"invalid gviz cell for worksheet {sheet!r}")
        value = cell.get("f") if cell.get("f") is not None else cell.get("v")
        if value is None:
            return ""
        if isinstance(value, bool):
            return "TRUE" if value else "FALSE"
        return str(value)

    data_rows: list[list[str]] = []
    for row in rows:
        if not isinstance(row, dict) or not isinstance(row.get("c", []), list):
            raise ValueError(f"invalid gviz row for worksheet {sheet!r}")
        cells = row["c"]
        if len(cells) > len(columns):
            raise ValueError(f"gviz row is wider than columns in worksheet {sheet!r}")
        values = [cell_value(cell) for cell in cells]
        data_rows.append((values + [""] * (len(columns) - len(values)))[: len(headers)])
    return [headers] + data_rows


def normalize_sheet(name: str, rows: list[list[str]]) -> tuple[list[str], list[list[str]]]:
    if not rows:
        return [], []
    headers = rows[0]
    if any(header == "" for header in headers):
        raise ValueError(f"worksheet {name!r} has an empty header")
    if len(set(headers)) != len(headers):
        raise ValueError(f"worksheet {name!r} has duplicate headers")

    final_headers = headers[:]
    if "LinkName" in final_headers:
        index = final_headers.index("LinkName")
        suffix = 1
        candidate = "LinkName_Source"
        while candidate in final_headers:
            suffix += 1
            candidate = f"LinkName_Source_{suffix}"
        final_headers[index] = candidate

    data = [row + [""] * (len(headers) - len(row)) for row in rows[1:]]
    return final_headers, data


def merge_sheets(sheets: list[tuple[str, list[str], list[list[str]]]]) -> tuple[list[str], list[list[str]]]:
    """Union columns by first appearance and append each source row once."""
    columns: list[str] = []
    for _, headers, _ in sheets:
        for header in headers:
            if header not in columns:
                columns.append(header)
    if not columns:
        return ["LinkName"], []

    output = [columns + ["LinkName"]]
    for sheet_name, headers, rows in sheets:
        positions = {header: index for index, header in enumerate(headers)}
        for row in rows:
            output.append([
                row[positions[column]] if column in positions else ""
                for column in columns
            ] + [sheet_name])
    return output[0], output[1:]


def render_html(headers: list[str], rows: list[list[str]]) -> bytes:
    def cells(tag: str, values: list[str]) -> str:
        return "<tr>" + "".join(
            f"<{tag}>{html.escape(value, quote=True)}</{tag}>" for value in values
        ) + "</tr>"

    lines = [
        "<!doctype html>",
        '<html><head><meta charset="utf-8"><title>TTSHOP</title></head><body>',
        "<table>",
        cells("th", headers),
        *[cells("td", row) for row in rows],
        "</table>",
        "</body></html>",
    ]
    return ("\n".join(lines) + "\n").encode("utf-8")


def run_git(args: list[str], env: dict[str, str]) -> subprocess.CompletedProcess[str]:
    return subprocess.run(
        ["git", *args], cwd=ROOT, env=env, check=True, text=True,
        stdout=subprocess.PIPE, stderr=subprocess.PIPE,
    )


def publish(content: bytes) -> None:
    target = ROOT / PUBLISH_PATH
    if target.exists() and target.read_bytes() == content:
        print("data/merged.html is unchanged")
        return
    target.parent.mkdir(parents=True, exist_ok=True)
    target.write_bytes(content)

    env = os.environ.copy()
    env["GIT_TERMINAL_PROMPT"] = "0"
    askpass_path: str | None = None
    credential = env.get("GIT_CREDENTIAL_SECRET", "")
    if credential:
        with tempfile.NamedTemporaryFile("w", prefix="ttshop-askpass-", delete=False) as file:
            file.write(
                "#!/bin/sh\n"
                'case "$1" in\n'
                '*Username*) printf "%s\\n" "${GITEA_OWNER:-oauth2}" ;;\n'
                '*Password*) printf "%s\\n" "$GIT_CREDENTIAL_SECRET" ;;\n'
                'esac\n'
            )
            askpass_path = file.name
        os.chmod(askpass_path, 0o700)
        env["GIT_ASKPASS"] = askpass_path

    try:
        run_git(["add", "--", str(PUBLISH_PATH)], env)
        staged = subprocess.run(
            ["git", "diff", "--cached", "--quiet", "--", str(PUBLISH_PATH)],
            cwd=ROOT, env=env, check=False,
        )
        if staged.returncode not in (0, 1):
            raise RuntimeError("could not inspect staged output")
        if staged.returncode == 0:
            print("data/merged.html is unchanged")
            return
        run_git([
            "-c", "user.name=TTSHOP Sync", "-c", "user.email=ttshop-sync@localhost",
            "commit", "--no-gpg-sign", "-m", "Update merged Google Sheet",
        ], env)
        run_git(["push", "origin", "HEAD:main"], env)
        print("published data/merged.html")
    finally:
        if askpass_path:
            Path(askpass_path).unlink(missing_ok=True)


def sync() -> None:
    export_url = f"https://docs.google.com/spreadsheets/d/{WORKBOOK_ID}/export?format=xlsx"
    response = requests.get(export_url, timeout=30)
    response.raise_for_status()
    sheets = []
    for name in sheet_names(response.content):
        headers, rows = normalize_sheet(name, read_gviz(name))
        if headers:
            sheets.append((name, headers, rows))
    headers, rows = merge_sheets(sheets)
    publish(render_html(headers, rows))


def notify_failure(error: Exception) -> None:
    url = os.getenv("SYNC_FAILURE_WEBHOOK_URL")
    if not url:
        return
    try:
        requests.post(url, json={"text": f"TTSHOP sync failed: {type(error).__name__}"}, timeout=10)
    except requests.RequestException:
        pass


if __name__ == "__main__":
    try:
        sync()
    except Exception as error:
        notify_failure(error)
        print(f"sync failed: {error}", file=sys.stderr)
        raise