standard_provider.py

# coding=utf-8
from __future__ import unicode_literals, print_function
import re
from volcenginesdkcore.endpoint.endpoint_provider import EndpointProvider, ResolvedEndpoint


# -----------------------------
# Errors (模仿 volcengineerr.New)
# -----------------------------
class StandProviderError(Exception):
    def __init__(self, code, message):
        Exception.__init__(self, "%s: %s" % (code, message))
        self.code = code
        self.message = message


# -----------------------------
# 常量
# -----------------------------
DEFAULT_FORMAT = "{Service}{Region}.{SiteStack}.com"

# 站点栈
SITE_STACK_VOLC_IPV4 = "volcengineapi"
SITE_STACK_VOLC_DUAL_STACK = "volcengine-api"


# -----------------------------
# 数据模型（兼容 2.7/3.x 的朴素类）
# -----------------------------
class StandardEndpointResolverVariable(object):
    def __init__(self):
        self.service = ""
        self.region = ""
        self.site_stack = ""
        self.extension = {}

    def to_dict(self):
        return {
            "Service": self.service,
            "Region": self.region,
            "SiteStack": self.site_stack,
            "Extension": self.extension,
        }


class ServiceInfo(object):
    def __init__(self, service, is_global):
        self.service = service
        self.IsGlobal = is_global

ServiceInfos = {
    "vpc": ServiceInfo("vpc", False),
    "ecs": ServiceInfo("ecs", False),
    "billing": ServiceInfo("billing", True),
    "ark": ServiceInfo("ark", False),
    "iam": ServiceInfo("iam", True),
    "mcs": ServiceInfo("mcs", False),
    "rocketmq": ServiceInfo("rocketmq", False),
    "bytehouse": ServiceInfo("bytehouse", False),
    "dns": ServiceInfo("dns", True),
    "autoscaling": ServiceInfo("autoscaling", False),
    "spark": ServiceInfo("spark", False),
    "cloud_detect": ServiceInfo("cloud_detect", False),
    "filenas": ServiceInfo("filenas", False),
    "escloud": ServiceInfo("escloud", False),
    "flink": ServiceInfo("flink", False),
    "cp": ServiceInfo("cp", False),
    "vefaas": ServiceInfo("vefaas", False),
    "ml_platform": ServiceInfo("ml_platform", False),
    "edx": ServiceInfo("edx", True),
    "dcdn": ServiceInfo("dcdn", True),
    "cdn": ServiceInfo("cdn", True),
    "kafka": ServiceInfo("kafka", False),
    "certificate_service": ServiceInfo("certificate_service", True),
    "waf": ServiceInfo("waf", True),
    "rds_mssql": ServiceInfo("rds_mssql", False),
    "cloudtrail": ServiceInfo("cloudtrail", False),
    "vei_api": ServiceInfo("vei_api", True),
    "cen": ServiceInfo("cen", True),
    "rabbitmq": ServiceInfo("rabbitmq", False),
    "vmp": ServiceInfo("vmp", False),
    "volc_observe": ServiceInfo("volc_observe", False),
    "dataleap": ServiceInfo("dataleap", False),
    "fw_center": ServiceInfo("fw_center", True),
    "redis": ServiceInfo("redis", False),
    "mcdn": ServiceInfo("mcdn", True),
    "cloudidentity": ServiceInfo("cloudidentity", False),
    "vedbm": ServiceInfo("vedbm", False),
    "cv": ServiceInfo("cv", True),
    "translate": ServiceInfo("translate", True),
    "cloud_trail": ServiceInfo("cloud_trail", False),
    "bio": ServiceInfo("bio", False),
    "nta": ServiceInfo("nta", True),
    "elasticmapreduce": ServiceInfo("elasticmapreduce", False),
    "vepfs": ServiceInfo("vepfs", False),
    "seccenter": ServiceInfo("seccenter", True),
    "advdefence": ServiceInfo("advdefence", True),
    "tis": ServiceInfo("tis", True),
    "organization": ServiceInfo("organization", True),
    "vke": ServiceInfo("vke", False),
    "Redis": ServiceInfo("Redis", False),
    "privatelink": ServiceInfo("privatelink", False),
    "RocketMQ": ServiceInfo("RocketMQ", False),
    "Kafka": ServiceInfo("Kafka", False),
    "rds_mysql": ServiceInfo("rds_mysql", False),
    "rds_postgresql": ServiceInfo("rds_postgresql", False),
    "storage_ebs": ServiceInfo("storage_ebs", False),
    "clb": ServiceInfo("clb", False),
    "alb": ServiceInfo("alb", False),
    "FileNAS": ServiceInfo("FileNAS", False),
    "configcenter": ServiceInfo("configcenter", False),
    "cr": ServiceInfo("cr", False),
    "sts": ServiceInfo("sts", False),
    "mongodb": ServiceInfo("mongodb", False),
    "transitrouter": ServiceInfo("transitrouter", False),
    "Volc_Observe": ServiceInfo("Volc_Observe", False),
    "dms": ServiceInfo("dms", False),
    "auto_scaling": ServiceInfo("auto_scaling", False),
    "directconnect": ServiceInfo("directconnect", False),
    "kms": ServiceInfo("kms", False),
    "dbw": ServiceInfo("dbw", False),
    "dts": ServiceInfo("dts", False),
    "natgateway": ServiceInfo("natgateway", False),
    "tos": ServiceInfo("tos", False),
    "TLS": ServiceInfo("TLS", False),
    "vpn": ServiceInfo("vpn", False),
    "vod": ServiceInfo("vod", False),
    "quota": ServiceInfo("quota", True),
    "ecs_ops": ServiceInfo("ecs_ops", True),
    "as_ops": ServiceInfo("as_ops", True),
    "account_management": ServiceInfo("account_management", True),
    "account_management_byteplus": ServiceInfo("account_management_byteplus", True),
    "bandwidthquota": ServiceInfo("bandwidthquota", True),
    "psa_manager": ServiceInfo("psa_manager", True),
    "dc_controller": ServiceInfo("dc_controller", False),
    "eps_platform_trade": ServiceInfo("eps_platform_trade", False),
    "eps_platform_fund": ServiceInfo("eps_platform_fund", False),
    "commercialization": ServiceInfo("commercialization", True),
    "veecp_openapi": ServiceInfo("veecp_openapi", False),
    "orgnization": ServiceInfo("orgnization", True),
    "coze": ServiceInfo("coze", True),
    "sec_agent": ServiceInfo("sec_agent", True),
    "sec_intelligent_dev": ServiceInfo("sec_intelligent_dev", True),
    "vegame": ServiceInfo("vegame", False),
    "acep": ServiceInfo("acep", True),
    "private_zone": ServiceInfo("private_zone", True),
    "sqs": ServiceInfo("sqs", False),
    "resourcecenter": ServiceInfo("resourcecenter", True),
    "aiotvideo": ServiceInfo("aiotvideo", True),
    "apig": ServiceInfo("apig", False),
    "bmq": ServiceInfo("bmq", False),
    "bytehouse_ce": ServiceInfo("bytehouse_ce", False),
    "cloudmonitor": ServiceInfo("cloudmonitor", False),
    "emr": ServiceInfo("emr", False),
    "ga": ServiceInfo("ga", True),
    "graph": ServiceInfo("graph", False),
    "gtm": ServiceInfo("gtm", True),
    "hbase": ServiceInfo("hbase", False),
    "metakms": ServiceInfo("metakms", False),
    "na": ServiceInfo("na", True),
    "resource_share": ServiceInfo("resource_share", True),
    "speech_saas_prod": ServiceInfo("speech_saas_prod", True),
    "tag": ServiceInfo("tag", True),
    "vefaas_dev": ServiceInfo("vefaas_dev", False),
    "vms": ServiceInfo("vms", False),
    "eco_partner": ServiceInfo("eco_partner", True),
    "smc": ServiceInfo("smc", True),
    "livesaas": ServiceInfo("livesaas", True),
}


# -----------------------------
# Region 校验
# -----------------------------
class RegionChecker(object):
    def __init__(self, white_regions, pattern):
        self.white_regions = white_regions or {}
        self.pattern = pattern

    def validate(self, region):
        if self.white_regions and region in self.white_regions:
            return True
        if self.pattern:
            return bool(self.pattern.match(region))
        return False


region_matcher = RegionChecker(
    {
        "ap-singapore-1": {},
        "ap-southeast-1": {},
        "ap-southeast-2": {},
        "ap-southeast-3": {},
        "byteplus-global": {},
        "cn-beijing": {},
        "cn-beijing-autodriving": {},
        "cn-beijing-selfdrive": {},
        "cn-beijing2": {},
        "cn-beijing300": {},
        "cn-changsha-sdv": {},
        "cn-chengdu": {},
        "cn-chengdu-sdv": {},
        "cn-chongqing-sdv": {},
        "cn-datong": {},
        "cn-east-1-dedicated": {},
        "cn-gaofang-bj": {},
        "cn-gaofang-gz1": {},
        "cn-gaofang-nt1": {},
        "cn-gaofang-nt2": {},
        "cn-gaofang-nt3": {},
        "cn-gaofang-nt4": {},
        "cn-gaofang-nt5": {},
        "cn-guangzhou": {},
        "cn-guilin-boe": {},
        "cn-hangzhou": {},
        "cn-hjxj": {},
        "cn-hjzg": {},
        "cn-hlbx": {},
        "cn-hlxj": {},
        "cn-hlzg": {},
        "cn-hongkong": {},
        "cn-hongkong-pop": {},
        "cn-lfbx": {},
        "cn-lfxj": {},
        "cn-lfzg": {},
        "cn-macau-pop-sdv": {},
        "cn-mainland": {},
        "cn-nanjing-bbit": {},
        "cn-ningbo-sdv": {},
        "cn-north-1": {},
        "cn-north-1-dedicated": {},
        "cn-north-boe": {},
        "cn-shanghai": {},
        "cn-shanghai-autodriving": {},
        "cn-taiwan-boe": {},
        "cn-wuhan": {},
        "cn-wulanchabu": {},
        "cn-xian-boe-sdv": {},
        "overseas-1": {},
        "rec-cn": {},
        "rec-sg": {},
    },
    re.compile(
        r"^(?:[a-z]{2}-[a-z]+(?:-[a-z]+)?|(?:cn|ap|eu|na|sa|me|af)-[a-z]+-\d+(?:-(?:finance|exclusive|local|inner))?)$"
    ),
)


# -----------------------------
# 工具函数
# -----------------------------
def standardize_domain_service_code(service_code):
    """lower + '_' -> '-'"""
    return service_code.lower().replace("_", "-")


# -----------------------------
# 解析器
# -----------------------------
class StandardEndpointResolver(EndpointProvider):
    def __init__(self, fmt=None, site_stack=None, extension=None, custom_services=None):
        self.fmt = fmt or DEFAULT_FORMAT
        self.variables = StandardEndpointResolverVariable()
        self.variables.site_stack = site_stack or SITE_STACK_VOLC_IPV4
        self.variables.extension = extension or {}
        self.custom_services = custom_services or {}

    def endpoint_for(self, service, region, custom_bootstrap_region=None, use_dual_stack=None, **kwargs):
        # 1) Region 校验
        if not region_matcher.validate(region):
            raise StandProviderError(
                "InvalidRegion",
                "invalid region %s for standard endpoint resolver, "
                "please upgrade the sdk endpoint resolver to the latest version" % region
            )

        # 2) 默认模板
        fmt = self.fmt or DEFAULT_FORMAT

        # 3) 变量准备
        self.variables.service = standardize_domain_service_code(service)

        svc_info = ServiceInfos.get(service)
        if not svc_info:
            svc_info = self.custom_services.get(service)
            if not svc_info:
                raise StandProviderError(
                    "ServiceNotFound",
                    "service %s not found in ServiceInfos  or customServices, "
                    "please upgrade the sdk endpoint resolver to the latest version" % service
                )

        # 非全局服务拼接 ".{region}"，全局则为空
        if not svc_info.IsGlobal:
            self.variables.region = "." + region
        else:
            self.variables.region = ""

        # 4) IP 版本 -> SiteStack
        if use_dual_stack:
            self.variables.site_stack = SITE_STACK_VOLC_DUAL_STACK
        else:
            self.variables.site_stack = SITE_STACK_VOLC_IPV4

        # 5) 渲染
        try:
            rendered = fmt.format(**self.variables.to_dict())
        except KeyError as err:
            raise StandProviderError(
                "TemplateExecuteError",
                "failed to execute template for format %s, missing variable %s" % (fmt, err)
            )
        except Exception as err:
            raise StandProviderError(
                "TemplateExecuteError",
                "failed to execute template for format %s, variable %r: %s" % (fmt, self.variables.to_dict(), err)
            )

        return ResolvedEndpoint(rendered)


# -----------------------------
if __name__ == "__main__":
    r = StandardEndpointResolver()
    try:
        ep = r.endpoint_for(
            service="vpc",
            region="ap-southeast-1",
            use_dual_stack=False,
        )
        print(ep.host)  # 例如：vpc.ap-southeast-1.volcengine-api.com
    except Exception as e:
        e.print()