69 lines
3.2 KiB
Python
69 lines
3.2 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
"""确认公开静态站已经提供本次生成的情报快照。"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import argparse
|
||
|
|
import json
|
||
|
|
import sys
|
||
|
|
import time
|
||
|
|
from datetime import datetime, timezone
|
||
|
|
from pathlib import Path
|
||
|
|
from urllib.error import HTTPError, URLError
|
||
|
|
from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit
|
||
|
|
from urllib.request import Request, urlopen
|
||
|
|
|
||
|
|
|
||
|
|
def check_url(base_url: str, path: str, expected_file: Path, timeout: int = 20) -> tuple[str, str]:
|
||
|
|
expected = json.loads(expected_file.read_text(encoding="utf-8"))
|
||
|
|
expected_generated = str(expected.get("generated_at") or "")
|
||
|
|
expected_as_of = str(expected.get("as_of_date") or "")
|
||
|
|
if not expected_generated:
|
||
|
|
raise ValueError("本地快照缺少 generated_at")
|
||
|
|
parts = urlsplit(base_url.rstrip("/") + "/" + path.lstrip("/"))
|
||
|
|
query = dict(parse_qsl(parts.query, keep_blank_values=True))
|
||
|
|
query["deploy-check"] = datetime.now(timezone.utc).strftime("%Y%m%d%H%M%S")
|
||
|
|
url = urlunsplit((parts.scheme, parts.netloc, parts.path, urlencode(query), ""))
|
||
|
|
request = Request(url, headers={
|
||
|
|
"Accept": "application/json",
|
||
|
|
"Cache-Control": "no-cache",
|
||
|
|
"User-Agent": "Sansheng-Intelligence-Deploy-Check/1.0",
|
||
|
|
})
|
||
|
|
with urlopen(request, timeout=timeout) as response:
|
||
|
|
remote = json.load(response)
|
||
|
|
remote_generated = str(remote.get("generated_at") or "")
|
||
|
|
remote_as_of = str(remote.get("as_of_date") or "")
|
||
|
|
if remote_generated != expected_generated or remote_as_of != expected_as_of:
|
||
|
|
raise RuntimeError(
|
||
|
|
f"公开快照仍不是本次版本：expected generated_at={expected_generated}, as_of_date={expected_as_of}; "
|
||
|
|
f"remote generated_at={remote_generated}, as_of_date={remote_as_of}"
|
||
|
|
)
|
||
|
|
return remote_generated, remote_as_of
|
||
|
|
|
||
|
|
|
||
|
|
def main() -> int:
|
||
|
|
parser = argparse.ArgumentParser(description="校验公开静态站是否已切换到指定情报快照")
|
||
|
|
parser.add_argument("--base-url", required=True, help="公开站点根地址")
|
||
|
|
parser.add_argument("--expected-file", default="data/textile-intelligence.json", help="本地情报快照")
|
||
|
|
parser.add_argument("--path", default="/data/textile-intelligence.json", help="公开 JSON 路径")
|
||
|
|
parser.add_argument("--retries", type=int, default=12, help="失败后重试次数")
|
||
|
|
parser.add_argument("--interval", type=int, default=30, help="重试间隔秒数")
|
||
|
|
args = parser.parse_args()
|
||
|
|
expected_file = Path(args.expected_file)
|
||
|
|
for attempt in range(max(0, args.retries) + 1):
|
||
|
|
try:
|
||
|
|
generated, as_of = check_url(args.base_url, args.path, expected_file)
|
||
|
|
print(f"公开快照已生效：generated_at={generated} as_of_date={as_of}")
|
||
|
|
return 0
|
||
|
|
except (HTTPError, URLError, TimeoutError, ValueError, RuntimeError) as exc:
|
||
|
|
if attempt >= max(0, args.retries):
|
||
|
|
print(f"公开快照校验失败：{exc}", file=sys.stderr)
|
||
|
|
return 1
|
||
|
|
print(f"公开快照尚未切换（{attempt + 1}/{args.retries}）：{exc}")
|
||
|
|
time.sleep(max(1, args.interval))
|
||
|
|
return 1
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
raise SystemExit(main())
|