Byted Volcengine Vmp

# Package marker for local imports.

from __future__ import annotations

import sys
import types
from pathlib import Path


PACKAGE_NAME = "_byted_volcengine_vmp_scripts"


def ensure_package() -> str:
    """
    为“直接执行脚本文件”场景提供稳定导入方式（不修改 sys.path）。

    用法：
    - 在 `python scripts/query_metrics.py ...` 这种运行方式下，`__package__` 为空，
      相对导入不可用；这里通过注入一个“临时包”并设置 `__path__` 指向当前目录，
      让 `from _byted_volcengine_vmp_scripts.xxx import ...` 能工作。
    """
    if PACKAGE_NAME in sys.modules:
        return PACKAGE_NAME

    pkg = types.ModuleType(PACKAGE_NAME)
    pkg.__path__ = [str(Path(__file__).resolve().parent)]
    sys.modules[PACKAGE_NAME] = pkg
    return PACKAGE_NAME

#!/usr/bin/env python3
from __future__ import annotations

import argparse
import json
import os
from typing import Callable

if __package__ in (None, ""):
    from _bootstrap import ensure_package

    ensure_package()
    from _byted_volcengine_vmp_scripts.vmp_client import VMPClient  # type: ignore
else:
    from .vmp_client import VMPClient


Handler = Callable[[VMPClient, argparse.Namespace], dict]


def _build_common_parser(description: str) -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description=description)
    parser.add_argument("--ak", help="火山引擎 Access Key")
    parser.add_argument("--sk", help="火山引擎 Secret Key")
    parser.add_argument("--region", default=os.getenv("VOLCENGINE_REGION", "cn-beijing"), help="地域")
    parser.add_argument("--endpoint", help="自定义 VMP Endpoint")
    parser.add_argument("--session-token", help="临时凭证 Session Token")
    parser.add_argument("--json", action="store_true", help="以 JSON 格式输出")
    return parser


def build_list_workspaces_parser() -> argparse.ArgumentParser:
    return _build_common_parser("查询 VMP 工作区列表")


def build_metric_names_parser() -> argparse.ArgumentParser:
    parser = _build_common_parser("查询指标名称列表")
    parser.add_argument("--workspace-id", "-w", required=True, help="工作区 ID")
    parser.add_argument("--match", "-m", help="PromQL 匹配条件")
    return parser


def build_metric_labels_parser() -> argparse.ArgumentParser:
    parser = _build_common_parser("查询指标标签列表")
    parser.add_argument("--workspace-id", "-w", required=True, help="工作区 ID")
    parser.add_argument("--metric-name", "-m", required=True, help="指标名称")
    return parser


def build_query_metrics_parser() -> argparse.ArgumentParser:
    parser = _build_common_parser("即时查询 VMP Metrics")
    parser.add_argument("--workspace-id", "-w", required=True, help="工作区 ID")
    parser.add_argument("--query", "-q", required=True, help="PromQL 查询语句")
    parser.add_argument("--time", "-t", help="查询时间，支持 RFC3339 或 Unix 时间戳")
    return parser


def build_query_range_metrics_parser() -> argparse.ArgumentParser:
    parser = _build_common_parser("范围查询 VMP Metrics")
    parser.add_argument("--workspace-id", "-w", required=True, help="工作区 ID")
    parser.add_argument("--query", "-q", required=True, help="PromQL 查询语句")
    parser.add_argument("--start", "-s", required=True, help="开始时间")
    parser.add_argument("--end", "-e", required=True, help="结束时间")
    parser.add_argument("--step", help="查询步长")
    return parser


def _create_client(args: argparse.Namespace) -> VMPClient:
    return VMPClient(
        ak=args.ak,
        sk=args.sk,
        region=args.region,
        endpoint=args.endpoint,
        session_token=args.session_token,
    )


def _print_result(result: dict, json_output: bool) -> None:
    text = json.dumps(result, indent=2, ensure_ascii=False)
    if json_output:
        print(text)
        return
    print(text)


def run_with_client(parser: argparse.ArgumentParser, handler: Handler) -> None:
    args = parser.parse_args()
    try:
        client = _create_client(args)
        result = handler(client, args)
        _print_result(result, args.json)
    except Exception as exc:
        error_payload = {"error": str(exc), "type": exc.__class__.__name__}
        print(json.dumps(error_payload, indent=2, ensure_ascii=False))
        raise SystemExit(2)

import os
from dataclasses import dataclass

import volcenginesdkcore
from volcenginesdkcore.interceptor import RuntimeOption

ENV_VOLCENGINE_ENDPOINT = "VOLCENGINE_ENDPOINT"
ENV_VOLCENGINE_REGION = "VOLCENGINE_REGION"
ENV_VOLCENGINE_ACCESS_KEY = "VOLCENGINE_ACCESS_KEY"
ENV_VOLCENGINE_SECRET_KEY = "VOLCENGINE_SECRET_KEY"
ENV_VOLCENGINE_SESSION_TOKEN = "VOLCENGINE_SESSION_TOKEN"

# 兼容常见命名（文档/历史脚本里经常使用）
ENV_VOLCENGINE_AK = "VOLCENGINE_AK"
ENV_VOLCENGINE_SK = "VOLCENGINE_SK"
ENV_VOLC_ACCESSKEY = "VOLC_ACCESSKEY"
ENV_VOLC_SECRETKEY = "VOLC_SECRETKEY"

ENV_MCP_SERVER_NAME = "MCP_SERVER_NAME"
ENV_MCP_SERVER_MODE = "MCP_SERVER_MODE"
ENV_MCP_SERVER_HOST = "MCP_SERVER_HOST"
ENV_MCP_SERVER_PORT = "MCP_SERVER_PORT"

ENV_POOL_CONCURRENCY = "POOL_CONCURRENCY"

@dataclass
class VMPConfig:
    """Configuration for VMP MCP Server."""
    volcengine_endpoint: str
    volcengine_region: str
    volcengine_ak: str
    volcengine_sk: str
    session_token: str

    pool_concurrency: int

    def is_valid(self) -> bool:
        """Check if the configuration is valid."""
        # session_token 仅用于临时凭证场景，AK/SK 仍是必需项
        return bool(self.volcengine_ak and self.volcengine_sk)
    
    def to_volc_configuration(self) -> volcenginesdkcore.Configuration:
        """Convert to volcengine configuration."""
        volcConf = volcenginesdkcore.Configuration()
        volcConf.host = self.volcengine_endpoint
        volcConf.region = self.volcengine_region
        volcConf.ak = self.volcengine_ak
        volcConf.sk = self.volcengine_sk
        volcConf.session_token = self.session_token
        volcConf.connection_pool_maxsize = self.pool_concurrency
        return volcConf

    def to_runtime_option(self) -> RuntimeOption:
        """Convert to RuntimeOption."""
        option = RuntimeOption(
            True,
            ak=self.volcengine_ak,
            sk=self.volcengine_sk,
            session_token=self.session_token,
            region=self.volcengine_region,
        )
        return option

def load_env_config() -> VMPConfig:
    """Load configuration from environment variables."""
    cpu = os.cpu_count() or 1
    return VMPConfig(
        volcengine_endpoint=os.getenv(ENV_VOLCENGINE_ENDPOINT, ""),
        volcengine_region=os.getenv(ENV_VOLCENGINE_REGION, "cn-beijing"),
        # 兼容多种变量名，优先使用更明确的标准命名
        volcengine_ak=os.getenv(ENV_VOLCENGINE_ACCESS_KEY, "")
        or os.getenv(ENV_VOLCENGINE_AK, "")
        or os.getenv(ENV_VOLC_ACCESSKEY, ""),
        volcengine_sk=os.getenv(ENV_VOLCENGINE_SECRET_KEY, "")
        or os.getenv(ENV_VOLCENGINE_SK, "")
        or os.getenv(ENV_VOLC_SECRETKEY, ""),
        session_token=os.getenv(ENV_VOLCENGINE_SESSION_TOKEN, ""),
        pool_concurrency=int(os.getenv(ENV_POOL_CONCURRENCY, "0")) or cpu * 32 + 1,
    )

#!/usr/bin/env python3
from __future__ import annotations

if __package__ in (None, ""):
    from _bootstrap import ensure_package

    ensure_package()
    from _byted_volcengine_vmp_scripts.cli_common import build_metric_labels_parser, run_with_client  # type: ignore
else:
    from .cli_common import build_metric_labels_parser, run_with_client


def _handle(client, args):
    return client.query_metric_labels(args.workspace_id, args.metric_name)


def main() -> None:
    run_with_client(build_metric_labels_parser(), _handle)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
from __future__ import annotations

if __package__ in (None, ""):
    from _bootstrap import ensure_package

    ensure_package()
    from _byted_volcengine_vmp_scripts.cli_common import build_metric_names_parser, run_with_client  # type: ignore
else:
    from .cli_common import build_metric_names_parser, run_with_client


def _handle(client, args):
    return client.query_metric_names(args.workspace_id, args.match)


def main() -> None:
    run_with_client(build_metric_names_parser(), _handle)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
from __future__ import annotations

if __package__ in (None, ""):
    from _bootstrap import ensure_package

    ensure_package()
    from _byted_volcengine_vmp_scripts.cli_common import build_list_workspaces_parser, run_with_client  # type: ignore
else:
    from .cli_common import build_list_workspaces_parser, run_with_client


def _handle(client, _args):
    return client.list_workspaces()


def main() -> None:
    run_with_client(build_list_workspaces_parser(), _handle)


if __name__ == "__main__":
    main()

from dataclasses import dataclass
from typing import List, Optional

from volcenginesdkcore.interceptor import RuntimeOption


@dataclass
class RequestBase:
    def __post_init__(self):
        self.build_swagger_types()
    
    def build_swagger_types(self):
        # 兼容 volcengine SDK 部分接口的参数描述结构（非强依赖）。
        self.swagger_types = {
            attr_name: type(attr_value).__name__
            for attr_name, attr_value in vars(self).items()
            if not attr_name.startswith("_") and not callable(attr_value)
        }
        self.attribute_map = {
            attr_name: attr_name
            for attr_name, attr_value in vars(self).items()
            if not attr_name.startswith("_") and not callable(attr_value)
        }

    def with_runtime_option(self, runtime_options: RuntimeOption):
        self._configuration = runtime_options
        return self

@dataclass
class ListWorkspacesRequest(RequestBase):
    PageNumber: int = 1
    PageSize: int = 100
    ShowAggregateQueryWorkspaces: bool = True

@dataclass
class QueryInstantMetricsRequest(RequestBase):
    query: str
    time: Optional[str] = None

@dataclass
class QueryRangeMetricsRequest(RequestBase):
    workspace: str
    query: str
    start: str
    end: str
    step: Optional[str] = None
       
@dataclass
class GetLabelValuesRequest(RequestBase):
    workspace: str
    label: str
    start: Optional[str] = None
    end: Optional[str] = None
    matches: Optional[List[str]] = None
    limit: Optional[int] = None

@dataclass
class GetLabelsRequest(RequestBase):
    workspace: str
    start: Optional[str] = None
    end: Optional[str] = None
    matches: Optional[List[str]] = None
    limit: Optional[int] = None

@dataclass
class GetSeriesRequest(RequestBase):
    workspace: str
    matches: Optional[List[str]]
    start: Optional[str] = None
    end: Optional[str] = None
    limit: Optional[int] = None

@dataclass
class Credentials:
    access_key_id: str
    secret_access_key: str
    region: str
    service: str
    session_token: Optional[str] = None

#!/usr/bin/env python3
from __future__ import annotations

if __package__ in (None, ""):
    from _bootstrap import ensure_package

    ensure_package()
    from _byted_volcengine_vmp_scripts.cli_common import build_query_metrics_parser, run_with_client  # type: ignore
else:
    from .cli_common import build_query_metrics_parser, run_with_client


def _handle(client, args):
    return client.query_instant_metrics(args.workspace_id, args.query, args.time)


def main() -> None:
    run_with_client(build_query_metrics_parser(), _handle)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
from __future__ import annotations

if __package__ in (None, ""):
    from _bootstrap import ensure_package

    ensure_package()
    from _byted_volcengine_vmp_scripts.cli_common import build_query_range_metrics_parser, run_with_client  # type: ignore
else:
    from .cli_common import build_query_range_metrics_parser, run_with_client


def _handle(client, args):
    return client.query_range_metrics(args.workspace_id, args.query, args.start, args.end, args.step)


def main() -> None:
    run_with_client(build_query_range_metrics_parser(), _handle)


if __name__ == "__main__":
    main()

# coding:utf-8
"""
Copyright (year) Beijing Volcano Engine Technology Ltd.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

     http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""
import json
import datetime
import hashlib
import hmac
import os
from urllib.parse import quote, urlencode
import dataclasses
from typing import Any, Dict, Union, List

import requests

def norm_query(params):
    query = ""
    for key in sorted(params.keys()):
        if type(params[key]) == list:
            for k in params[key]:
                query = (
                        query + quote(key, safe="-_.~") + "=" + quote(k, safe="-_.~") + "&"
                )
        else:
            query = (query + quote(key, safe="-_.~") + "=" + quote(params[key], safe="-_.~") + "&")
    query = query[:-1]
    return query.replace("+", "%20")


def hmac_sha256(key: bytes, content: str):
    return hmac.new(key, content.encode("utf-8"), hashlib.sha256).digest()


def hash_sha256(content: str):
    return hashlib.sha256(content.encode("utf-8")).hexdigest()

# ref: https://www.volcengine.com/docs/6369/67269
def generate_signature(request_param, credential):
    """
    生成请求的签名和必要的头信息
    
    Args:
        request_param: 请求参数字典，包含body、host、path、method、content_type、query等
        credential: 凭证信息，包含access_key_id、secret_access_key、region、service等
        
    Returns:
        dict: 包含签名信息的头信息字典
    """
    x_date = utc_now().strftime("%Y%m%dT%H%M%SZ")
    short_x_date = x_date[:8]
    x_content_sha256 = hash_sha256(request_param["body"])
    
    # init sign result
    sign_result = {
        "Host": request_param["host"],
        "X-Content-Sha256": x_content_sha256,
        "X-Date": x_date,
        "Content-Type": request_param["content_type"],
    }

    # build signed headers string
    signed_headers_str = ";".join(
        ["content-type", "host", "x-content-sha256", "x-date"]
    )
    
    # build canonical request string
    canonical_request_str = "\n".join([
        request_param["method"].upper(),
        request_param["path"],
        norm_query(request_param["query"]),
        "\n".join([
            "content-type:" + request_param["content_type"],
            "host:" + request_param["host"],
            "x-content-sha256:" + x_content_sha256,
            "x-date:" + x_date,
        ]),
        "",
        signed_headers_str,
        x_content_sha256,
    ])
    
    # calculate hashed canonical request
    hashed_canonical_request = hash_sha256(canonical_request_str)

    # build credential scope and string to sign
    credential_scope = "/".join([short_x_date, credential.region, credential.service, "request"])
    string_to_sign = "\n".join(["HMAC-SHA256", x_date, credential_scope, hashed_canonical_request])

    # calculate HMAC key chain
    k_date = hmac_sha256(credential.secret_access_key.encode("utf-8"), short_x_date)
    k_region = hmac_sha256(k_date, credential.region)
    k_service = hmac_sha256(k_region, credential.service)
    k_signing = hmac_sha256(k_service, "request")
    
    # calculate final signature
    signature = hmac_sha256(k_signing, string_to_sign).hex()

    # build Authorization header
    sign_result["Authorization"] = "HMAC-SHA256 Credential={}, SignedHeaders={}, Signature={}".format(
        credential.access_key_id + "/" + credential_scope,
        signed_headers_str,
        signature,
    )
    
    # if session token is not empty, add it to sign result
    if not is_empty_value(credential.session_token):
        sign_result["x-security-token"] = credential.session_token
    
    return sign_result

def is_empty_value(value: Any) -> bool:
    if value is None:
        return True
    if isinstance(value, (str, list, dict)) and len(value) == 0:
        return True
    return False

def to_serializable_dict(obj: Any) -> Union[Dict[str, Any], List[Any], Any]:
    if dataclasses.is_dataclass(obj):
        result_dict = {}
        for field in dataclasses.fields(obj):
            value = getattr(obj, field.name)
            if not is_empty_value(value):
                serialized_value = to_serializable_dict(value)
                if not is_empty_value(serialized_value):
                    result_dict[field.name] = serialized_value
        return result_dict
    
    if isinstance(obj, dict):
        result_dict = {}
        for k, v in obj.items():
            if not is_empty_value(v):
                serialized_value = to_serializable_dict(v)
                if not is_empty_value(serialized_value):
                    result_dict[k] = serialized_value
        return result_dict

    if isinstance(obj, list):
        result_list = []
        for item in obj:
            if not is_empty_value(item):
                serialized_item = to_serializable_dict(item)
                if not is_empty_value(serialized_item):
                    result_list.append(serialized_item)
        return result_list

    return obj

def utc_now():
    try:
        from datetime import timezone
        return datetime.datetime.now(timezone.utc)
    except ImportError:
        class UTC(datetime.tzinfo):
            def utcoffset(self, dt):
                return datetime.timedelta(0)
            def tzname(self, dt):
                return "UTC"
            def dst(self, dt):
                return datetime.timedelta(0)
        return datetime.datetime.now(UTC())

def get_content_type(headers):
    lower_headers = {k.lower(): v for k, v in headers.items()}
    return lower_headers.get('content-type')

def sign_and_request(info, credential, query, header, host, body=None):
    content_type = get_content_type(header)
    
    if body is not None:
        body_dict = to_serializable_dict(body)
        
        if content_type == "application/x-www-form-urlencoded":
            body_str = urlencode(body_dict, doseq=True)
        else:
            body_str = json.dumps(body_dict)
    else:
        body_str = ""

    request_param = {
        "body": body_str,
        "host": host,
        "path": "/",
        "method": info.method,
        "content_type": content_type,
        "query": {"Action": info.action, "Version": info.version, **query},
    }
   
    sign_result = generate_signature(request_param, credential)
    
    header = {**header, **sign_result}
    # 默认不输出签名/请求体，避免污染脚本输出；需要排障时再打开
    if os.getenv("VMP_DEBUG", "") in ("1", "true", "TRUE"):
        print(header)
        print(request_param["body"])
        print(request_param["query"])
    
    timeout = (10, 30)
    
    try:
        r = requests.request(
            method=info.method,
            url="https://{}{}".format(request_param["host"], request_param["path"]),
            headers=header,
            params=request_param["query"],
            data=request_param["body"],
            timeout=timeout  
        )
        
        try:
            return r.json()
        except json.JSONDecodeError:
            return {
                "error": "Invalid JSON response",
                "status_code": r.status_code,
                "response_text": r.text
            }
    except requests.exceptions.Timeout:
        return {
            "error": "Request timed out",
            "timeout": timeout
        }
    except requests.exceptions.ConnectionError:
        return {
            "error": "Connection error",
            "host": request_param["host"]
        }
    except requests.exceptions.RequestException as e:
        return {
            "error": "Request failed",
            "message": str(e)
        }

import asyncio
from multiprocessing.pool import AsyncResult

import six
import volcenginesdkcore


class UniversalApi(volcenginesdkcore.UniversalApi):
    def __init__(self, api_client=None):
        super().__init__(api_client)

    async def do_call_async(self, info, body, **kwargs):  # noqa: E501
        kwargs['async_req'] = True
        result = self.do_call(info, body, **kwargs)
        return await wait_for_async_result(result)

    def do_call_with_http_info(self, info, body, **kwargs):  # noqa: E501
        all_params = ['body', 'query', 'async_req', '_return_http_data_only', '_preload_content',
                      '_request_timeout']  # noqa: E501

        params = locals()
        for key, val in six.iteritems(params['kwargs']):
            if key not in all_params:
                raise TypeError(
                    "Got an unexpected keyword argument '%s'"
                    " to method do_call" % key
                )
            params[key] = val
        del params['kwargs']

        # verify the required parameter 'body' is set
        if self.api_client.client_side_validation and ('body' not in params or
                                                       params['body'] is None):  # noqa: E501
            raise ValueError(
                "Missing the required parameter `body` when calling `do_call`")  # noqa: E501

        # if type(params['body']) is not dict:
        #    raise TypeError(
        #        "The required parameter `body` must be dict")  # noqa: E501

        collection_formats = {}

        path_params = {}

        query_params = []

        header_params = {}

        form_params = []
        local_var_files = {}

        body_params = None
        if 'body' in params:
            body_params = params['body']

        # HTTP header `Accept`
        header_params['Accept'] = self.api_client.select_header_accept(
            ['application/json'])  # noqa: E501

        if info.content_type is not None:
            # HTTP header `Content-Type`
            header_params['Content-Type'] = self.api_client.select_header_content_type(  # noqa: E501
                [info.content_type])  # noqa: E501

        if info.method.lower() == "get":
            query_params = list(body.items())

        # Authentication setting
        auth_settings = ['volcengineSign']  # noqa: E501

        path = '/' + info.action + '/' + info.version + '/' + info.service + '/' + info.method.lower() + '/'
        return self.api_client.call_api(
            path, info.method.upper(),
            path_params,
            query_params,
            header_params,
            body=body_params,
            post_params=form_params,
            files=local_var_files,
            response_type=object,  # noqa: E501
            auth_settings=auth_settings,
            async_req=params.get('async_req'),
            _return_http_data_only=params.get('_return_http_data_only'),
            _preload_content=params.get('_preload_content', True),
            _request_timeout=params.get('_request_timeout'),
            collection_formats=collection_formats)

async def wait_for_async_result(async_result: AsyncResult):
    while not async_result.ready():
        # check if the async result is ready every 100ms
        await asyncio.sleep(0.1)
    return async_result.get()

#!/usr/bin/env python3
"""
火山引擎托管 Prometheus (VMP) API 客户端
用于查询 Prometheus 指标数据
"""

from __future__ import print_function
import os
from typing import Optional

if __package__ in (None, ""):
    from _bootstrap import ensure_package

    ensure_package()
    from _byted_volcengine_vmp_scripts import config, models, sign  # type: ignore
else:
    from . import config, models, sign
import volcenginesdkcore


class VMPClient:
    """VMP API 客户端"""
    
    def __init__(
        self,
        ak: str = None,
        sk: str = None,
        region: str = "cn-beijing",
        endpoint: str = None,
        session_token: str = None,
    ):
        """
        初始化 VMP 客户端
        
        Args:
            ak: Access Key（可选，优先从环境变量或 .env 文件加载）
            sk: Secret Key（可选，优先从环境变量或 .env 文件加载）
            region: 区域（默认 cn-beijing）
            endpoint: 自定义域名（可选）
            session_token: 临时凭证 Token（可选）
        """
        self.conf = self._load_config(ak, sk, region, endpoint, session_token)
        self.service_code = "vmp"
        self.service_version = "2021-03-03"
        self.content_type_json = "application/json"
        self.content_type_form = "application/x-www-form-urlencoded"
    
    def _load_config(
        self,
        ak: str = None,
        sk: str = None,
        region: str = "cn-beijing",
        endpoint: str = None,
        session_token: str = None,
    ) -> config.VMPConfig:
        """加载配置"""
        # 先尝试从环境变量加载
        conf = config.load_env_config()
        
        # 如果提供了参数，覆盖环境变量
        if ak:
            conf.volcengine_ak = ak
        if sk:
            conf.volcengine_sk = sk
        if region:
            conf.volcengine_region = region
            if not conf.volcengine_endpoint:
                conf.volcengine_endpoint = f"vmp.{region}.volcengineapi.com"
        if endpoint:
            conf.volcengine_endpoint = endpoint
        if session_token:
            conf.session_token = session_token
        
        # 尝试从 .env 文件加载
        env_file = os.path.expanduser("~/.openclaw/workspace/.env")
        if os.path.exists(env_file):
            try:
                with open(env_file, "r", encoding="utf-8") as f:
                    for line in f:
                        line = line.strip()
                        if line and not line.startswith("#") and "=" in line:
                            key, value = line.split("=", 1)
                            clean_value = value.strip().strip("\"'")
                            if key in ("VOLCENGINE_ACCESS_KEY", "VOLCENGINE_AK", "VOLC_ACCESSKEY") and not conf.volcengine_ak:
                                conf.volcengine_ak = clean_value
                            elif key in ("VOLCENGINE_SECRET_KEY", "VOLCENGINE_SK", "VOLC_SECRETKEY") and not conf.volcengine_sk:
                                conf.volcengine_sk = clean_value
                            elif key == "VOLCENGINE_REGION" and region == "cn-beijing":
                                conf.volcengine_region = clean_value or conf.volcengine_region
                            elif key == "VOLCENGINE_ENDPOINT" and not endpoint:
                                conf.volcengine_endpoint = clean_value
                            elif key == "VOLCENGINE_SESSION_TOKEN" and not session_token:
                                conf.session_token = clean_value
            except Exception as e:
                print(f"警告: 无法读取 .env 文件: {e}")

        if not conf.volcengine_endpoint:
            conf.volcengine_endpoint = f"vmp.{conf.volcengine_region}.volcengineapi.com"
        
        if not conf.is_valid():
            raise ValueError("未配置有效的 Access Key 和 Secret Key")
        
        return conf
    
    def query_instant_metrics(self, workspace_id: str, query: str, time: Optional[str] = None) -> dict:
        """
        即时查询 Metrics
        
        Args:
            workspace_id: 工作区 ID
            query: PromQL 查询语句
            time: 查询时间（可选，RFC3339 或 Unix 时间戳）
        
        Returns:
            查询结果
        """
        credentials = models.Credentials(
            access_key_id=self.conf.volcengine_ak,
            secret_access_key=self.conf.volcengine_sk,
            session_token=self.conf.session_token,
            region=self.conf.volcengine_region,
            service=self.service_code,
        )
        
        resp = sign.sign_and_request(
            volcenginesdkcore.UniversalInfo(
                method="POST",
                service=self.service_code,
                version=self.service_version,
                action="QueryMetrics",
                content_type=self.content_type_form,
            ),
            credentials,
            host=self.conf.volcengine_endpoint,
            query={
                'workspace': workspace_id,
            },
            header={
                "Content-Type": self.content_type_form,
            },
            body=models.QueryInstantMetricsRequest(
                query=query,
                time=time,
            ),
        )
        return resp
    
    def query_range_metrics(self, workspace_id: str, query: str, 
                          start: str, end: str, step: Optional[str] = None) -> dict:
        """
        范围查询 Metrics
        
        Args:
            workspace_id: 工作区 ID
            query: PromQL 查询语句
            start: 起始时间（RFC3339 或 Unix 时间戳）
            end: 结束时间（RFC3339 或 Unix 时间戳）
            step: 查询步长（可选，duration 格式，如 '15s'、'1m'、'1h'）
        
        Returns:
            查询结果
        """
        # 自动计算 step
        if step is None:
            step = self._calculate_default_step(start, end)
            print(f"自动计算 step: {step}")
        
        credentials = models.Credentials(
            access_key_id=self.conf.volcengine_ak,
            secret_access_key=self.conf.volcengine_sk,
            session_token=self.conf.session_token,
            region=self.conf.volcengine_region,
            service=self.service_code,
        )
        
        resp = sign.sign_and_request(
            volcenginesdkcore.UniversalInfo(
                method="POST",
                service=self.service_code,
                version=self.service_version,
                action="QueryMetricsRange",
                content_type=self.content_type_json,
            ),
            credentials,
            host=self.conf.volcengine_endpoint,
            query={
                'workspace': workspace_id,
            },
            header={
                # 范围查询 body 为 JSON
                "Content-Type": self.content_type_json,
            },
            body=models.QueryRangeMetricsRequest(
                workspace=workspace_id,
                query=query,
                start=start,
                end=end,
                step=step,
            ),
        )
        return resp
    
    def query_metric_names(self, workspace_id: str, match: Optional[str] = None) -> dict:
        """
        查询指标名称列表
        
        Args:
            workspace_id: 工作区 ID
            match: 匹配条件（可选，如 '{job=~"kubelet"}'）
        
        Returns:
            指标名称列表
        """
        credentials = models.Credentials(
            access_key_id=self.conf.volcengine_ak,
            secret_access_key=self.conf.volcengine_sk,
            session_token=self.conf.session_token,
            region=self.conf.volcengine_region,
            service=self.service_code,
        )
        
        match_list = [match] if match else None
        
        resp = sign.sign_and_request(
            volcenginesdkcore.UniversalInfo(
                method="POST",
                service=self.service_code,
                version=self.service_version,
                action="GetLabelValues",
                content_type=self.content_type_json,
            ),
            credentials,
            host=self.conf.volcengine_endpoint,
            query={
                'workspace': workspace_id,
                'label': '__name__',
            },
            header={
                "Content-Type": self.content_type_json,
            },
            body=models.GetLabelValuesRequest(
                workspace=workspace_id,
                label='__name__',
                matches=match_list,
            ),
        )
        return resp
    
    def query_metric_labels(self, workspace_id: str, metric_name: str) -> dict:
        """
        查询指标的标签列表
        
        Args:
            workspace_id: 工作区 ID
            metric_name: 指标名称
        
        Returns:
            标签列表
        """
        credentials = models.Credentials(
            access_key_id=self.conf.volcengine_ak,
            secret_access_key=self.conf.volcengine_sk,
            session_token=self.conf.session_token,
            region=self.conf.volcengine_region,
            service=self.service_code,
        )
        
        match_list = [metric_name] if metric_name else None
        
        resp = sign.sign_and_request(
            volcenginesdkcore.UniversalInfo(
                method="POST",
                service=self.service_code,
                version=self.service_version,
                action="GetLabels",
                content_type=self.content_type_json,
            ),
            credentials,
            host=self.conf.volcengine_endpoint,
            query={
                'workspace': workspace_id,
            },
            header={
                "Content-Type": self.content_type_json,
            },
            body=models.GetLabelsRequest(
                workspace=workspace_id,
                matches=match_list,
            ),
        )
        return resp

    def list_workspaces(self) -> dict:
        """
        查询 VMP 工作区列表

        说明：
        - 这部分能力使用官方 SDK（`volcenginesdkvmp`）实现，更稳定也更贴近控制台行为
        - SDK 缺失时返回结构化错误，避免脚本直接崩溃
        """
        try:
            import volcenginesdkvmp
            from volcenginesdkcore.rest import ApiException
        except Exception as exc:
            return {
                "error": "missing_dependency",
                "message": "缺少依赖 volcenginesdkvmp，请先安装 volcenginesdkvmp 或 volcengine-python-sdk",
                "detail": str(exc),
            }

        configuration = volcenginesdkcore.Configuration()
        configuration.ak = self.conf.volcengine_ak
        configuration.sk = self.conf.volcengine_sk
        configuration.region = self.conf.volcengine_region
        volcenginesdkcore.Configuration.set_default(configuration)

        api_instance = volcenginesdkvmp.VMPApi()
        req = volcenginesdkvmp.ListWorkspacesRequest()
        try:
            resp = api_instance.list_workspaces(req)
            return resp.to_dict() if hasattr(resp, "to_dict") else {"Result": resp}
        except ApiException as e:
            return {"error": "api_error", "message": str(e)}
    
    def query_series(self, workspace_id: str, match: str, 
                    start: Optional[str] = None, end: Optional[str] = None) -> dict:
        """
        查询时间序列
        
        Args:
            workspace_id: 工作区 ID
            match: 匹配条件（如 'up{job="node"}'）
            start: 起始时间（可选）
            end: 结束时间（可选）
        
        Returns:
            时间序列列表
        """
        credentials = models.Credentials(
            access_key_id=self.conf.volcengine_ak,
            secret_access_key=self.conf.volcengine_sk,
            session_token=self.conf.session_token,
            region=self.conf.volcengine_region,
            service=self.service_code,
        )
        
        match_list = [match] if match else None
        
        resp = sign.sign_and_request(
            volcenginesdkcore.UniversalInfo(
                method="POST",
                service=self.service_code,
                version=self.service_version,
                action="GetSeries",
                content_type=self.content_type_json,
            ),
            credentials,
            host=self.conf.volcengine_endpoint,
            query={
                'workspace': workspace_id,
            },
            header={
                "Content-Type": self.content_type_form,
            },
            body=models.GetSeriesRequest(
                workspace=workspace_id,
                matches=match_list,
                start=start,
                end=end,
            ),
        )
        return resp
    
    def _calculate_default_step(self, start: str, end: str) -> str:
        """
        计算默认的 step 值
        
        Args:
            start: 起始时间
            end: 结束时间
        
        Returns:
            step 值（秒数）
        """
        try:
            # 简单的计算逻辑
            import datetime
            
            def parse_time(time_str):
                if time_str.isdigit():
                    return float(time_str)
                # RFC3339/ISO8601：兼容 `Z` 结尾；无时区时默认按 UTC 处理
                normalized = time_str.replace("Z", "+00:00")
                dt = datetime.datetime.fromisoformat(normalized)
                if dt.tzinfo is None:
                    dt = dt.replace(tzinfo=datetime.timezone.utc)
                return dt.timestamp()
            
            start_time = parse_time(start)
            end_time = parse_time(end)
            
            duration_seconds = end_time - start_time
            if duration_seconds <= 0:
                return "5"
            
            step_seconds = int(duration_seconds / 100)
            step_seconds = max(step_seconds, 5)
            
            return f"{int(round(step_seconds))}"
        except Exception:
            return "5"