327 lines
12 KiB
Python
327 lines
12 KiB
Python
from __future__ import annotations
|
||
|
||
import json
|
||
import time
|
||
from datetime import datetime, timedelta
|
||
from typing import Any
|
||
from urllib.parse import urlparse, urlunparse
|
||
|
||
from core.config import (
|
||
CDN_ALLOWED_DOMAINS,
|
||
CDN_PURGE_TASK_LIMIT,
|
||
DEPLOY_REGION,
|
||
TENCENT_SECRET_ID,
|
||
TENCENT_SECRET_KEY,
|
||
utc_now,
|
||
)
|
||
|
||
|
||
URL_PURGE_BATCH_LIMIT = 1000
|
||
PATH_PURGE_BATCH_LIMIT = 500
|
||
|
||
|
||
def cdn_configuration_error() -> str:
|
||
# CDN 刷新和现有腾讯云模块共用同一组凭据；少任意一项都直接视为未配置。
|
||
missing: list[str] = []
|
||
if not TENCENT_SECRET_ID:
|
||
missing.append("TENCENT_SECRET_ID")
|
||
if not TENCENT_SECRET_KEY:
|
||
missing.append("TENCENT_SECRET_KEY")
|
||
if not DEPLOY_REGION:
|
||
missing.append("DEPLOY_REGION")
|
||
|
||
if missing:
|
||
return f"cdn refresh requires .env: {', '.join(missing)}"
|
||
return ""
|
||
|
||
|
||
def normalized_cdn_allowed_domains() -> set[str]:
|
||
# 域名白名单统一转成小写集合，前后端展示和后端校验保持一套口径。
|
||
return {str(item or "").strip().lower() for item in CDN_ALLOWED_DOMAINS if str(item or "").strip()}
|
||
|
||
|
||
def build_cdn_client() -> tuple[Any | None, str]:
|
||
# 先兜底检查配置，避免没有凭据时还继续初始化 SDK。
|
||
error = cdn_configuration_error()
|
||
if error:
|
||
return None, error
|
||
|
||
try:
|
||
from tencentcloud.cdn.v20180606 import cdn_client
|
||
from tencentcloud.common import credential
|
||
from tencentcloud.common.profile.client_profile import ClientProfile
|
||
from tencentcloud.common.profile.http_profile import HttpProfile
|
||
except Exception as exc: # noqa: BLE001
|
||
return None, f"tencentcloud sdk unavailable: {exc}"
|
||
|
||
cred = credential.Credential(TENCENT_SECRET_ID, TENCENT_SECRET_KEY)
|
||
client = cdn_client.CdnClient(
|
||
cred,
|
||
DEPLOY_REGION,
|
||
ClientProfile(httpProfile=HttpProfile(endpoint="cdn.tencentcloudapi.com")),
|
||
)
|
||
return client, ""
|
||
|
||
|
||
def invoke_cdn(action: Any, label: str, retries: int = 3, delay_seconds: int = 2) -> Any:
|
||
# CDN SDK 异常和 CLB/TAT 一样集中做轻量重试，减少瞬时网络抖动带来的假失败。
|
||
from tencentcloud.common.exception.tencent_cloud_sdk_exception import TencentCloudSDKException
|
||
|
||
last_error: Exception | None = None
|
||
for attempt in range(1, retries + 1):
|
||
try:
|
||
return action()
|
||
except TencentCloudSDKException as exc:
|
||
last_error = exc
|
||
if attempt == retries:
|
||
break
|
||
time.sleep(delay_seconds * attempt)
|
||
|
||
if last_error is not None:
|
||
raise last_error
|
||
raise RuntimeError(f"{label} failed without error")
|
||
|
||
|
||
def call_cdn(method_name: str, payload: dict[str, Any]) -> dict[str, Any]:
|
||
# 所有 CDN 调用统一走 JSON 请求体，后续补字段时不需要反复修改 SDK 模型赋值逻辑。
|
||
client, error = build_cdn_client()
|
||
if client is None:
|
||
raise RuntimeError(error)
|
||
|
||
from tencentcloud.cdn.v20180606 import models as cdn_models
|
||
|
||
request_class = getattr(cdn_models, f"{method_name}Request")
|
||
request = request_class()
|
||
request.from_json_string(json.dumps(payload))
|
||
response = invoke_cdn(lambda: getattr(client, method_name)(request), method_name)
|
||
payload = json.loads(response.to_json_string())
|
||
if isinstance(payload, dict) and isinstance(payload.get("Response"), dict):
|
||
return dict(payload.get("Response") or {})
|
||
return dict(payload or {})
|
||
|
||
|
||
def normalize_cdn_purge_type(value: Any) -> str:
|
||
# 工具页只开放 URL 和目录两种刷新类型。
|
||
purge_type = str(value or "url").strip().lower()
|
||
if purge_type not in {"url", "path"}:
|
||
raise ValueError("purgeType must be url or path")
|
||
return purge_type
|
||
|
||
|
||
def normalize_cdn_area(value: Any, *, allow_global: bool = False) -> str:
|
||
# 刷新提交只接受境内/境外；记录筛选额外兼容 global。
|
||
area = str(value or "").strip().lower()
|
||
if not area:
|
||
return ""
|
||
|
||
allowed = {"mainland", "overseas"}
|
||
if allow_global:
|
||
allowed.add("global")
|
||
if area not in allowed:
|
||
raise ValueError(f"area must be one of: {', '.join(sorted(allowed))}")
|
||
return area
|
||
|
||
|
||
def normalize_cdn_flush_type(purge_type: str, value: Any) -> str:
|
||
# URL 刷新没有 flushType；目录刷新默认走更保守的 flush。
|
||
flush_type = str(value or "flush").strip().lower()
|
||
if purge_type == "url":
|
||
return ""
|
||
if flush_type not in {"flush", "delete"}:
|
||
raise ValueError("flushType must be flush or delete")
|
||
return flush_type
|
||
|
||
|
||
def normalize_cdn_items(raw_items: Any) -> list[str]:
|
||
# 前端可直接传数组；兜底兼容换行文本，方便后续脚本接口复用。
|
||
if isinstance(raw_items, list):
|
||
candidates = raw_items
|
||
elif isinstance(raw_items, str):
|
||
candidates = raw_items.splitlines()
|
||
else:
|
||
raise ValueError("items must be a list or string")
|
||
|
||
items: list[str] = []
|
||
seen: set[str] = set()
|
||
for item in candidates:
|
||
text = str(item or "").strip()
|
||
if not text or text in seen:
|
||
continue
|
||
seen.add(text)
|
||
items.append(text)
|
||
|
||
if not items:
|
||
raise ValueError("至少填写一条 URL 或目录")
|
||
return items
|
||
|
||
|
||
def normalize_cdn_item(item: str, *, purge_type: str, allowed_domains: set[str]) -> str:
|
||
# 每条刷新目标都强制要求完整 URL，避免误把裸路径提交到错误域名。
|
||
parsed = urlparse(item)
|
||
if parsed.scheme not in {"http", "https"} or not parsed.netloc:
|
||
raise ValueError(f"无效 CDN 地址: {item}")
|
||
|
||
hostname = str(parsed.hostname or "").strip().lower()
|
||
if allowed_domains and hostname not in allowed_domains:
|
||
raise ValueError(f"域名未加入 CDN_ALLOWED_DOMAINS: {hostname}")
|
||
|
||
if purge_type == "path":
|
||
normalized_path = parsed.path or "/"
|
||
if not normalized_path.endswith("/"):
|
||
raise ValueError(f"目录刷新必须以 / 结尾: {item}")
|
||
parsed = parsed._replace(path=normalized_path, query="", fragment="")
|
||
else:
|
||
parsed = parsed._replace(fragment="")
|
||
|
||
return urlunparse(parsed)
|
||
|
||
|
||
def normalize_cdn_purge_items(raw_items: Any, *, purge_type: str) -> list[str]:
|
||
# 去重、域名校验和目录校验统一收口，提交接口和测试都复用这套规则。
|
||
items = normalize_cdn_items(raw_items)
|
||
allowed_domains = normalized_cdn_allowed_domains()
|
||
max_batch = URL_PURGE_BATCH_LIMIT if purge_type == "url" else PATH_PURGE_BATCH_LIMIT
|
||
if len(items) > max_batch:
|
||
raise ValueError(f"{'URL' if purge_type == 'url' else '目录'} 单次最多提交 {max_batch} 条")
|
||
|
||
normalized: list[str] = []
|
||
seen: set[str] = set()
|
||
for item in items:
|
||
value = normalize_cdn_item(item, purge_type=purge_type, allowed_domains=allowed_domains)
|
||
if value in seen:
|
||
continue
|
||
seen.add(value)
|
||
normalized.append(value)
|
||
return normalized
|
||
|
||
|
||
def normalize_quota_rows(items: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||
# 配额结构只保留页面展示会用到的字段，前端不需要感知腾讯云原始大小写命名。
|
||
rows: list[dict[str, Any]] = []
|
||
for item in items:
|
||
rows.append(
|
||
{
|
||
"area": str(item.get("Area") or "").strip().lower(),
|
||
"batch": int(item.get("Batch") or 0),
|
||
"total": int(item.get("Total") or 0),
|
||
"available": int(item.get("Available") or 0),
|
||
}
|
||
)
|
||
return rows
|
||
|
||
|
||
def normalize_purge_task_rows(items: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||
# 刷新记录统一转成前端直接可表格渲染的结构，并按提交时间倒序排列。
|
||
rows: list[dict[str, Any]] = []
|
||
for item in items:
|
||
rows.append(
|
||
{
|
||
"taskId": str(item.get("TaskId") or "").strip(),
|
||
"url": str(item.get("Url") or "").strip(),
|
||
"status": str(item.get("Status") or "").strip().lower(),
|
||
"purgeType": str(item.get("PurgeType") or "").strip().lower(),
|
||
"flushType": str(item.get("FlushType") or "").strip().lower(),
|
||
"createTime": str(item.get("CreateTime") or "").strip(),
|
||
}
|
||
)
|
||
return sorted(rows, key=lambda item: item.get("createTime") or "", reverse=True)
|
||
|
||
|
||
def recent_task_query_window() -> tuple[str, str]:
|
||
# CDN 记录查询要求传 TaskId 或时间范围；工具页默认拉最近 7 天窗口。
|
||
end_at = datetime.now().replace(microsecond=0)
|
||
start_at = end_at - timedelta(days=7)
|
||
return start_at.strftime("%Y-%m-%d %H:%M:%S"), end_at.strftime("%Y-%m-%d %H:%M:%S")
|
||
|
||
|
||
def get_cdn_overview_payload() -> dict[str, Any]:
|
||
# 通用工具页一次拿到配置状态、配额和最近刷新记录，减少前端额外往返。
|
||
payload: dict[str, Any] = {
|
||
"ok": False,
|
||
"configured": False,
|
||
"updatedAt": utc_now(),
|
||
"error": "",
|
||
"warnings": [],
|
||
"settings": {
|
||
"allowedDomains": sorted(normalized_cdn_allowed_domains()),
|
||
"taskLimit": max(int(CDN_PURGE_TASK_LIMIT or 20), 1),
|
||
},
|
||
"quota": {
|
||
"url": [],
|
||
"path": [],
|
||
},
|
||
"tasks": [],
|
||
"taskTotal": 0,
|
||
}
|
||
|
||
error = cdn_configuration_error()
|
||
if error:
|
||
payload["error"] = error
|
||
return payload
|
||
|
||
payload["configured"] = True
|
||
|
||
try:
|
||
quota_payload = call_cdn("DescribePurgeQuota", {})
|
||
payload["quota"]["url"] = normalize_quota_rows(list(quota_payload.get("UrlPurge") or []))
|
||
payload["quota"]["path"] = normalize_quota_rows(list(quota_payload.get("PathPurge") or []))
|
||
except Exception as exc: # noqa: BLE001
|
||
payload["warnings"].append(f"quota: {exc}")
|
||
|
||
try:
|
||
start_time, end_time = recent_task_query_window()
|
||
tasks_payload = call_cdn(
|
||
"DescribePurgeTasks",
|
||
{
|
||
"StartTime": start_time,
|
||
"EndTime": end_time,
|
||
"Offset": 0,
|
||
"Limit": max(int(CDN_PURGE_TASK_LIMIT or 20), 1),
|
||
},
|
||
)
|
||
payload["tasks"] = normalize_purge_task_rows(list(tasks_payload.get("PurgeLogs") or []))
|
||
payload["taskTotal"] = int(tasks_payload.get("TotalCount") or len(payload["tasks"]))
|
||
except Exception as exc: # noqa: BLE001
|
||
payload["warnings"].append(f"tasks: {exc}")
|
||
|
||
payload["ok"] = True
|
||
return payload
|
||
|
||
|
||
def submit_cdn_purge_payload(payload: dict[str, Any]) -> dict[str, Any]:
|
||
# 刷新任务提交前先做本地校验，把明显错误拦在页面内。
|
||
error = cdn_configuration_error()
|
||
if error:
|
||
raise RuntimeError(error)
|
||
|
||
purge_type = normalize_cdn_purge_type(payload.get("purgeType"))
|
||
items = normalize_cdn_purge_items(payload.get("items"), purge_type=purge_type)
|
||
area = normalize_cdn_area(payload.get("area"))
|
||
flush_type = normalize_cdn_flush_type(purge_type, payload.get("flushType"))
|
||
url_encode = bool(payload.get("urlEncode"))
|
||
|
||
request_payload: dict[str, Any] = {"UrlEncode": url_encode}
|
||
if area:
|
||
request_payload["Area"] = area
|
||
|
||
if purge_type == "url":
|
||
request_payload["Urls"] = items
|
||
response = call_cdn("PurgeUrlsCache", request_payload)
|
||
else:
|
||
request_payload["Paths"] = items
|
||
request_payload["FlushType"] = flush_type or "flush"
|
||
response = call_cdn("PurgePathCache", request_payload)
|
||
|
||
return {
|
||
"ok": True,
|
||
"taskId": str(response.get("TaskId") or "").strip(),
|
||
"requestId": str(response.get("RequestId") or "").strip(),
|
||
"purgeType": purge_type,
|
||
"area": area,
|
||
"flushType": flush_type,
|
||
"urlEncode": url_encode,
|
||
"itemCount": len(items),
|
||
"items": items,
|
||
"createdAt": utc_now(),
|
||
}
|