Migraq

{
  "name": "MigraQ",
  "installedAt": 1776152039285,
  "source": "marketplace",
  "iconSource": "migraq",
  "version": "1.0.10"
}

#!/usr/bin/env python3
"""
MigraQ 环境检测脚本（只读，不修改任何配置）

功能：检测 Python 版本、Skill 版本更新（含 changelog）、AK/SK 配置、Gateway 连通性

用法:
    python3 check_env.py                    # 标准模式：输出详细检测结果
    python3 check_env.py --quiet            # 静默模式：仅输出错误信息
    python3 check_env.py --skip-update      # 跳过版本更新检查

返回码:
    0 - 环境就绪（AK/SK + Gateway 全部正常）
    1 - Python 版本不满足（需要 3.7+）
    2 - AK/SK 未配置或鉴权失败
    3 - Gateway 连通性失败
    4 - Skill 版本过旧，需要更新（可用 --skip-update 跳过强制检查）

跨平台支持: Windows / Linux / macOS
"""

import json
import os
import platform
import sys
from datetime import datetime, timezone
from pathlib import Path

SCRIPT_DIR = Path(__file__).resolve().parent
SCRIPTS_DIR = SCRIPT_DIR / "scripts"
sys.path.insert(0, str(SCRIPTS_DIR))

DEFAULT_GATEWAY_URL = "https://cmg.ai.tencentcloudapi.com"
DEFAULT_REGION = "ap-shanghai"
_SERVICE = "cmg"
_VERSION = "2024-10-15"
_ACTION = "ChatCompletions"

# 版本检查配置
#
# 版本号存在两处，必须保持同步：
#   1. SKILL.md front matter 中的 `version` 字段  ← check_env.py 读取此处做版本对比
#   2. _skillhub_meta.json 中的 `version` 字段   ← SkillHub 平台读取此处做安装/更新管理
#
# 每次升级版本时，两处都必须同步修改，否则会出现 check_env 版本与平台显示版本不一致的问题。
SKILL_MD_FILE = SCRIPT_DIR.parent / "SKILL.md"
VERSION_CHECK_TIMEOUT = 15  # 秒
VERSION_CHECK_URL = "https://msp.cloud.tencent.com/skill/version"

# ============== 输出控制 ==============
QUIET_MODE = "--quiet" in sys.argv
SKIP_UPDATE = "--skip-update" in sys.argv


def log_info(msg: str):
    if not QUIET_MODE:
        print(msg)


def log_ok(msg: str):
    if not QUIET_MODE:
        print(f"  [OK] {msg}")


def log_warn(msg: str):
    if not QUIET_MODE:
        print(f"  [WARN] {msg}")


def log_fail(msg: str):
    print(f"  [FAIL] {msg}")


def log_section(title: str):
    if not QUIET_MODE:
        print(f"\n=== {title} ===")


# ============== 版本检查 ==============

def parse_version(version_str: str) -> tuple:
    """解析语义化版本号，如 '1.0.0' -> (1, 0, 0)"""
    try:
        parts = version_str.strip().lstrip("v").split(".")
        return tuple(int(p) for p in parts)
    except (ValueError, AttributeError):
        return (0, 0, 0)


def get_local_version():
    """从 SKILL.md front matter 读取本地版本号，返回 (name, version_str) 或 (None, None)。

    注意：本地版本读取自 SKILL.md，而非 _skillhub_meta.json。
    _skillhub_meta.json 的 version 字段由 SkillHub 平台独立读取，两者需手动保持一致。

    解析 YAML front matter（--- 包裹的头部），提取 name 和 version 字段。
    使用简单的行扫描，无需依赖 PyYAML。
    """
    if not SKILL_MD_FILE.exists():
        return None, None
    try:
        text = SKILL_MD_FILE.read_text(encoding="utf-8")
        lines = text.splitlines()
        # 必须以 --- 开头才是 front matter
        if not lines or lines[0].strip() != "---":
            return None, None
        name, version = None, None
        for line in lines[1:]:
            if line.strip() == "---":
                break
            if line.startswith("name:"):
                name = line.split(":", 1)[1].strip().strip('"').strip("'")
            elif line.startswith("version:"):
                version = line.split(":", 1)[1].strip().strip('"').strip("'")
        return name, version
    except IOError:
        return None, None


def get_remote_version(name: str):
    """从 SkillHub 查询最新 Skill 版本，返回 (version_str, raw_data) 或 (None, None)

    接口: GET https://msp.cloud.tencent.com/skill/version
    响应: {"Response": {"Data": {"Version": "x.y.z"}}}
    """
    try:
        from urllib.request import urlopen, Request as _Req
        import ssl
        req = _Req(VERSION_CHECK_URL, headers={"Accept": "application/json"})
        ctx = ssl.create_default_context()
        with urlopen(req, context=ctx, timeout=VERSION_CHECK_TIMEOUT) as resp:
            data = json.loads(resp.read().decode("utf-8"))
            version = data.get("Response", {}).get("Data", {}).get("Version")
            return version, data
    except Exception:
        return None, None


def check_version_update() -> dict:
    """检查本地版本与远端版本是否一致"""
    name, local_ver = get_local_version()
    if not name or not local_ver:
        return {
            "status": "no_meta",
            "message": "未找到 _skillhub_meta.json 或版本信息缺失",
        }

    remote_ver, remote_data = get_remote_version(name)
    if not remote_ver:
        return {
            "status": "check_failed",
            "local_version": local_ver,
            "message": "无法获取远端版本信息（网络问题或接口不可用）",
        }

    local_parsed = parse_version(local_ver)
    remote_parsed = parse_version(remote_ver)

    if remote_parsed <= local_parsed:
        return {
            "status": "up_to_date",
            "local_version": local_ver,
            "remote_version": remote_ver,
            "message": f"当前已是最新版本: {local_ver}",
        }

    # 收集 changelog（新接口暂不返回 changelog，预留扩展）
    changelog_lines = []
    versions = remote_data.get("Response", {}).get("Data", {}).get("Versions", []) if remote_data else []
    for v in versions:
        v_str = v.get("Version") or v.get("version", "")
        v_parsed = parse_version(v_str)
        if v_parsed > local_parsed:
            desc = v.get("Changelog") or v.get("changelog") or v.get("description") or ""
            changelog_lines.append(f"  {v_str}: {desc}" if desc else f"  {v_str}")

    if not changelog_lines and remote_data:
        latest_cl = (remote_data.get("Response", {}).get("Data", {}).get("Changelog", "")
                     or remote_data.get("latestVersion", {}).get("changelog", ""))
        if latest_cl:
            changelog_lines.append(f"  {remote_ver}: {latest_cl}")

    return {
        "status": "update_available",
        "local_version": local_ver,
        "remote_version": remote_ver,
        "changelog": changelog_lines,
        "message": f"发现新版本: {local_ver} → {remote_ver}",
    }


# ============== Gateway 连通性检测 ==============

def _tc3_sign(secret_key: str, secret_id: str, host: str, payload_str: str,
              action: str, version: str, region: str, timestamp: int) -> dict:
    """生成腾讯云 TC3-HMAC-SHA256 签名，返回请求头字典"""
    import hashlib, hmac as _hmac
    date = datetime.fromtimestamp(timestamp, tz=timezone.utc).strftime("%Y-%m-%d")

    ct = "application/json"
    canonical_headers = f"content-type:{ct}\nhost:{host}\nx-tc-action:{action.lower()}\n"
    signed_headers = "content-type;host;x-tc-action"
    hashed_payload = hashlib.sha256(payload_str.encode("utf-8")).hexdigest()
    canonical_request = "\n".join([
        "POST", "/", "",
        canonical_headers, signed_headers, hashed_payload,
    ])

    algorithm = "TC3-HMAC-SHA256"
    credential_scope = f"{date}/{_SERVICE}/tc3_request"
    hashed_cr = hashlib.sha256(canonical_request.encode("utf-8")).hexdigest()
    string_to_sign = f"{algorithm}\n{timestamp}\n{credential_scope}\n{hashed_cr}"

    def _hmac_sha256(key: bytes, msg: str) -> bytes:
        return _hmac.new(key, msg.encode("utf-8"), hashlib.sha256).digest()

    secret_date = _hmac_sha256(("TC3" + secret_key).encode("utf-8"), date)
    secret_service = _hmac_sha256(secret_date, _SERVICE)
    secret_signing = _hmac_sha256(secret_service, "tc3_request")
    signature = _hmac.new(secret_signing, string_to_sign.encode("utf-8"), hashlib.sha256).hexdigest()

    authorization = (
        f"{algorithm} Credential={secret_id}/{credential_scope}, "
        f"SignedHeaders={signed_headers}, Signature={signature}"
    )
    return {
        "Host": host, "Content-Type": ct,
        "X-TC-Action": action, "X-TC-Version": version,
        "X-TC-Timestamp": str(timestamp), "X-TC-Region": region,
        "Authorization": authorization,
    }


def check_gateway_connectivity(gateway_url: str, secret_key: str,
                                secret_id: str = "") -> dict:
    """
    验证 CMG API 连通性和鉴权状态：发送 SSE 请求，读取前几行 SSE 数据判断结果。

    - HTTP 非 200 → 网络/HTTP 错误
    - HTTP 200 + SSE 流内含错误事件 → 鉴权失败或业务错误
    - HTTP 200 + 收到 delta 或 completed → 连通且鉴权正常

    Returns:
        {"ok": bool, "code": str, "message": str}
    """
    from http.client import HTTPSConnection
    import ssl

    host = "cmg.ai.tencentcloudapi.com"
    region = os.environ.get("CMG_REGION", DEFAULT_REGION)
    payload_str = json.dumps({"Input": "ping", "Stream": True}, separators=(",", ":"))

    now_ts = int(datetime.now(tz=timezone.utc).timestamp())
    headers = _tc3_sign(secret_key, secret_id, host, payload_str,
                        _ACTION, _VERSION, region, now_ts)
    headers["Accept"] = "text/event-stream"

    _AUTH_FAIL_CODES = {
        "AuthFailure.SecretIdNotFound",
        "AuthFailure.SignatureFailure",
        "AuthFailure.SignatureExpire",
        "AuthFailure.InvalidSecretId",
        "AuthFailure.TokenFailure",
        "AuthFailure.InvalidAuthorization",
    }

    try:
        ctx = ssl.create_default_context()
        conn = HTTPSConnection(host, context=ctx, timeout=20)
        conn.request("POST", "/", body=payload_str.encode("utf-8"), headers=headers)
        resp = conn.getresponse()
        status = resp.status

        if status in (401, 403):
            conn.close()
            return {"ok": False, "code": "AuthError", "message": f"鉴权失败 (HTTP {status})，请检查 AK/SK 是否正确"}

        if status != 200:
            conn.close()
            return {"ok": False, "code": "HTTPError", "message": f"CMG API 返回 HTTP {status}"}

        # HTTP 200：先检查 Content-Type
        # 鉴权失败时后端返回 text/plain + JSON 错误体（非 SSE 流）
        content_type = resp.getheader("Content-Type", "")
        if "text/plain" in content_type or "application/json" in content_type:
            try:
                body = resp.read().decode("utf-8")
                data = json.loads(body)
                err = data.get("Response", {}).get("Error", {})
                err_code = err.get("Code", "")
                err_msg = err.get("Message", "")
                if err_code in _AUTH_FAIL_CODES:
                    return {"ok": False, "code": "AuthError", "message": f"鉴权失败: {err_msg}"}
                if err_code:
                    return {"ok": False, "code": err_code, "message": err_msg}
            except Exception:
                pass
            finally:
                conn.close()
            return {"ok": True, "code": "OK", "message": "CMG API 连通正常"}

        # HTTP 200：读取前几行 SSE 数据判断鉴权和业务状态
        try:
            for _ in range(30):  # 最多读 30 行，避免无限等待
                raw = resp.readline()
                if not raw:
                    break
                line = raw.decode("utf-8").rstrip("\r\n")
                if not line or not line.startswith("data:"):
                    continue
                data_str = line[5:].lstrip()
                if data_str == "[DONE]":
                    break
                try:
                    data = json.loads(data_str)
                except (json.JSONDecodeError, ValueError):
                    continue

                event_type = data.get("type", "")

                # 收到任意进度/增量/完成事件 → 鉴权通过，连通正常
                if event_type in (
                    "run.started", "run.progress",
                    "message.delta", "message.completed",
                    # 兼容旧格式
                    "response.output_text.delta", "response.completed",
                ):
                    return {"ok": True, "code": "OK", "message": "CMG API 连通正常"}

                # 失败事件 → 提取错误信息
                if event_type in ("response.failed", "error"):
                    err = data.get("response", data).get("error", {})
                    err_code = err.get("code", "StreamError")
                    err_msg = err.get("message", str(data))
                    if err_code in _AUTH_FAIL_CODES or "auth" in err_code.lower():
                        return {"ok": False, "code": "AuthError", "message": f"鉴权失败: {err_msg}"}
                    return {"ok": False, "code": "StreamError", "message": err_msg}

                # 腾讯云标准错误结构（Response.Error）
                if "Response" in data:
                    err = data["Response"].get("Error", {})
                    err_code = err.get("Code", "")
                    err_msg = err.get("Message", "")
                    if err_code in _AUTH_FAIL_CODES:
                        return {"ok": False, "code": "AuthError", "message": f"鉴权失败: {err_msg}"}
                    if err_code:
                        return {"ok": False, "code": err_code, "message": err_msg}

            # 读完了但没有明确成功/失败事件，视为连通（接口已响应）
            return {"ok": True, "code": "OK", "message": "CMG API 连通正常"}
        finally:
            conn.close()

    except Exception as e:
        return {"ok": False, "code": "NetworkError", "message": f"连接异常: {e}"}


# ============== 主流程 ==============

def main():
    # ver_result 在版本检查后用于最终 JSON 摘要
    ver_result = None

    # ============== 1. 检查 Python 版本 ==============
    log_section("1. 检查运行环境")

    py_ver = sys.version_info
    if py_ver < (3, 7):
        log_fail(f"Python 版本过低: {sys.version}，需要 Python 3.7+")
        sys.exit(1)

    log_ok(f"Python {py_ver.major}.{py_ver.minor}.{py_ver.micro} ({platform.system()} {platform.machine()})")

    # ============== 2. 检查 Skill 版本更新 ==============
    log_section("2. 检查 Skill 版本")

    if SKIP_UPDATE:
        log_ok("已跳过版本更新检查（--skip-update）")
    else:
        ver_result = check_version_update()
        status = ver_result["status"]

        if status == "up_to_date":
            log_ok(ver_result["message"])
        elif status == "update_available":
            log_warn(f"发现新版本 {ver_result['remote_version']}（当前 {ver_result['local_version']}），可前往 SkillHub 更新")
            changelog = ver_result.get("changelog", [])
            if changelog:
                log_info("")
                log_info("  === Changelog ===")
                for line in changelog:
                    log_info(line)
                log_info("")
        elif status in ("check_failed", "no_meta"):
            log_warn(ver_result["message"])
            log_info("  版本检查跳过，继续后续检测...")

    # ============== 3. 检查 AK/SK 配置 ==============
    log_section("3. 检查 AK/SK 配置")

    secret_id = os.environ.get("TENCENTCLOUD_SECRET_ID", "")
    secret_key = os.environ.get("TENCENTCLOUD_SECRET_KEY", "")

    if not secret_id or not secret_key:
        missing = []
        if not secret_id:
            missing.append("TENCENTCLOUD_SECRET_ID")
        if not secret_key:
            missing.append("TENCENTCLOUD_SECRET_KEY")
        log_fail(f"未配置以下环境变量: {', '.join(missing)}")
        log_info("")
        log_info("  Linux / macOS（写入 ~/.zshrc 或 ~/.bashrc）:")
        log_info('    echo \'export TENCENTCLOUD_SECRET_ID="your-secret-id"\' >> ~/.zshrc')
        log_info('    echo \'export TENCENTCLOUD_SECRET_KEY="your-secret-key"\' >> ~/.zshrc')
        log_info("    source ~/.zshrc")
        log_info("")
        log_info("  Windows PowerShell（写入用户级环境变量）:")
        log_info('    [Environment]::SetEnvironmentVariable("TENCENTCLOUD_SECRET_ID", "your-secret-id", "User")')
        log_info('    [Environment]::SetEnvironmentVariable("TENCENTCLOUD_SECRET_KEY", "your-secret-key", "User")')
        log_info("")
        log_info("  密钥获取地址: https://console.cloud.tencent.com/cam/capi")
        sys.exit(2)

    masked_id = f"{secret_id[:4]}****{secret_id[-4:]}" if len(secret_id) > 8 else "****"
    log_ok(f"TENCENTCLOUD_SECRET_ID 已配置: {masked_id}")
    log_ok("TENCENTCLOUD_SECRET_KEY 已配置: ****")

    # ============== 4. 验证 CMG API 连通性 ==============
    log_section("4. 验证 CMG API 连通性")

    cmg_host = "cmg.ai.tencentcloudapi.com"
    log_ok(f"CMG API: https://{cmg_host}/")

    conn_result = check_gateway_connectivity(cmg_host, secret_key, secret_id)

    if conn_result["ok"]:
        log_ok(conn_result["message"])
    else:
        code = conn_result["code"]
        msg = conn_result["message"]
        if code == "AuthError":
            log_fail(f"鉴权失败: {msg}")
            log_info("  请检查 TENCENTCLOUD_SECRET_KEY 是否正确")
            sys.exit(2)
        else:
            log_fail(f"Gateway 连通性失败: {msg}")
            log_info(f"  请检查网络是否可达 ({cmg_host})")
            sys.exit(3)

    # ============== 检测完成 ==============
    log_info("")
    log_info("=== 检测完成 ===")
    log_ok("环境就绪，所有功能可用")
    log_info("")
    log_info(f"  [OK] Python {py_ver.major}.{py_ver.minor} ({platform.system()})")
    log_info(f"  [OK] AK/SK 已配置（SecretId: {masked_id}）")
    log_info(f"  [OK] Gateway 连通正常")

    # 输出结构化 JSON 摘要（供 AI 解析），始终打印（不受 --quiet 影响）
    summary = {"status": "ready"}
    if not SKIP_UPDATE and ver_result and ver_result.get("status") == "update_available":
        summary["update_available"] = True
        summary["local_version"] = ver_result.get("local_version", "")
        summary["remote_version"] = ver_result.get("remote_version", "")
    print(json.dumps(summary, ensure_ascii=False))

    sys.exit(0)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
MigraQ ChatCompletions SSE 流式调用脚本

通过腾讯云 TC3-HMAC-SHA256 签名调用 CMG ChatCompletions 接口。

接口固定参数：
    host:    cmg.ai.tencentcloudapi.com
    action:  ChatCompletions
    version: 2024-10-15
    region:  ap-shanghai（可通过 CMG_REGION 环境变量覆盖）

    请求格式：
    {"Input": "...", "Stream": true}

响应格式（SSE，text/event-stream）：
    event: run.started
    data: {"type":"run.started","session_id":"..."}

    event: message.delta
    data: {"type":"message.delta","delta":"..."}

    event: message.completed
    data: {"type":"message.completed","reply":"...","usage":{...}}

纯 Python 标准库实现，无外部依赖，支持 Windows / Linux / macOS。

用法 (命令行):
    python3 migrateq_sse_api.py <question> [session_id]
    python3 migrateq_sse_api.py --clear-session
    python3 migrateq_sse_api.py --dry-run <question> [session_id]

示例:
    python3 migrateq_sse_api.py '阿里云50台ECS如何迁移到腾讯云？'
    python3 migrateq_sse_api.py '详细说说 go2tencentcloud 步骤' '550e8400-e29b-41d4-a716-446655440000'
    python3 migrateq_sse_api.py --dry-run '测试鉴权是否正确'

作为模块导入:
    from migrateq_sse_api import call_sse_api, generate_session_id
    session_id = generate_session_id()
    result = call_sse_api(question="如何评估迁移成本？", session_id=session_id)

环境变量:
    TENCENTCLOUD_SECRET_ID     - 腾讯云 SecretId（必填）
    TENCENTCLOUD_SECRET_KEY    - 腾讯云 SecretKey（必填）
    CMG_REGION                 - 地域（可选，默认 ap-shanghai）

输出格式（统一 JSON）:
    成功: {"success": true, "action": "ChatCompletions", "data": {"content": "...", "is_final": true, "session_id": "..."}, "requestId": "..."}
    失败: {"success": false, "action": "ChatCompletions", "error": {"code": "...", "message": "..."}, "requestId": ""}
"""

import datetime
import hashlib
import hmac
import json
import os
import ssl
import sys
import threading
import time
import uuid
from http.client import HTTPSConnection
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen

# ---------------------------------------------------------------------------
# 固定参数
# ---------------------------------------------------------------------------
ACTION = "ChatCompletions"
_HOST = "cmg.ai.tencentcloudapi.com"
_VERSION = "2024-10-15"
_SERVICE = "cmg"
_DEFAULT_REGION = "ap-shanghai"
_AUTH_FAIL_CODES = {
    "AuthFailure.SecretIdNotFound",
    "AuthFailure.SignatureFailure",
    "AuthFailure.SignatureExpire",
    "AuthFailure.InvalidSecretId",
    "AuthFailure.TokenFailure",
    "AuthFailure.InvalidAuthorization",
}


# ---------------------------------------------------------------------------
# 会话管理
# ---------------------------------------------------------------------------

def generate_session_id() -> str:
    """
    生成新的 SessionID（UUID v4）。

    SessionID 用于控制多轮对话上下文：
    - 同一对话的所有轮次必须使用同一个 SessionID
    - 用户开启新对话时调用本函数生成新的 SessionID

    Returns:
        str: UUID v4 格式的 SessionID
    """
    return str(uuid.uuid4())


# ---------------------------------------------------------------------------
# 内部工具函数
# ---------------------------------------------------------------------------

def _get_ssl_context():
    """获取 SSL 上下文，始终启用证书验证"""
    try:
        import certifi
        return ssl.create_default_context(cafile=certifi.where())
    except ImportError:
        return ssl.create_default_context()


def _make_error(code: str, message: str, request_id: str = "") -> dict:
    """构造统一错误结果"""
    return {
        "success": False,
        "action": ACTION,
        "error": {"code": code, "message": message},
        "requestId": request_id,
    }


def _make_success(data: dict, request_id: str = "") -> dict:
    """构造统一成功结果"""
    return {
        "success": True,
        "action": ACTION,
        "data": data,
        "requestId": request_id,
    }


def _resolve_credentials(secret_id: str = None, secret_key: str = None):
    """读取 AK/SK 凭证（优先参数，其次环境变量）"""
    secret_id = secret_id or os.environ.get("TENCENTCLOUD_SECRET_ID", "")
    secret_key = secret_key or os.environ.get("TENCENTCLOUD_SECRET_KEY", "")
    return secret_id, secret_key


def _check_credentials(secret_id: str, secret_key: str):
    """
    检查 AK/SK 是否已配置，未配置则返回错误结果。

    Returns:
        None 表示凭证完整；dict 表示缺失，包含引导信息。
    """
    missing = []
    if not secret_id:
        missing.append("TENCENTCLOUD_SECRET_ID")
    if not secret_key:
        missing.append("TENCENTCLOUD_SECRET_KEY")
    if not missing:
        return None

    guide = (
        "请先配置腾讯云 API 密钥后再使用 MigraQ。\n"
        f"  缺少环境变量: {', '.join(missing)}\n"
        "\n"
        "  Linux / macOS（写入 ~/.zshrc 或 ~/.bashrc）:\n"
        '    echo \'export TENCENTCLOUD_SECRET_ID="your-secret-id"\' >> ~/.zshrc\n'
        '    echo \'export TENCENTCLOUD_SECRET_KEY="your-secret-key"\' >> ~/.zshrc\n'
        "    source ~/.zshrc\n"
        "\n"
        "  Windows PowerShell（写入用户级环境变量）:\n"
        '    [Environment]::SetEnvironmentVariable("TENCENTCLOUD_SECRET_ID", "your-secret-id", "User")\n'
        '    [Environment]::SetEnvironmentVariable("TENCENTCLOUD_SECRET_KEY", "your-secret-key", "User")\n'
        "\n"
        "  密钥获取地址: https://console.cloud.tencent.com/cam/capi"
    )
    return _make_error("MissingCredentials", guide)


def _tc3_sign(secret_key: str, secret_id: str, host: str, payload_str: str,
              action: str, version: str, region: str, timestamp: int) -> dict:
    """
    生成腾讯云 TC3-HMAC-SHA256 签名，返回请求头字典。

    参考: https://cloud.tencent.com/document/api/213/30654
    """
    date = datetime.datetime.fromtimestamp(timestamp, tz=datetime.timezone.utc).strftime("%Y-%m-%d")

    # Step 1: 构造规范请求（SignedHeaders 包含 x-tc-action，提升签名安全性）
    ct = "application/json"
    canonical_headers = f"content-type:{ct}\nhost:{host}\nx-tc-action:{action.lower()}\n"
    signed_headers = "content-type;host;x-tc-action"
    hashed_payload = hashlib.sha256(payload_str.encode("utf-8")).hexdigest()
    canonical_request = "\n".join([
        "POST", "/", "",
        canonical_headers, signed_headers, hashed_payload,
    ])

    # Step 2: 构造待签字符串
    algorithm = "TC3-HMAC-SHA256"
    credential_scope = f"{date}/{_SERVICE}/tc3_request"
    hashed_cr = hashlib.sha256(canonical_request.encode("utf-8")).hexdigest()
    string_to_sign = f"{algorithm}\n{timestamp}\n{credential_scope}\n{hashed_cr}"

    # Step 3: 计算签名
    def _hmac_sha256(key: bytes, msg: str) -> bytes:
        return hmac.new(key, msg.encode("utf-8"), hashlib.sha256).digest()

    secret_date = _hmac_sha256(("TC3" + secret_key).encode("utf-8"), date)
    secret_service = _hmac_sha256(secret_date, _SERVICE)
    secret_signing = _hmac_sha256(secret_service, "tc3_request")
    signature = hmac.new(secret_signing, string_to_sign.encode("utf-8"), hashlib.sha256).hexdigest()

    # Step 4: 构造 Authorization
    authorization = (
        f"{algorithm} Credential={secret_id}/{credential_scope}, "
        f"SignedHeaders={signed_headers}, Signature={signature}"
    )

    return {
        "Host": host,
        "Content-Type": ct,
        "X-TC-Action": action,
        "X-TC-Version": version,
        "X-TC-Timestamp": str(timestamp),
        "X-TC-Region": region,
        "X-TC-Language": "zh-CN",
        "Authorization": authorization,
    }


# ---------------------------------------------------------------------------
# SSE 行解析
# ---------------------------------------------------------------------------

def parse_sse_line(line: str):
    """
    解析单行 SSE 数据。

    Returns:
        dict | None:
            - id 行:               {"type": "id", "value": "..."}
            - event 行:            {"event": "<value>"}
            - data 行(JSON 有效):  {"event": "data", "data": {...}}
            - data 行(JSON 无效):  {"event": "data", "raw": "..."}
            - 空行/注释行:          None
    """
    if not line or line.startswith(":"):
        return None

    if line.startswith("id:"):
        return {"type": "id", "value": line[3:].strip()}

    if line.startswith("data:"):
        payload = line[5:].lstrip()
        try:
            return {"event": "data", "data": json.loads(payload)}
        except (json.JSONDecodeError, ValueError):
            return {"event": "data", "raw": payload}

    if line.startswith("event:"):
        return {"event": line[6:].strip()}

    return None


# ---------------------------------------------------------------------------
# SSE 流式 API 调用
# ---------------------------------------------------------------------------

def call_sse_api(question: str, session_id: str = None,
                 region: str = None,
                 secret_id: str = None, secret_key: str = None,
                 on_delta=None, timeout: int = 600) -> dict:
    """
    调用 ChatCompletions SSE 流式 API。

    Args:
        question:    用户问题（必填）
        session_id:  会话 ID（必传，首次调用不传则自动生成 UUID v4）。
                     作为 SessionKey 传入服务端，服务端按此字段隔离对话上下文。
                     多轮对话时必须传入上次返回的 session_id 以保持上下文。
        region:      地域，不传则从 CMG_REGION 环境变量读取，默认 ap-shanghai
        secret_id:   腾讯云 SecretId，不传则从环境变量读取
        secret_key:  腾讯云 SecretKey，不传则从环境变量读取
        on_delta:    回调函数，每收到一段流式文本时调用，参数为 str
        timeout:     请求超时秒数，默认 600

    Returns:
        dict: 统一格式的结果字典
    """
    region = region or os.environ.get("CMG_REGION", _DEFAULT_REGION)
    secret_id, secret_key = _resolve_credentials(secret_id, secret_key)

    cred_err = _check_credentials(secret_id, secret_key)
    if cred_err:
        return cred_err

    # 首次调用未传入 session_id 时自动生成，多轮对话时沿用上次返回的值
    if not session_id:
        session_id = generate_session_id()

    payload = {"Input": question, "Stream": True, "SessionKey": session_id}
    payload_str = json.dumps(payload, ensure_ascii=False, separators=(",", ":"))

    now = datetime.datetime.now(datetime.timezone.utc)
    timestamp = int(now.timestamp())

    headers = _tc3_sign(secret_key, secret_id, _HOST, payload_str,
                        ACTION, _VERSION, region, timestamp)
    headers["Accept"] = "text/event-stream"

    # 使用 HTTPSConnection 实现无缓冲实时 SSE 读取
    try:
        ctx = _get_ssl_context()
        conn = HTTPSConnection(_HOST, context=ctx, timeout=timeout)
        conn.request("POST", "/", body=payload_str.encode("utf-8"), headers=headers)
        resp = conn.getresponse()
    except Exception as e:
        return _make_error("NetworkError", f"无法连接 CMG API ({_HOST}): {e}")

    if resp.status != 200:
        try:
            body = resp.read().decode("utf-8")
            data = json.loads(body)
            err = data.get("Response", {}).get("Error", {})
            code = err.get("Code", "")
            msg = err.get("Message") or f"HTTP {resp.status}"
        except Exception:
            code = ""
            msg = f"HTTP {resp.status}"
        conn.close()
        # 鉴权失败（401/403 或 AuthFailure 错误码）单独返回，方便上层区分处理
        if resp.status in (401, 403) or code in _AUTH_FAIL_CODES:
            return _make_error("AuthError", f"鉴权失败，请检查 AK/SK 是否正确: {msg}")
        return _make_error("HTTPError", f"CMG API 返回错误: {msg}")

    # HTTP 200：检查 Content-Type
    # 鉴权失败时后端返回 text/plain + JSON 错误体，不是 SSE 流
    content_type = resp.getheader("Content-Type", "")
    if "text/plain" in content_type or "application/json" in content_type:
        try:
            body = resp.read().decode("utf-8")
            data = json.loads(body)
            err = data.get("Response", {}).get("Error", {})
            code = err.get("Code", "")
            msg = err.get("Message", body)
            request_id = data.get("Response", {}).get("RequestId", "")
            if code in _AUTH_FAIL_CODES or code.startswith("AuthFailure"):
                return _make_error("AuthError", f"鉴权失败，请检查 AK/SK 是否正确: {msg}", request_id)
            if code:
                return _make_error("HTTPError", f"CMG API 返回错误 [{code}]: {msg}", request_id)
            return _make_error("StreamError", "CMG API 返回非流式响应且无法解析")
        except Exception as e:
            return _make_error("StreamError", f"CMG API 返回非流式响应: {e}")
        finally:
            conn.close()

    try:
        return _parse_sse_stream(resp, session_id, on_delta)
    finally:
        conn.close()


def clear_session(region: str = None,
                  secret_id: str = None, secret_key: str = None) -> dict:
    """
    清除当前会话上下文（本接口无状态，调用方重新生成 session_id 即可）。

    为保持接口兼容性而保留，实际执行为空操作并返回成功。

    Returns:
        dict: 统一格式的结果字典
    """
    return _make_success({"message": "session cleared"})


def _parse_sse_stream(resp, session_id: str, on_delta) -> dict:
    """
    解析 ChatCompletions SSE 流并构建结果。

    实际响应格式：
        event: run.started
        data: {"type":"run.started","session_id":"..."}

        event: run.progress
        data: {"type":"run.progress","stage":"...","summary":"..."}

        event: message.delta
        data: {"type":"message.delta","delta":"..."}

        event: message.completed
        data: {"type":"message.completed","reply":"...","usage":{...}}
    """
    content_parts = []
    request_id = ""
    usage = {}
    first_delta_received = False
    stream_error = None  # 记录 SSE 流内的业务错误

    # 心跳线程：在收到第一个文本增量前，每 10 秒向 stderr 输出一次等待提示
    stop_heartbeat = threading.Event()

    def _heartbeat():
        elapsed = 0
        while not stop_heartbeat.wait(10):
            if first_delta_received:
                break
            elapsed += 10
            print(f"[MigraQ] 远端专家处理中，已等待 {elapsed} 秒，请勿中断……", file=sys.stderr, flush=True)

    heartbeat_thread = threading.Thread(target=_heartbeat, daemon=True)
    heartbeat_thread.start()

    try:
        # 使用 readline() 逐行读取，无缓冲，保证 SSE 实时性
        while True:
            raw = resp.readline()
            if not raw:
                break

            line = raw.decode("utf-8").rstrip("\r\n")

            if line == "" or line.startswith(":"):
                continue

            if line.startswith("event:"):
                continue  # event 类型已通过 data.type 区分，无需单独处理

            if not line.startswith("data:"):
                continue

            data_str = line[5:].lstrip()

            if data_str == "[DONE]":
                break

            try:
                data = json.loads(data_str)
            except (json.JSONDecodeError, ValueError):
                continue

            event_type = data.get("type", "")

            # 流式文本增量
            if event_type == "message.delta":
                delta = data.get("delta", "")
                if delta:
                    if not first_delta_received:
                        first_delta_received = True
                        stop_heartbeat.set()
                    content_parts.append(delta)
                    if on_delta:
                        on_delta(delta)

            # 完成事件：提取完整回复和 usage
            elif event_type == "message.completed":
                # 若流式 delta 已拼接内容则保留，否则用 reply 字段兜底
                if not content_parts:
                    reply = data.get("reply", "")
                    if reply:
                        content_parts.append(reply)
                        if on_delta:
                            on_delta(reply)
                usage = data.get("usage", {})
                break

            # 失败事件：后端通过 SSE 流返回的业务错误
            elif event_type in ("response.failed", "error"):
                err_obj = data.get("response", data)
                err_detail = err_obj.get("error", {})
                err_code = err_detail.get("code") or err_obj.get("code", "StreamError")
                err_msg = err_detail.get("message") or err_obj.get("message", str(data))
                stream_error = _make_error("StreamError", f"远端服务返回错误 [{err_code}]: {err_msg}", request_id)
                break

            # 兼容：data 中直接含 Response.Error 字段（部分腾讯云接口格式）
            elif "Response" in data:
                resp_err = data["Response"].get("Error")
                if resp_err:
                    err_code = resp_err.get("Code", "StreamError")
                    err_msg = resp_err.get("Message", str(resp_err))
                    request_id = data["Response"].get("RequestId", "")
                    stream_error = _make_error("StreamError", f"远端服务返回错误 [{err_code}]: {err_msg}", request_id)
                    break

    finally:
        stop_heartbeat.set()

    # 优先返回流内业务错误
    if stream_error:
        return stream_error

    # SSE 流结束但无任何内容且无错误事件，视为异常
    if not content_parts:
        return _make_error("StreamError", "远端服务未返回任何内容，请稍后重试或检查网络连接", request_id)

    return _make_success(
        {
            "content": "".join(content_parts),
            "is_final": True,
            "session_id": session_id,
            "usage": usage,
        },
        request_id,
    )


# ---------------------------------------------------------------------------
# 命令行入口
# ---------------------------------------------------------------------------

def _output_json(obj: dict) -> str:
    return json.dumps(obj, ensure_ascii=False)


def main():
    """命令行入口：python3 migrateq_sse_api.py <question> [session_id]"""
    args = sys.argv[1:]

    # --clear-session 模式
    if args and args[0] == "--clear-session":
        result = clear_session()
        print(_output_json(result))
        sys.exit(0 if result.get("success") else 1)

    # --dry-run 模式：仅打印签名请求头和 payload，不发送请求
    dry_run = False
    if args and args[0] == "--dry-run":
        dry_run = True
        args = args[1:]

    if len(args) < 1:
        print(_output_json(_make_error(
            "MissingParameter",
            "用法: python3 migrateq_sse_api.py <question> [session_id]\n"
            "     python3 migrateq_sse_api.py --dry-run <question> [session_id]\n"
            "     python3 migrateq_sse_api.py --clear-session"
        )))
        sys.exit(1)

    question = args[0]
    session_id = args[1] if len(args) > 1 else None  # 首次调用由 call_sse_api 内部生成

    if dry_run:
        region, secret_id, secret_key = (
            os.environ.get("CMG_REGION", _DEFAULT_REGION),
            *_resolve_credentials(),
        )
        cred_err = _check_credentials(secret_id, secret_key)
        if cred_err:
            print(_output_json(cred_err))
            sys.exit(1)
        if not session_id:
            session_id = generate_session_id()
        payload = {"Input": question, "Stream": True, "SessionKey": session_id}
        payload_str = json.dumps(payload, ensure_ascii=False, separators=(",", ":"))
        timestamp = int(datetime.datetime.now(datetime.timezone.utc).timestamp())
        headers = _tc3_sign(secret_key, secret_id, _HOST, payload_str,
                            ACTION, _VERSION, region, timestamp)
        headers["Accept"] = "text/event-stream"
        dry_run_info = {
            "success": True,
            "action": "DryRun",
            "data": {
                "endpoint": f"https://{_HOST}",
                "session_id": session_id,
                "payload": payload,
                "headers": headers,
            },
        }
        print(_output_json(dry_run_info))
        sys.exit(0)

    def on_delta(delta: str):
        print(delta, end="", flush=True)

    result = call_sse_api(question=question, session_id=session_id, on_delta=on_delta)

    # 流式输出结束后换行，再打印统一 JSON 结果
    print()
    print(_output_json(result))

    if not result.get("success"):
        sys.exit(1)


if __name__ == "__main__":
    main()