#!/usr/bin/env python3 from __future__ import annotations import argparse import json import os import sys import time from datetime import datetime, timezone from pathlib import Path from typing import Any from urllib.parse import urlparse, urlunparse DEFAULT_EDGEONE_ZONE_ID = "zone-3qwew52ihaj2" DEFAULT_EDGEONE_ZONE_NAME = "global-interaction.com" DEFAULT_EDGEONE_ALLOWED_HOSTS = "h5.global-interaction.com,vivagames.global-interaction.com,asset.global-interaction.com" TERMINAL_STATUSES = {"success", "failed", "timeout", "canceled"} def utc_now() -> str: return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z") def load_env_file(path: str) -> None: if not path: return env_path = Path(path).expanduser().resolve() if not env_path.exists(): raise RuntimeError(f"env file not found: {env_path}") for raw_line in env_path.read_text(encoding="utf-8", errors="replace").splitlines(): line = raw_line.strip() if not line or line.startswith("#") or "=" not in line: continue key, value = line.split("=", 1) key = key.strip() value = value.strip().strip('"').strip("'") if key and key not in os.environ: os.environ[key] = value def env_value(*names: str) -> str: for name in names: value = str(os.environ.get(name) or "").strip() if value: return value return "" def require_tencent_credentials() -> tuple[str, str, str]: secret_id = env_value("TENCENTCLOUD_SECRET_ID", "TENCENT_SECRET_ID") secret_key = env_value("TENCENTCLOUD_SECRET_KEY", "TENCENT_SECRET_KEY") region = env_value("TENCENTCLOUD_REGION", "DEPLOY_REGION") or "ap-guangzhou" missing = [] if not secret_id: missing.append("TENCENTCLOUD_SECRET_ID") if not secret_key: missing.append("TENCENTCLOUD_SECRET_KEY") if missing: raise RuntimeError("missing Tencent Cloud credentials: " + ", ".join(missing)) return secret_id, secret_key, region def edgeone_allowed_hosts(raw: str) -> set[str]: hosts = { item.strip().lower() for item in str(raw or DEFAULT_EDGEONE_ALLOWED_HOSTS).split(",") if item.strip() } if not hosts: raise RuntimeError("EDGEONE_ALLOWED_HOSTS is empty") return hosts def normalize_url_item(item: str, allowed_hosts: set[str]) -> str: parsed = urlparse(str(item or "").strip()) if parsed.scheme not in {"http", "https"} or not parsed.netloc: raise ValueError(f"invalid URL: {item}") hostname = str(parsed.hostname or "").strip().lower() if hostname not in allowed_hosts: raise ValueError(f"host is not allowed for EdgeOne purge: {hostname}") return urlunparse(parsed._replace(fragment="")) def normalize_url_items(raw_items: list[str], allowed_hosts: set[str]) -> list[str]: normalized: list[str] = [] seen: set[str] = set() for item in raw_items: value = normalize_url_item(item, allowed_hosts) if value in seen: continue seen.add(value) normalized.append(value) if not normalized: raise ValueError("at least one URL is required") if len(normalized) > 200: raise ValueError("URL refresh supports at most 200 URLs per request") return normalized def build_teo_client() -> Any: secret_id, secret_key, region = require_tencent_credentials() try: from tencentcloud.common import credential from tencentcloud.common.profile.client_profile import ClientProfile from tencentcloud.common.profile.http_profile import HttpProfile from tencentcloud.teo.v20220901 import teo_client except Exception as exc: # noqa: BLE001 raise RuntimeError(f"tencentcloud EdgeOne SDK unavailable: {exc}") from exc return teo_client.TeoClient( credential.Credential(secret_id, secret_key), region, ClientProfile(httpProfile=HttpProfile(endpoint="teo.tencentcloudapi.com")), ) def call_teo(client: Any, method_name: str, payload: dict[str, Any]) -> dict[str, Any]: from tencentcloud.teo.v20220901 import models as teo_models request_class = getattr(teo_models, f"{method_name}Request") request = request_class() request.from_json_string(json.dumps(payload)) response = getattr(client, method_name)(request) return json.loads(response.to_json_string()) def create_url_purge(client: Any, zone_id: str, targets: list[str]) -> dict[str, Any]: return call_teo( client, "CreatePurgeTask", { "ZoneId": zone_id, "Type": "purge_url", "Targets": targets, }, ) def describe_purge_job(client: Any, zone_id: str, job_id: str) -> dict[str, Any]: return call_teo( client, "DescribePurgeTasks", { "ZoneId": zone_id, "Limit": 20, "Filters": [{"Name": "job-id", "Values": [job_id]}], }, ) def wait_for_purge_job(client: Any, zone_id: str, job_id: str, timeout_seconds: int) -> tuple[str, list[dict[str, Any]], dict[str, Any]]: deadline = time.time() + max(timeout_seconds, 1) latest: dict[str, Any] = {} tasks: list[dict[str, Any]] = [] while time.time() < deadline: latest = describe_purge_job(client, zone_id, job_id) tasks = list(latest.get("Tasks") or []) statuses = {str(item.get("Status") or "").strip().lower() for item in tasks} if statuses and all(status in TERMINAL_STATUSES for status in statuses): if any(status != "success" for status in statuses): return "failed", tasks, latest return "success", tasks, latest time.sleep(3) return "timeout", tasks, latest def run(args: argparse.Namespace) -> dict[str, Any]: load_env_file(args.env_file) zone_id = args.zone_id or env_value("EDGEONE_ZONE_ID") or DEFAULT_EDGEONE_ZONE_ID zone_name = args.zone_name or env_value("EDGEONE_ZONE_NAME") or DEFAULT_EDGEONE_ZONE_NAME allowed_hosts = edgeone_allowed_hosts(args.allowed_hosts or env_value("EDGEONE_ALLOWED_HOSTS")) targets = normalize_url_items(args.urls, allowed_hosts) if args.dry_run: return { "allowedHosts": sorted(allowed_hosts), "dryRun": True, "mode": "url", "ok": True, "targets": targets, "zoneId": zone_id, "zoneName": zone_name, } client = build_teo_client() submit = create_url_purge(client, zone_id, targets) job_id = str(submit.get("JobId") or "").strip() status = "submitted" tasks: list[dict[str, Any]] = [] describe: dict[str, Any] = {} if job_id and args.wait: status, tasks, describe = wait_for_purge_job(client, zone_id, job_id, args.timeout_seconds) return { "allowedHosts": sorted(allowed_hosts), "createdAt": utc_now(), "dryRun": False, "failedList": list(submit.get("FailedList") or []), "jobId": job_id, "mode": "url", "ok": status != "failed" and not submit.get("FailedList"), "requestId": str(submit.get("RequestId") or "").strip(), "status": status, "submit": submit, "targets": targets, "tasks": tasks, "describe": describe, "zoneId": zone_id, "zoneName": zone_name, } def parse_args(argv: list[str]) -> argparse.Namespace: parser = argparse.ArgumentParser(description="Submit Tencent EdgeOne URL purge tasks.") parser.add_argument("action", choices=["purge-url"]) parser.add_argument("--env-file", default="") parser.add_argument("--zone-id", default="") parser.add_argument("--zone-name", default="") parser.add_argument("--allowed-hosts", default="") parser.add_argument("--urls", action="append", default=[]) parser.add_argument("--dry-run", action="store_true") parser.add_argument("--wait", action="store_true") parser.add_argument("--timeout-seconds", type=int, default=90) args = parser.parse_args(argv) if args.action == "purge-url" and not args.urls: parser.error("--urls is required") return args def main(argv: list[str]) -> int: try: payload = run(parse_args(argv)) except Exception as exc: # noqa: BLE001 print(json.dumps({"error": str(exc), "ok": False}, ensure_ascii=False, indent=2)) return 1 print(json.dumps(payload, ensure_ascii=False, indent=2)) return 0 if payload.get("ok") is not False else 1 if __name__ == "__main__": raise SystemExit(main(sys.argv[1:]))