227 lines
7.5 KiB
Python
Executable File
227 lines
7.5 KiB
Python
Executable File
#!/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.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}")
|
|
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
|