commit 17aba70ffad1686d5654c7bd0bba8a82db104fba Author: liuji Date: Sat Oct 10 14:48:41 2026 +0800 Add Google Sheets sync script diff --git a/.env.example b/.env.example new file mode 100755 index 0000000..f78c6b9 --- /dev/null +++ b/.env.example @@ -0,0 +1,5 @@ +# Optional local environment values. Never commit a populated .env file. +GITEA_BASE_URL=https://gitea.example.invalid +GITEA_OWNER=replace-with-gitea-owner +GIT_CREDENTIAL_SECRET= +SYNC_FAILURE_WEBHOOK_URL= diff --git a/.gitea/workflows/sync.yml b/.gitea/workflows/sync.yml new file mode 100755 index 0000000..d430a57 --- /dev/null +++ b/.gitea/workflows/sync.yml @@ -0,0 +1,39 @@ +name: Sync public Google Sheet + +on: + schedule: + # 03:00 Asia/Shanghai is 19:00 UTC on the previous day. + - cron: "0 19 * * *" + workflow_dispatch: + +permissions: + contents: write + +jobs: + sync: + runs-on: ubuntu-latest + env: + SOURCE_WORKBOOK_ID: 1DPbdVsX3SG_kgDXs2j34mDAgplRSfcAWKKG9yqalFmw + TARGET_REPOSITORY: ttshop + TARGET_BRANCH: main + PUBLISH_PATH: data/merged.html + GITEA_BASE_URL: ${{ vars.GITEA_BASE_URL }} + GITEA_OWNER: ${{ vars.GITEA_OWNER }} + GIT_CREDENTIAL_SECRET: ${{ secrets.GIT_CREDENTIAL_SECRET }} + SYNC_FAILURE_WEBHOOK_URL: ${{ secrets.SYNC_FAILURE_WEBHOOK_URL }} + steps: + - name: Check out repository + uses: actions/checkout@v4 + with: + ref: main + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: "3.11" + + - name: Install requests + run: python -m pip install --disable-pip-version-check -r requirements.txt + + - name: Synchronize workbook + run: python sync.py diff --git a/.gitignore b/.gitignore new file mode 100755 index 0000000..493d510 --- /dev/null +++ b/.gitignore @@ -0,0 +1,9 @@ +# Local environment and secrets +.env +.env.* +!.env.example + +# Python caches +__pycache__/ +*.py[cod] +.pytest_cache/ diff --git a/REAMME.md b/REAMME.md new file mode 100755 index 0000000..25fe22b --- /dev/null +++ b/REAMME.md @@ -0,0 +1,25 @@ +# TTSHOP 同步脚本 + +`sync.py` 使用 Python 标准库和 `requests` 读取公开 Google Sheets 工作簿，并把所有工作表合并成一个 UTF-8 HTML 表格，固定写入 `data/merged.html`。工作表名称和顺序来自 XLSX 导出，单元格使用公开 gviz 接口返回的显示值。 + +合并规则：每张表第一行是标题；列按首次出现顺序取并集；源表已有的 `LinkName` 会改名为 `LinkName_Source`、`LinkName_Source_2` 等并保留；脚本最后追加生成的 `LinkName` 列。HTML 文本会进行转义。内容没有变化时不产生 Git 提交，变化时只提交 `data/merged.html` 并推送到 `main`。 + +## 本地运行 + +```sh +python -m pip install -r requirements.txt +python sync.py +``` + +脚本默认使用固定工作簿 ID `1DPbdVsX3SG_kgDXs2j34mDAgplRSfcAWKKG9yqalFmw`、仓库 `ttshop`、分支 `main` 和输出路径 `data/merged.html`。本地运行需要当前目录是该仓库，并且 Git 已配置可推送的 `origin`；不要把凭据写入 remote URL。 + +## 环境变量 + +- `GIT_CREDENTIAL_SECRET`：可选的 Git 密码/token。脚本通过临时 `GIT_ASKPASS` 使用，不写入 URL、日志或提交文件。 +- `GITEA_OWNER`：Git askpass 的用户名提示值；workflow 从仓库变量注入。 +- `GITEA_BASE_URL`：Gitea 地址，作为 workflow 配置保留；实际推送使用 checkout 提供的 `origin`。 +- `SYNC_FAILURE_WEBHOOK_URL`：可选的失败通知 URL。失败时最多发送一次简短通知，通知失败不会覆盖原始异常。 + +## Gitea Actions + +`.gitea/workflows/sync.yml` 每天执行一次 `0 19 * * *`（按 UTC 解释时为 Asia/Shanghai 次日 03:00），也支持手动触发。workflow 只安装 `requirements.txt` 中的 `requests` 并运行 `python sync.py`，不依赖包 CLI、测试 extras、runner、scheduler 或报告系统。 diff --git a/requirements.txt b/requirements.txt new file mode 100755 index 0000000..d80d9fc --- /dev/null +++ b/requirements.txt @@ -0,0 +1 @@ +requests==2.32.3 diff --git a/sync.py b/sync.py new file mode 100755 index 0000000..97d5f2a --- /dev/null +++ b/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 "" + "".join( + f"<{tag}>{html.escape(value, quote=True)}" for value in values + ) + "" + + lines = [ + "", + 'TTSHOP', + "", + cells("th", headers), + *[cells("td", row) for row in rows], + "
", + "", + ] + 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