335 lines
12 KiB
Python
Executable File
335 lines
12 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 datetime import datetime, timezone
|
|
from io import BytesIO
|
|
from pathlib import Path
|
|
from urllib.parse import urlencode
|
|
from zoneinfo import ZoneInfo
|
|
import xml.etree.ElementTree as ET
|
|
|
|
import requests
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parent
|
|
WORKBOOK_ID = os.getenv("WORKBOOK_ID", "1DPbdVsX3SG_kgDXs2j34mDAgplRSfcAWKKG9yqalFmw")
|
|
PUBLISH_PATH = Path("data/merged.html")
|
|
SYNC_LOG_PATH = Path("data/sync-log.json")
|
|
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 read_sync_log() -> dict[str, object]:
|
|
"""Read the previously published sync log when it is available."""
|
|
path = ROOT / SYNC_LOG_PATH
|
|
if not path.exists():
|
|
return {}
|
|
try:
|
|
payload = json.loads(path.read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError):
|
|
return {}
|
|
return payload if isinstance(payload, dict) else {}
|
|
|
|
|
|
def git_data_update_times() -> tuple[str, str] | None:
|
|
"""Return the latest committed data time in UTC and Shanghai time."""
|
|
try:
|
|
result = subprocess.run(
|
|
["git", "log", "-1", "--format=%cI", "--", str(PUBLISH_PATH)],
|
|
cwd=ROOT,
|
|
check=True,
|
|
text=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
)
|
|
committed_at = result.stdout.strip()
|
|
if not committed_at:
|
|
return None
|
|
timestamp = datetime.fromisoformat(committed_at)
|
|
if timestamp.tzinfo is None:
|
|
return None
|
|
return (
|
|
timestamp.astimezone(timezone.utc).isoformat().replace("+00:00", "Z"),
|
|
timestamp.astimezone(ZoneInfo("Asia/Shanghai")).isoformat(),
|
|
)
|
|
except (OSError, subprocess.SubprocessError, ValueError):
|
|
return None
|
|
|
|
|
|
def publish(content: bytes, sync_log: dict[str, object]) -> None:
|
|
log_content = (json.dumps(sync_log, ensure_ascii=False, indent=2) + "\n").encode("utf-8")
|
|
targets = {
|
|
PUBLISH_PATH: content,
|
|
SYNC_LOG_PATH: log_content,
|
|
}
|
|
changed = False
|
|
for relative_path, target_content in targets.items():
|
|
target = ROOT / relative_path
|
|
if not target.exists() or target.read_bytes() != target_content:
|
|
changed = True
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
target.write_bytes(target_content)
|
|
if not changed:
|
|
print("data/merged.html and data/sync-log.json are unchanged")
|
|
return
|
|
|
|
env = os.environ.copy()
|
|
env["GIT_TERMINAL_PROMPT"] = "0"
|
|
askpass_path: str | None = None
|
|
credential = env.get("GIT_CREDENTIAL_SECRET", "")
|
|
paths = [str(PUBLISH_PATH), str(SYNC_LOG_PATH)]
|
|
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", "--", *paths], env)
|
|
staged = subprocess.run(
|
|
["git", "diff", "--cached", "--quiet", "--", *paths],
|
|
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 and data/sync-log.json are unchanged")
|
|
return
|
|
run_git([
|
|
"-c", "user.name=TTSHOP Sync", "-c", "user.email=ttshop-sync@localhost",
|
|
"commit", "--only", "--no-gpg-sign", "-m", "Update merged Google Sheet and sync log",
|
|
"--", *paths,
|
|
], env)
|
|
run_git(["push", "origin", "HEAD:main"], env)
|
|
print("published data/merged.html and data/sync-log.json")
|
|
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()
|
|
worksheet_names = sheet_names(response.content)
|
|
sheets = []
|
|
for name in worksheet_names:
|
|
headers, rows = normalize_sheet(name, read_gviz(name))
|
|
if headers:
|
|
sheets.append((name, headers, rows))
|
|
headers, rows = merge_sheets(sheets)
|
|
content = render_html(headers, rows)
|
|
previous_sync_log = read_sync_log()
|
|
previous_data_path = ROOT / PUBLISH_PATH
|
|
data_changed = (
|
|
not previous_data_path.exists()
|
|
or previous_data_path.read_bytes() != content
|
|
)
|
|
success_at = datetime.now(timezone.utc)
|
|
if data_changed:
|
|
last_data_update_at = success_at.isoformat().replace("+00:00", "Z")
|
|
last_data_update_at_shanghai = success_at.astimezone(ZoneInfo("Asia/Shanghai")).isoformat()
|
|
else:
|
|
last_data_update_at = previous_sync_log.get("last_data_update_at")
|
|
last_data_update_at_shanghai = previous_sync_log.get("last_data_update_at_shanghai")
|
|
if not isinstance(last_data_update_at, str) or not last_data_update_at:
|
|
historical_times = git_data_update_times()
|
|
if historical_times is None:
|
|
last_data_update_at = None
|
|
last_data_update_at_shanghai = None
|
|
else:
|
|
last_data_update_at, last_data_update_at_shanghai = historical_times
|
|
elif not isinstance(last_data_update_at_shanghai, str) or not last_data_update_at_shanghai:
|
|
last_data_update_at_shanghai = None
|
|
sync_log = {
|
|
"last_success_at": success_at.isoformat().replace("+00:00", "Z"),
|
|
"last_success_at_shanghai": success_at.astimezone(ZoneInfo("Asia/Shanghai")).isoformat(),
|
|
"last_data_update_at": last_data_update_at,
|
|
"last_data_update_at_shanghai": last_data_update_at_shanghai,
|
|
"worksheet_count": len(worksheet_names),
|
|
"record_count": len(rows),
|
|
"artifact_bytes": len(content),
|
|
}
|
|
publish(content, sync_log)
|
|
|
|
|
|
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
|