姚章浩
1fa35e9870
按生产级备份组合落地：
- internal/releasesnapshot：新增门禁包，要求 root 下存在覆盖目标库、未过期
（默认 24h）、含 wiki_pages.jsonl 与 sha256 的快照；本地文件再校验哈希。
7 条单测（过期/换库/篡改/缺页/未接线/半成品）。
- 发布器：写目标库之前（Publish/Delete/Rollback 三条路）先过门禁；映射目标
（源库→目标库，共享真实库）或 policy=always 时必须有快照；门禁未接线
同样拒绝（fail-closed）。新错误码 release_target_snapshot_missing（不可重试）
+ 页面中文提示。6 条发布器单测。
- worker 配置新增 target_snapshot{root,max_age_hours,policy}，校验 policy/
max_age/绝对路径；compose.release-sim.yaml 只读挂载快照目录。
- scripts/target_kb_snapshot.sh：默认云内执行（与 DB 同 VPC，凭据取自 WeKnora
容器）+ 默认不导 chunks（44s 出 8 张表）；修复 \copy 不能跨行/不能带 WITH、
ES 计数改用 knowledge_base_id.keyword。
- ES：注册 kb_backup 仓库（path.repo），首个快照 36s/4.7GB/0 分片失败，
SLM kb-hourly(3d)+kb-daily(7d) 已建并执行成功。
- OSS：目标桶开启版本控制（此前未开启）。
- 文档 P6 增补硬门禁与云内执行；真实快照目录上门禁实测：两个绑定库通过、
未覆盖库拒绝、未接线拒绝。
80 lines
2.9 KiB
Python
80 lines
2.9 KiB
Python
#!/usr/bin/env python3
|
|
"""目标 OSS 桶的保护设置：读取状态，或开启版本控制 / 跨区域复制检查。
|
|
|
|
只处理目标环境的桶（secrets/release-sim 下的 key 文件），凭据不打印。
|
|
|
|
用法：
|
|
python3 scripts/oss_bucket_settings.py status # 版本控制 / 复制 / 生命周期
|
|
python3 scripts/oss_bucket_settings.py enable-versioning # 写操作：开启版本控制
|
|
"""
|
|
import base64
|
|
import hashlib
|
|
import hmac
|
|
import pathlib
|
|
import sys
|
|
import urllib.error
|
|
import urllib.request
|
|
from email.utils import formatdate
|
|
|
|
ROOT = pathlib.Path(__file__).resolve().parents[1]
|
|
SECRETS = ROOT / "secrets" / "release-sim"
|
|
BUCKET = "dobot-weknora-test"
|
|
ENDPOINT = "oss-cn-shenzhen.aliyuncs.com"
|
|
|
|
|
|
def read_secret(name):
|
|
return (SECRETS / name).read_text().strip()
|
|
|
|
|
|
def request(method, subresource, body=""):
|
|
access_key = read_secret("target_object_access_key")
|
|
secret_key = read_secret("target_object_secret_key")
|
|
date = formatdate(usegmt=True)
|
|
content_type = "application/xml" if body else ""
|
|
canonical = f"/{BUCKET}/?{subresource}"
|
|
string_to_sign = f"{method}\n\n{content_type}\n{date}\n{canonical}"
|
|
signature = base64.b64encode(
|
|
hmac.new(secret_key.encode(), string_to_sign.encode(), hashlib.sha1).digest()
|
|
).decode()
|
|
headers = {"Date": date, "Authorization": f"OSS {access_key}:{signature}"}
|
|
if content_type:
|
|
headers["Content-Type"] = content_type
|
|
http_request = urllib.request.Request(
|
|
f"https://{BUCKET}.{ENDPOINT}/?{subresource}",
|
|
data=body.encode() if body else None, headers=headers, method=method)
|
|
try:
|
|
with urllib.request.urlopen(http_request, timeout=20) as response:
|
|
return response.status, response.read().decode("utf-8", "replace")
|
|
except urllib.error.HTTPError as error:
|
|
return error.code, error.read().decode("utf-8", "replace")
|
|
|
|
|
|
def versioning_state():
|
|
status, body = request("GET", "versioning")
|
|
if status != 200:
|
|
return f"HTTP {status}"
|
|
return "Enabled" if "<Status>Enabled</Status>" in body else "Disabled（未开启）"
|
|
|
|
|
|
def main():
|
|
action = sys.argv[1] if len(sys.argv) > 1 else "status"
|
|
if action == "status":
|
|
print("versioning:", versioning_state())
|
|
for name in ("replication", "lifecycle"):
|
|
status, body = request("GET", name)
|
|
summary = "none" if status == 404 else " ".join(body.split())[:120]
|
|
print(f"{name}: HTTP {status} {summary}")
|
|
return
|
|
if action == "enable-versioning":
|
|
print("before:", versioning_state())
|
|
body = "<VersioningConfiguration><Status>Enabled</Status></VersioningConfiguration>"
|
|
status, response = request("PUT", "versioning", body)
|
|
print("PUT versioning:", f"HTTP {status}", " ".join(response.split())[:160])
|
|
print("after:", versioning_state())
|
|
return
|
|
print("unknown action:", action, file=sys.stderr)
|
|
sys.exit(2)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main() |