Add Google Sheets sync script

sync.py

@@ -0,0 +1,224 @@
+#!/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, "headers": "0"})
+    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.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}")
+    payload = json.loads(text[len(GVIZ_PREFIX) : -1])
+    if payload.get("status") != "ok":
+        raise ValueError(f"gviz query failed for worksheet {sheet!r}")
+    table = payload.get("table", {})
+    columns = table.get("cols", [])
+    rows = table.get("rows", [])
+    width = len(columns)
+
+    def cell_value(cell: dict | None) -> str:
+        if not cell:
+            return ""
+        value = cell.get("f", cell.get("v", ""))
+        if value is None:
+            return ""
+        if isinstance(value, bool):
+            return "TRUE" if value else "FALSE"
+        return str(value)
+
+    result: list[list[str]] = []
+    for row in rows:
+        cells = row.get("c", [])
+        if len(cells) > width:
+            raise ValueError(f"gviz row is wider than columns in worksheet {sheet!r}")
+        values = [cell_value(cell) for cell in cells]
+        result.append(values + [""] * (width - len(values)))
+    return result
+
+
+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