#!/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())