2645212d47
- Updated the API base URLs in app.js to point to local development server for testing. - Enhanced SVG icons with gradient fills for improved visual appeal across various components. - Refactored the avatar upload functionality in avatar-nickname.js and profile-edit.js to utilize a new upload utility for better code maintainability. - Improved the layout and styling of the my page and super article editor for a more user-friendly interface. This update aims to streamline development processes and enhance the overall user experience in the application.
134 lines
3.9 KiB
Python
134 lines
3.9 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
测试阿里云 OSS 是否可用（RAM AccessKey 能否访问指定 Bucket）。
|
||
|
||
依赖: pip install oss2
|
||
|
||
用法:
|
||
1) 将真实 Secret 放在环境变量（推荐）:
|
||
set OSS_ACCESS_KEY_SECRET=你的Secret
|
||
python test_oss_connect.py
|
||
|
||
2) 指定 JSON 配置文件（勿把含密钥的文件提交 Git）:
|
||
python test_oss_connect.py --config my-oss.json
|
||
|
||
默认会使用脚本内示例 JSON；若 accessKeySecret 为 **** 或空，必须从环境变量读取:
|
||
OSS_ACCESS_KEY_SECRET 或 ALIYUN_OSS_ACCESS_KEY_SECRET
|
||
可选覆盖:
|
||
OSS_ACCESS_KEY_ID 或 ALIYUN_OSS_ACCESS_KEY_ID
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import os
|
||
import sys
|
||
from typing import Any
|
||
|
||
|
||
DEFAULT_CONFIG: dict[str, Any] = {
|
||
"accessKeyId": "LTAI5t7ixwYZBqYc4bFpe5tc",
|
||
"accessKeySecret": "****",
|
||
"bucket": "kr-cypd",
|
||
"endpoint": "oss-cn-beijing.aliyuncs.com",
|
||
"region": "oss-cn-beijing",
|
||
}
|
||
|
||
|
||
def _is_placeholder_secret(s: str) -> bool:
|
||
s = (s or "").strip()
|
||
if not s or s == "****":
|
||
return True
|
||
return s.strip("*") == ""
|
||
|
||
|
||
def _norm_endpoint(ep: str) -> str:
|
||
ep = (ep or "").strip().rstrip("/")
|
||
if not ep:
|
||
return ""
|
||
if not ep.startswith("http://") and not ep.startswith("https://"):
|
||
ep = "https://" + ep
|
||
return ep
|
||
|
||
|
||
def main() -> int:
|
||
parser = argparse.ArgumentParser(description="测试 OSS Bucket 连通性")
|
||
parser.add_argument(
|
||
"--config",
|
||
"-c",
|
||
type=str,
|
||
default="",
|
||
help="JSON 路径，字段: accessKeyId, accessKeySecret, bucket, endpoint",
|
||
)
|
||
args = parser.parse_args()
|
||
|
||
if args.config:
|
||
with open(args.config, encoding="utf-8") as f:
|
||
cfg = json.load(f)
|
||
else:
|
||
cfg = dict(DEFAULT_CONFIG)
|
||
|
||
ak = (
|
||
os.environ.get("OSS_ACCESS_KEY_ID")
|
||
or os.environ.get("ALIYUN_OSS_ACCESS_KEY_ID")
|
||
or str(cfg.get("accessKeyId") or "")
|
||
).strip()
|
||
sk = (
|
||
os.environ.get("OSS_ACCESS_KEY_SECRET")
|
||
or os.environ.get("ALIYUN_OSS_ACCESS_KEY_SECRET")
|
||
or str(cfg.get("accessKeySecret") or "")
|
||
).strip()
|
||
|
||
if _is_placeholder_secret(sk):
|
||
print(
|
||
"错误: accessKeySecret 为占位符或未设置。\n"
|
||
"请设置环境变量 OSS_ACCESS_KEY_SECRET（或 ALIYUN_OSS_ACCESS_KEY_SECRET）后再运行。",
|
||
file=sys.stderr,
|
||
)
|
||
return 2
|
||
|
||
bucket_name = str(cfg.get("bucket") or "").strip()
|
||
endpoint = _norm_endpoint(str(cfg.get("endpoint") or ""))
|
||
if not ak or not bucket_name or not endpoint:
|
||
print("错误: 缺少 accessKeyId / bucket / endpoint", file=sys.stderr)
|
||
return 2
|
||
|
||
try:
|
||
import oss2
|
||
except ImportError:
|
||
print("请先安装: pip install oss2", file=sys.stderr)
|
||
return 1
|
||
|
||
print(f"Endpoint : {endpoint}")
|
||
print(f"Bucket : {bucket_name}")
|
||
print(f"AccessKey: {ak[:6]}…{ak[-4:] if len(ak) > 10 else ak}")
|
||
|
||
auth = oss2.Auth(ak, sk)
|
||
bucket = oss2.Bucket(auth, endpoint, bucket_name)
|
||
|
||
try:
|
||
# 列出最多 1 个对象，验证鉴权与 Bucket 是否存在、地域是否匹配
|
||
result = bucket.list_objects(max_keys=1)
|
||
print("结果: OSS 可用（list_objects 成功）。")
|
||
if result.object_list:
|
||
o = result.object_list[0]
|
||
print(f" 示例对象 key: {o.key}")
|
||
else:
|
||
print(" （Bucket 内暂无对象或列表为空）")
|
||
return 0
|
||
except oss2.exceptions.NoSuchBucket as e:
|
||
print(f"失败: Bucket 不存在或无权限 — {e}", file=sys.stderr)
|
||
return 3
|
||
except oss2.exceptions.AccessDenied as e:
|
||
print(f"失败: 拒绝访问（检查 RAM 策略是否包含 ListObjects/GetBucket）— {e}", file=sys.stderr)
|
||
return 3
|
||
except Exception as e:
|
||
print(f"失败: {type(e).__name__}: {e}", file=sys.stderr)
|
||
return 3
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|