Byted Voice To Text

                                 Apache License
                           Version 2.0, January 2004
                        http://www.apache.org/licenses/

   TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION

   1. Definitions.

      "License" shall mean the terms and conditions for use, reproduction,
      and distribution as defined by Sections 1 through 9 of this document.

      "Licensor" shall mean the copyright owner or entity authorized by
      the copyright owner that is granting the License.

      "Legal Entity" shall mean the union of the acting entity and all
      other entities that control, are controlled by, or are under common
      control with that entity. For the purposes of this definition,
      "control" means (i) the power, direct or indirect, to cause the
      direction or management of such entity, whether by contract or
      otherwise, or (ii) ownership of fifty percent (50%) or more of the
      outstanding shares, or (iii) beneficial ownership of such entity.

      "You" (or "Your") shall mean an individual or Legal Entity
      exercising permissions granted by this License.

      "Source" form shall mean the preferred form for making modifications,
      including but not limited to software source code, documentation
      source, and configuration files.

      "Object" form shall mean any form resulting from mechanical
      transformation or translation of a Source form, including but
      not limited to compiled object code, generated documentation,
      and conversions to other media types.

      "Work" shall mean the work of authorship, whether in Source or
      Object form, made available under the License, as indicated by a
      copyright notice that is included in or attached to the work
      (an example is provided in the Appendix below).

      "Derivative Works" shall mean any work, whether in Source or Object
      form, that is based on (or derived from) the Work and for which the
      editorial revisions, annotations, elaborations, or other modifications
      represent, as a whole, an original work of authorship. For the purposes
      of this License, Derivative Works shall not include works that remain
      separable from, or merely link (or bind by name) to the interfaces of,
      the Work and Derivative Works thereof.

      "Contribution" shall mean any work of authorship, including
      the original version of the Work and any modifications or additions
      to that Work or Derivative Works thereof, that is intentionally
      submitted to Licensor for inclusion in the Work by the copyright owner
      or by an individual or Legal Entity authorized to submit on behalf of
      the copyright owner. For the purposes of this definition, "submitted"
      means any form of electronic, verbal, or written communication sent
      to the Licensor or its representatives, including but not limited to
      communication on electronic mailing lists, source code control systems,
      and issue tracking systems that are managed by, or on behalf of, the
      Licensor for the purpose of discussing and improving the Work, but
      excluding communication that is conspicuously marked or otherwise
      designated in writing by the copyright owner as "Not a Contribution."

      "Contributor" shall mean Licensor and any individual or Legal Entity
      on behalf of whom a Contribution has been received by Licensor and
      subsequently incorporated within the Work.

   2. Grant of Copyright License. Subject to the terms and conditions of
      this License, each Contributor hereby grants to You a perpetual,
      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
      copyright license to reproduce, prepare Derivative Works of,
      publicly display, publicly perform, sublicense, and distribute the
      Work and such Derivative Works in Source or Object form.

   3. Grant of Patent License. Subject to the terms and conditions of
      this License, each Contributor hereby grants to You a perpetual,
      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
      (except as stated in this section) patent license to make, have made,
      use, offer to sell, sell, import, and otherwise transfer the Work,
      where such license applies only to those patent claims licensable
      by such Contributor that are necessarily infringed by their
      Contribution(s) alone or by combination of their Contribution(s)
      with the Work to which such Contribution(s) was submitted. If You
      institute patent litigation against any entity (including a
      cross-claim or counterclaim in a lawsuit) alleging that the Work
      or a Contribution incorporated within the Work constitutes direct
      or contributory patent infringement, then any patent licenses
      granted to You under this License for that Work shall terminate
      as of the date such litigation is filed.

   4. Redistribution. You may reproduce and distribute copies of the
      Work or Derivative Works thereof in any medium, with or without
      modifications, and in Source or Object form, provided that You
      meet the following conditions:

      (a) You must give any other recipients of the Work or
          Derivative Works a copy of this License; and

      (b) You must cause any modified files to carry prominent notices
          stating that You changed the files; and

      (c) You must retain, in the Source form of any Derivative Works
          that You distribute, all copyright, patent, trademark, and
          attribution notices from the Source form of the Work,
          excluding those notices that do not pertain to any part of
          the Derivative Works; and

      (d) If the Work includes a "NOTICE" text file as part of its
          distribution, then any Derivative Works that You distribute must
          include a readable copy of the attribution notices contained
          within such NOTICE file, excluding those notices that do not
          pertain to any part of the Derivative Works, in at least one
          of the following places: within a NOTICE text file distributed
          as part of the Derivative Works; within the Source form or
          documentation, if provided along with the Derivative Works; or,
          within a display generated by the Derivative Works, if and
          wherever such third-party notices normally appear. The contents
          of the NOTICE file are for informational purposes only and
          do not modify the License. You may add Your own attribution
          notices within Derivative Works that You distribute, alongside
          or as an addendum to the NOTICE text from the Work, provided
          that such additional attribution notices cannot be construed
          as modifying the License.

      You may add Your own copyright statement to Your modifications and
      may provide additional or different license terms and conditions
      for use, reproduction, or distribution of Your modifications, or
      for any such Derivative Works as a whole, provided Your use,
      reproduction, and distribution of the Work otherwise complies with
      the conditions stated in this License.

   5. Submission of Contributions. Unless You explicitly state otherwise,
      any Contribution intentionally submitted for inclusion in the Work
      by You to the Licensor shall be under the terms and conditions of
      this License, without any additional terms or conditions.
      Notwithstanding the above, nothing herein shall supersede or modify
      the terms of any separate license agreement you may have executed
      with Licensor regarding such Contributions.

   6. Trademarks. This License does not grant permission to use the trade
      names, trademarks, service marks, or product names of the Licensor,
      except as required for reasonable and customary use in describing the
      origin of the Work and reproducing the content of the NOTICE file.

   7. Disclaimer of Warranty. Unless required by applicable law or
      agreed to in writing, Licensor provides the Work (and each
      Contributor provides its Contributions) on an "AS IS" BASIS,
      WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
      implied, including, without limitation, any warranties or conditions
      of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
      PARTICULAR PURPOSE. You are solely responsible for determining the
      appropriateness of using or redistributing the Work and assume any
      risks associated with Your exercise of permissions under this License.

   8. Limitation of Liability. In no event and under no legal theory,
      whether in tort (including negligence), contract, or otherwise,
      unless required by applicable law (such as deliberate and grossly
      negligent acts) or agreed to in writing, shall any Contributor be
      liable to You for damages, including any direct, indirect, special,
      incidental, or consequential damages of any character arising as a
      result of this License or out of the use or inability to use the
      Work (including but not limited to damages for loss of goodwill,
      work stoppage, computer failure or malfunction, or any and all
      other commercial damages or losses), even if such Contributor
      has been advised of the possibility of such damages.

   9. Accepting Warranty or Additional Liability. While redistributing
      the Work or Derivative Works thereof, You may choose to offer,
      and charge a fee for, acceptance of support, warranty, indemnity,
      or other liability obligations and/or rights consistent with this
      License. However, in accepting such obligations, You may act only
      on Your own behalf and on Your sole responsibility, not on behalf
      of any other Contributor, and only if You agree to indemnify,
      defend, and hold each Contributor harmless for any liability
      incurred by, or claims asserted against, such Contributor by reason
      of your accepting any such warranty or additional liability.

   END OF TERMS AND CONDITIONS

   APPENDIX: How to apply the Apache License to your work.

      To apply the Apache License to your work, attach the following
      boilerplate notice, with the fields enclosed by brackets "[]"
      replaced with your own identifying information. (Don't include
      the brackets!)  The text should be enclosed in the appropriate
      comment syntax for the file format. We also recommend that a
      file or class name and description of purpose be included on the
      same "printed page" as the copyright notice for easier
      identification within third-party archives.

   Copyright [yyyy] [name of copyright owner]

   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.

# Voice-to-Text skill dependencies
# Ref: https://www.volcengine.com/docs/6561/1354870
requests>=2.28.0

import json
import os
import urllib.error
import urllib.request
from urllib.parse import parse_qsl, urlencode, urlparse, urlunparse


def _build_action_url(base: str, action: str, version: str = "2025-05-20") -> str:
    base = (base or "").strip()
    if not base:
        return ""
    parsed = urlparse(base)
    path = parsed.path or "/"
    if not path.endswith("/"):
        path = f"{path}/"
    query = dict(parse_qsl(parsed.query))
    query["Action"] = action
    query["Version"] = version
    return urlunparse(parsed._replace(path=path, query=urlencode(query)))


def _extract_json(text: str) -> dict | None:
    if not text:
        return None
    text = text.strip()
    try:
        return json.loads(text)
    except Exception:
        pass
    start = text.find("{")
    end = text.rfind("}")
    if start == -1 or end == -1 or end <= start:
        return None
    try:
        return json.loads(text[start : end + 1])
    except Exception:
        return None


def _post_json(url: str, headers: dict, payload: dict, timeout_s: int = 15) -> dict | None:
    if not url:
        return None
    data = json.dumps(payload, ensure_ascii=False).encode("utf-8")
    req = urllib.request.Request(url, data=data, method="POST")
    for k, v in (headers or {}).items():
        if v is None:
            continue
        req.add_header(k, v)
    try:
        with urllib.request.urlopen(req, timeout=timeout_s) as resp:
            body = resp.read().decode("utf-8", errors="replace")
            return _extract_json(body)
    except urllib.error.HTTPError as e:
        try:
            body = e.read().decode("utf-8", errors="replace")
            return _extract_json(body)
        except Exception:
            return None
    except Exception:
        return None


def _dotenv_path() -> str:
    return os.path.join(os.path.dirname(os.path.abspath(__file__)), ".env")


def _load_dotenv_if_available(dotenv_path: str) -> None:
    try:
        from dotenv import load_dotenv  # type: ignore
    except Exception:
        return
    try:
        load_dotenv(dotenv_path=dotenv_path, override=False)
    except Exception:
        return


def _persist_env_to_dotenv_if_available(dotenv_path: str, key: str, value: str) -> None:
    value = (value or "").strip()
    if not value:
        return
    try:
        from dotenv import set_key  # type: ignore
    except Exception:
        return
    try:
        os.makedirs(os.path.dirname(dotenv_path) or ".", exist_ok=True)
        set_key(dotenv_path, key, value, quote_mode="never")
        try:
            os.chmod(dotenv_path, 0o600)
        except Exception:
            pass
    except Exception:
        return


def get_speech_api_key(project_name: str = "default", create_name: str = "arkclaw") -> str:
    _load_dotenv_if_available(_dotenv_path())

    speech_key = os.getenv("MODEL_SPEECH_API_KEY", "").strip()
    if speech_key:
        return speech_key

    ark_key = os.getenv("ARK_SKILL_API_KEY", "").strip()
    base = os.getenv("ARK_SKILL_API_BASE", "").strip()
    if not ark_key or not base:
        return ""

    headers = {
        "ServiceName": "speech_saas_prod",
        "Authorization": f"Bearer {ark_key}",
        "Content-Type": "application/json",
    }

    list_url = _build_action_url(base, "ListAPIKeys")
    list_data = _post_json(
        list_url,
        headers=headers,
        payload={"ProjectName": project_name, "OnlyAvailable": True},
    )
    if isinstance(list_data, dict):
        api_keys = (list_data.get("Result") or {}).get("APIKeys") or []
        if isinstance(api_keys, list) and api_keys:
            first = api_keys[0] if isinstance(api_keys[0], dict) else {}
            first_key = (first.get("APIKey") or "").strip()
            if first_key:
                os.environ["MODEL_SPEECH_API_KEY"] = first_key
                _persist_env_to_dotenv_if_available(_dotenv_path(), "MODEL_SPEECH_API_KEY", first_key)
                return first_key

    create_url = _build_action_url(base, "CreateAPIKey")
    create_data = _post_json(
        create_url,
        headers=headers,
        payload={"ProjectName": project_name, "Name": create_name},
    )
    if isinstance(create_data, dict):
        created_key = ((create_data.get("Result") or {}).get("APIKey") or "").strip()
        if created_key:
            os.environ["MODEL_SPEECH_API_KEY"] = created_key
            _persist_env_to_dotenv_if_available(_dotenv_path(), "MODEL_SPEECH_API_KEY", created_key)
            return created_key

    return ""

# -*- coding: utf-8 -*-
# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates.
#
# 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.

"""
Byted-Voice-to-Text (ASR) using Volcengine BigModel ASR API.
Ref: https://www.volcengine.com/docs/6561/1354870
鉴权: 新版控制台 API Key 方案 https://www.volcengine.com/docs/6561/2119699
"""

from __future__ import annotations

import argparse
import base64
import json
import os
import sys
import tempfile
import time
import uuid

_SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
if _SCRIPT_DIR not in sys.path:
    sys.path.insert(0, _SCRIPT_DIR)

from api_key import get_speech_api_key

ASR_ENDPOINT = os.getenv(
    "MODEL_SPEECH_ASR_API_BASE",
    "https://openspeech.bytedance.com/api/v3/auc/bigmodel/recognize/flash",
)
ASR_RESOURCE_ID = os.getenv("MODEL_SPEECH_ASR_RESOURCE_ID", "volc.bigasr.auc_turbo")


def log(msg: str) -> None:
    print(f"[byted-voice-to-text] {msg}", file=sys.stderr)


def fail(msg: str) -> None:
    full = f"[byted-voice-to-text ERROR] {msg}"
    print(full, file=sys.stderr)
    print(full)
    sys.exit(1)


try:
    import requests
except ImportError:
    fail("requests 库未安装，请执行 pip install requests")


# ============================================================
# 飞书语音文件下载
# ============================================================
def download_feishu_audio(file_key: str, tenant_token: str) -> str:
    """
    通过飞书 file_key 下载语音文件到临时目录，返回本地文件路径。

    飞书语音消息格式：
      - message_type: "audio"
      - content: {"file_key": "file_v2_xxxx"}
      - 音频编码: Opus in OGG 容器

    调用飞书「下载文件」接口:
      GET https://open.feishu.cn/open-apis/im/v1/files/{file_key}
    """
    download_url = f"https://open.feishu.cn/open-apis/im/v1/files/{file_key}"
    headers = {
        "Authorization": f"Bearer {tenant_token}",
        "Content-Type": "application/json; charset=utf-8",
    }

    log(f"正在从飞书下载语音文件: file_key={file_key}")
    resp = requests.get(download_url, headers=headers, timeout=30)

    if resp.status_code != 200:
        fail(
            f"飞书文件下载失败: HTTP {resp.status_code}, "
            f"响应: {resp.text[:500]}"
        )

    tmp_file = tempfile.NamedTemporaryFile(suffix=".ogg", delete=False)
    tmp_file.write(resp.content)
    tmp_file.close()

    file_size = len(resp.content)
    log(f"语音文件已下载: {tmp_file.name} ({file_size} bytes)")
    return tmp_file.name


# ============================================================
# 火山引擎 ASR 识别
# ============================================================
def file_to_base64(file_path: str) -> str:
    with open(file_path, "rb") as f:
        return base64.b64encode(f.read()).decode("utf-8")


def _build_headers(appid: str | None = None, token: str | None = None) -> dict:
    """构建请求头。支持新版控制台(X-Api-Key)和旧版(X-Api-App-Key + X-Api-Access-Key)。"""
    app_id = appid or os.getenv("MODEL_SPEECH_APP_ID", "").strip()
    api_key = token or get_speech_api_key().strip()

    if not api_key:
        raise PermissionError(
            "MODEL_SPEECH_API_KEY 需在环境变量中配置。"
            "见 https://www.volcengine.com/docs/6561/2119699"
        )

    headers = {
        "X-Api-Resource-Id": ASR_RESOURCE_ID,
        "X-Api-Request-Id": str(uuid.uuid4()),
        "X-Api-Sequence": "-1",
    }

    if app_id:
        headers["X-Api-App-Key"] = app_id
        headers["X-Api-Access-Key"] = api_key
    else:
        headers["X-Api-Key"] = api_key

    return headers


def recognize(
    audio_url: str | None = None,
    file_path: str | None = None,
    appid: str | None = None,
    token: str | None = None,
    language: str | None = None,
) -> str:
    """调用火山引擎 BigModel ASR Flash 接口，返回识别文本。"""
    headers = _build_headers(appid=appid, token=token)

    audio_data: dict = {}
    if audio_url:
        audio_data["url"] = audio_url
    elif file_path:
        audio_data["data"] = file_to_base64(file_path)
    else:
        raise ValueError("必须提供 audio_url 或 file_path")

    if language:
        audio_data["language"] = language

    payload = {
        "user": {"uid": headers.get("X-Api-App-Key", "skill_asr_user")},
        "audio": audio_data,
        "request": {
            "model_name": "bigmodel",
            "enable_itn": True,
            "enable_punc": True,
        },
    }

    log("正在调用火山引擎 ASR...")
    resp = requests.post(ASR_ENDPOINT, json=payload, headers=headers, timeout=60)

    status_code = resp.headers.get("X-Api-Status-Code", "")
    if status_code != "20000000":
        msg = resp.headers.get("X-Api-Message", "未知错误")
        logid = resp.headers.get("X-Tt-Logid", "N/A")
        fail(f"ASR 请求失败: code={status_code}, msg={msg}, logid={logid}")

    result = resp.json()
    text_parts: list[str] = []

    if "result" in result:
        res = result["result"]
        if isinstance(res, list):
            for item in res:
                if isinstance(item, dict) and "text" in item:
                    text_parts.append(item["text"])
        elif isinstance(res, dict):
            if "text" in res:
                text_parts.append(res["text"])
            elif "utterances" in res:
                for utt in res["utterances"]:
                    if "text" in utt:
                        text_parts.append(utt["text"])

    if not text_parts and "utterances" in result:
        for utt in result["utterances"]:
            if "text" in utt:
                text_parts.append(utt["text"])

    if not text_parts:
        fail(f"无法从响应中提取文本，原始响应: {json.dumps(result, ensure_ascii=False)}")

    text = "".join(text_parts)
    log(f"识别成功，文本长度: {len(text)} 字符")
    return text


# ============================================================
# 主流程
# ============================================================
def main() -> None:
    parser = argparse.ArgumentParser(
        description="语音转文字工具（火山引擎 BigModel ASR）",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
示例用法:

  # 方式1: 直接传音频 URL
  python asr_flash.py --url "https://example.com/audio.mp3"

  # 方式2: 传本地音频文件
  python asr_flash.py --file "/path/to/audio.ogg"

  # 方式3: 传飞书语音消息的 file_key
  python asr_flash.py --file-key "file_v2_xxxx" --feishu-token "t-g104xxx"

飞书语音消息处理流程:
  收到 audio 消息 → 提取 content.file_key → 本脚本下载+识别 → 返回文字
        """,
    )

    group = parser.add_mutually_exclusive_group(required=True)
    group.add_argument("--url", help="音频文件的 URL 地址")
    group.add_argument("--file", help="本地音频文件路径")
    group.add_argument("--file-key", help="飞书语音消息的 file_key")

    parser.add_argument(
        "--feishu-token",
        help="飞书 tenant_access_token (也可通过环境变量 FEISHU_TENANT_TOKEN 设置)",
    )
    parser.add_argument(
        "--appid",
        help="火山引擎 ASR App ID (也可通过环境变量 MODEL_SPEECH_APP_ID 设置)",
    )
    parser.add_argument(
        "--token",
        help="火山引擎 ASR API Key (也可通过环境变量 MODEL_SPEECH_API_KEY 设置)",
    )
    parser.add_argument(
        "--language",
        help="语言代码，如 zh-CN, ja-JP, en-US (可选，不传则自动识别)",
    )

    args = parser.parse_args()

    log(f"启动参数: file={args.file}, url={args.url}, file_key={args.file_key}")
    log(
        f"环境变量: MODEL_SPEECH_APP_ID={'已设置' if os.environ.get('MODEL_SPEECH_APP_ID') else '未设置'}, "
        f"MODEL_SPEECH_API_KEY={'已设置' if os.environ.get('MODEL_SPEECH_API_KEY') else '未设置'}, "
        f"FEISHU_TENANT_TOKEN={'已设置' if os.environ.get('FEISHU_TENANT_TOKEN') else '未设置'}"
    )

    if args.file:
        if not os.path.exists(args.file):
            fail(f"音频文件不存在: {args.file}")
        file_size = os.path.getsize(args.file)
        log(f"音频文件: {args.file} ({file_size} bytes)")
        if file_size == 0:
            fail(f"音频文件为空: {args.file}")

    local_file = None
    try:
        if args.file_key:
            feishu_token = args.feishu_token or os.environ.get(
                "FEISHU_TENANT_TOKEN", ""
            )
            if not feishu_token:
                fail(
                    "使用 --file-key 时需要飞书 token。"
                    "请通过 --feishu-token 参数或环境变量 FEISHU_TENANT_TOKEN 提供"
                )
            local_file = download_feishu_audio(args.file_key, feishu_token)

        text = recognize(
            audio_url=args.url,
            file_path=args.file or local_file,
            appid=args.appid,
            token=args.token,
            language=args.language,
        )

        print(text)

    except PermissionError as e:
        fail(str(e))
    except SystemExit:
        raise
    except Exception as e:
        fail(str(e))
    finally:
        if local_file and os.path.exists(local_file):
            os.unlink(local_file)


if __name__ == "__main__":
    main()

# -*- coding: utf-8 -*-
# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates.
#
# 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.

"""
asr_standard.py — 火山引擎 BigModel ASR 标准版（录音文件识别）。

异步 submit + poll 模式，支持 ≤5 小时音频。
参考: auc_http_demo.py (submit_task → query_task → result)
鉴权: 新版控制台 API Key 方案 https://www.volcengine.com/docs/6561/2119699

用法:
  # URL 识别（最常用）
  python3 asr_standard.py --url "https://example.com/audio.mp3"

  # 本地文件识别
  python3 asr_standard.py --file "/path/to/long_audio.wav"

  # 仅提交任务（不轮询）
  python3 asr_standard.py --url "https://..." --no-poll

  # 查询已有任务
  python3 asr_standard.py --query-task-id <TASK_ID> --query-logid <X_TT_LOGID>
"""

from __future__ import annotations

import argparse
import base64
import json
import os
import sys
import time
import uuid

_SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
if _SCRIPT_DIR not in sys.path:
    sys.path.insert(0, _SCRIPT_DIR)

from api_key import get_speech_api_key

# ============================================================
# 端点与资源 ID 配置
# ============================================================
ASR_STANDARD_SUBMIT_URL = os.getenv(
    "MODEL_SPEECH_ASR_STANDARD_SUBMIT_URL",
    "https://openspeech.bytedance.com/api/v3/auc/bigmodel/submit",
)
ASR_STANDARD_QUERY_URL = os.getenv(
    "MODEL_SPEECH_ASR_STANDARD_QUERY_URL",
    "https://openspeech.bytedance.com/api/v3/auc/bigmodel/query",
)
ASR_STANDARD_RESOURCE_ID = os.getenv(
    "MODEL_SPEECH_ASR_STANDARD_RESOURCE_ID", "volc.bigasr.auc"
)

# 轮询参数
DEFAULT_POLL_INTERVAL = 3       # 秒
DEFAULT_POLL_MAX_TIME = 10800   # 3 小时

# 标准版最大支持时长
STANDARD_MAX_SECONDS = 5 * 60 * 60  # 5 小时


def log(msg: str) -> None:
    print(f"[byted-asr-standard] {msg}", file=sys.stderr)


def fail_json(error_code: str, message: str, **extra) -> None:
    """输出 JSON 格式的错误并退出。"""
    payload = {"error": error_code, "message": message}
    payload.update(extra)
    print(json.dumps(payload, ensure_ascii=False, indent=2))
    sys.exit(1)


try:
    import requests
except ImportError:
    fail_json("MISSING_DEPENDENCY", "requests 库未安装，请执行 pip install requests")


# ============================================================
# 鉴权 Header 构建
# ============================================================
def _build_headers(appid: str | None = None, token: str | None = None) -> dict:
    """
    构建标准版请求头。

    鉴权方式（与极速版 asr_flash.go 保持一致）:
      - 新版控制台: X-Api-Key
      - 旧版: X-Api-App-Key + X-Api-Access-Key
    """
    app_id = appid or os.getenv("MODEL_SPEECH_APP_ID", "").strip()
    api_key = token or get_speech_api_key().strip()

    if not api_key:
        fail_json(
            "CREDENTIALS_NOT_CONFIGURED",
            "MODEL_SPEECH_API_KEY 需在环境变量中配置。"
            "见 https://www.volcengine.com/docs/6561/2119699",
            missing_credentials=["MODEL_SPEECH_API_KEY"],
        )

    headers = {
        "X-Api-Resource-Id": ASR_STANDARD_RESOURCE_ID,
        "X-Api-Request-Id": str(uuid.uuid4()),
        "X-Api-Sequence": "-1",
    }

    if app_id:
        headers["X-Api-App-Key"] = app_id
        headers["X-Api-Access-Key"] = api_key
    else:
        headers["X-Api-Key"] = api_key

    return headers


# ============================================================
# 文件 → Base64
# ============================================================
def file_to_base64(file_path: str) -> str:
    with open(file_path, "rb") as f:
        return base64.b64encode(f.read()).decode("utf-8")


# ============================================================
# Submit Task
# ============================================================
def submit_task(
    audio_url: str | None = None,
    file_path: str | None = None,
    appid: str | None = None,
    token: str | None = None,
    language: str | None = None,
) -> dict:
    """
    向火山标准版提交识别任务。

    Args:
        audio_url: 音频 URL（与 file_path 二选一）。
        file_path: 本地文件路径（会 base64 上传）。
        appid: App ID (可选)。
        token: API Key (可选)。
        language: 语言代码 (可选)。

    Returns:
        {"task_id": str, "x_tt_logid": str}
    """
    headers = _build_headers(appid=appid, token=token)
    task_id = headers["X-Api-Request-Id"]

    audio_payload: dict = {}
    if audio_url:
        audio_payload["url"] = audio_url
    elif file_path:
        if not os.path.isfile(file_path):
            fail_json("FILE_NOT_FOUND", f"音频文件不存在: {file_path}")
        if os.path.getsize(file_path) == 0:
            fail_json("FILE_EMPTY", f"音频文件为空: {file_path}")
        audio_payload["data"] = file_to_base64(file_path)
    else:
        fail_json("NO_INPUT", "必须提供 --url 或 --file")

    if language:
        audio_payload["language"] = language

    request_body = {
        "user": {"uid": headers.get("X-Api-App-Key", "skill_asr_user")},
        "audio": audio_payload,
        "request": {
            "model_name": "bigmodel",
            "enable_itn": True,
            "enable_punc": True,
            "show_utterances": True,
        },
    }

    log(f"正在提交标准版识别任务: task_id={task_id}")
    try:
        resp = requests.post(
            ASR_STANDARD_SUBMIT_URL,
            data=json.dumps(request_body),
            headers=headers,
            timeout=60,
        )
    except requests.RequestException as e:
        fail_json("NETWORK_ERROR", f"提交请求网络异常: {e}")

    resp_status = resp.headers.get("X-Api-Status-Code", "")
    x_tt_logid = resp.headers.get("X-Tt-Logid", "")

    if resp_status != "20000000":
        msg = resp.headers.get("X-Api-Message", "未知错误")
        fail_json(
            "SUBMIT_FAILED",
            f"标准版任务提交失败: code={resp_status}, msg={msg}",
            task_id=task_id,
            logid=x_tt_logid,
        )

    log(f"任务已提交: task_id={task_id}, logid={x_tt_logid}")
    return {"task_id": task_id, "x_tt_logid": x_tt_logid}


# ============================================================
# Query Task
# ============================================================
def query_task(
    task_id: str,
    x_tt_logid: str,
    appid: str | None = None,
    token: str | None = None,
) -> dict:
    """
    查询标准版识别任务状态。

    Returns:
        {"status_code": str, "message": str, "logid": str, "body": dict}
    """
    headers = _build_headers(appid=appid, token=token)
    headers["X-Api-Request-Id"] = task_id
    headers["X-Tt-Logid"] = x_tt_logid

    try:
        resp = requests.post(
            ASR_STANDARD_QUERY_URL,
            data=json.dumps({}),
            headers=headers,
            timeout=30,
        )
    except requests.RequestException as e:
        return {
            "status_code": "NETWORK_ERROR",
            "message": str(e),
            "logid": "",
            "body": {},
        }

    return {
        "status_code": resp.headers.get("X-Api-Status-Code", ""),
        "message": resp.headers.get("X-Api-Message", ""),
        "logid": resp.headers.get("X-Tt-Logid", ""),
        "body": resp.json() if resp.text.strip() else {},
    }


# ============================================================
# Poll until done
# ============================================================
def poll_until_done(
    task_id: str,
    x_tt_logid: str,
    appid: str | None = None,
    token: str | None = None,
    poll_interval: int = DEFAULT_POLL_INTERVAL,
    poll_max_time: int = DEFAULT_POLL_MAX_TIME,
) -> dict:
    """
    轮询标准版识别任务直到完成或超时。

    Returns:
        ASR 结果 JSON (dict)。
    """
    start = time.time()
    while True:
        elapsed = time.time() - start
        if elapsed > poll_max_time:
            fail_json(
                "POLL_TIMEOUT",
                f"标准版识别任务超时: task_id={task_id}, "
                f"已等待 {int(elapsed)} 秒（上限 {poll_max_time} 秒）",
                task_id=task_id,
            )

        qr = query_task(task_id, x_tt_logid, appid=appid, token=token)
        code = qr["status_code"]

        if code == "20000000":
            log(f"识别完成: task_id={task_id}")
            return qr["body"]
        elif code in ("20000001", "20000002"):
            log(
                f"任务进行中: code={code}, "
                f"已等待 {int(elapsed)}s, {poll_interval}s 后重试..."
            )
            time.sleep(poll_interval)
        else:
            fail_json(
                "TASK_FAILED",
                f"标准版识别失败: code={code}, msg={qr['message']}",
                task_id=task_id,
                logid=qr["logid"],
            )


# ============================================================
# 文本提取 (统一格式)
# ============================================================
def extract_text(result: dict) -> str:
    """
    从火山 ASR 标准版响应 JSON 中提取纯文本。

    兼容多种返回格式:
      - result.text
      - result.utterances[].text
      - utterances[].text (顶层)
      - result[] (列表形式)
    """
    text_parts: list[str] = []

    if "result" in result:
        res = result["result"]
        if isinstance(res, list):
            for item in res:
                if isinstance(item, dict) and "text" in item:
                    text_parts.append(item["text"])
        elif isinstance(res, dict):
            if "text" in res:
                text_parts.append(res["text"])
            elif "utterances" in res:
                for utt in res["utterances"]:
                    if "text" in utt:
                        text_parts.append(utt["text"])

    if not text_parts and "utterances" in result:
        for utt in result["utterances"]:
            if "text" in utt:
                text_parts.append(utt["text"])

    return "".join(text_parts)


# ============================================================
# CLI 主流程
# ============================================================
def main() -> None:
    parser = argparse.ArgumentParser(
        description="火山引擎 BigModel ASR 标准版（录音文件识别，异步 submit+poll）",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""\
示例:
  # URL 识别
  python3 asr_standard.py --url "https://example.com/audio.mp3"

  # 本地文件识别
  python3 asr_standard.py --file "/path/to/long_audio.wav"

  # 仅提交，不轮询
  python3 asr_standard.py --url "https://..." --no-poll

  # 查询已有任务
  python3 asr_standard.py --query-task-id <ID> --query-logid <LOGID>
        """,
    )

    input_group = parser.add_mutually_exclusive_group()
    input_group.add_argument("--url", help="音频文件的 URL 地址")
    input_group.add_argument("--file", help="本地音频文件路径")

    parser.add_argument(
        "--appid",
        help="火山引擎 ASR App ID (也可通过环境变量 MODEL_SPEECH_APP_ID 设置)",
    )
    parser.add_argument(
        "--token",
        help="火山引擎 ASR API Key (也可通过环境变量 MODEL_SPEECH_API_KEY 设置)",
    )
    parser.add_argument(
        "--language",
        help="语言代码，如 zh-CN, ja-JP, en-US (可选，不传则自动识别)",
    )
    parser.add_argument(
        "--no-poll",
        action="store_true",
        help="仅提交任务，不轮询结果（返回 task_id 供后续查询）",
    )
    parser.add_argument(
        "--poll-interval",
        type=int,
        default=DEFAULT_POLL_INTERVAL,
        help=f"轮询间隔秒数 (默认: {DEFAULT_POLL_INTERVAL})",
    )
    parser.add_argument(
        "--poll-max-time",
        type=int,
        default=DEFAULT_POLL_MAX_TIME,
        help=f"最大轮询时间秒数 (默认: {DEFAULT_POLL_MAX_TIME})",
    )

    # 查询子功能
    parser.add_argument(
        "--query-task-id",
        help="查询已有任务的 task_id（与 --query-logid 配合使用）",
    )
    parser.add_argument(
        "--query-logid",
        default="",
        help="查询时传入的 X-Tt-Logid（可选）",
    )

    args = parser.parse_args()

    # 模式 1: 查询已有任务
    if args.query_task_id:
        log(f"查询任务: task_id={args.query_task_id}")
        result = poll_until_done(
            task_id=args.query_task_id,
            x_tt_logid=args.query_logid,
            appid=args.appid,
            token=args.token,
            poll_interval=args.poll_interval,
            poll_max_time=args.poll_max_time,
        )
        text = extract_text(result)
        if text:
            print(text)
        else:
            print(json.dumps(result, ensure_ascii=False, indent=2))
        return

    # 模式 2: 提交新任务
    if not args.url and not args.file:
        fail_json("NO_INPUT", "必须提供 --url 或 --file（或 --query-task-id 查询已有任务）")

    submit_result = submit_task(
        audio_url=args.url,
        file_path=args.file,
        appid=args.appid,
        token=args.token,
        language=args.language,
    )

    task_id = submit_result["task_id"]
    x_tt_logid = submit_result["x_tt_logid"]

    if args.no_poll:
        print(json.dumps({
            "task_id": task_id,
            "x_tt_logid": x_tt_logid,
            "status": "submitted",
            "message": "任务已提交，使用以下命令查询结果:",
            "query_command": (
                f"python3 asr_standard.py "
                f"--query-task-id {task_id} --query-logid {x_tt_logid}"
            ),
        }, ensure_ascii=False, indent=2))
        return

    # 轮询直到完成
    result = poll_until_done(
        task_id=task_id,
        x_tt_logid=x_tt_logid,
        appid=args.appid,
        token=args.token,
        poll_interval=args.poll_interval,
        poll_max_time=args.poll_max_time,
    )
    text = extract_text(result)
    if text:
        log(f"识别成功，文本长度: {len(text)} 字符")
        print(text)
    else:
        log("警告: 未提取到文本，输出原始 JSON")
        print(json.dumps(result, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()

# -*- coding: utf-8 -*-
"""
ensure_ffmpeg.py — 自动检测并安装 ffmpeg / ffprobe。

用法:
  # 仅检测，输出安装计划 (JSON)
  python3 ensure_ffmpeg.py

  # 检测 + 自动安装
  python3 ensure_ffmpeg.py --execute

返回 JSON:
  status: "already_available" | "installable" | "installed" | "failed" | "blocked"
  ffmpeg_path / ffprobe_path: 可执行文件路径 (成功时)
  commands: 将要/已经执行的安装命令
  source_policy: 始终为 "package_manager_only"（不从 GitHub/npm 下载）

"""

import json
import os
import platform
import shutil
import subprocess
import sys


def print_json(payload, exit_code=0):
    print(json.dumps(payload, ensure_ascii=False, indent=2))
    if exit_code:
        sys.exit(exit_code)


def has_command(name):
    return shutil.which(name) is not None


def as_command_text(command):
    if isinstance(command, dict):
        return " ".join(command["command"])
    return " ".join(command)


def run_command(command):
    return subprocess.run(
        command,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        universal_newlines=True,
    )


def linux_privilege_prefix():
    if os.name == "nt":
        return []
    if hasattr(os, "geteuid") and os.geteuid() == 0:
        return []
    if has_command("sudo"):
        return ["sudo"]
    return None


def get_rhel_major_version():
    if not has_command("rpm"):
        return None
    result = run_command(["rpm", "-E", "%rhel"])
    if result.returncode != 0:
        return None
    version = result.stdout.strip()
    return version or None


def dnf_yum_repo_fallback(package_manager, prefix):
    rhel_major = get_rhel_major_version()
    if not rhel_major:
        return None

    commands = [
        {
            "command": prefix + [package_manager, "install", "-y", "epel-release"],
            "optional": True,
        },
    ]

    if has_command(package_manager):
        commands.extend([
            {
                "command": prefix + [package_manager, "config-manager", "--set-enabled", "crb"],
                "optional": True,
            },
            {
                "command": prefix + [package_manager, "config-manager", "--set-enabled", "powertools"],
                "optional": True,
            },
        ])

    rpmfusion_url = (
        "https://mirrors.rpmfusion.org/free/el/"
        f"rpmfusion-free-release-{rhel_major}.noarch.rpm"
    )
    commands.append({
        "command": prefix + [package_manager, "install", "-y", rpmfusion_url],
        "optional": False,
    })
    commands.append({
        "command": prefix + [package_manager, "install", "-y", "ffmpeg"],
        "optional": False,
    })

    return {
        "reason": "ffmpeg_not_in_enabled_repos",
        "rhel_major": rhel_major,
        "commands": commands,
    }


def command_failed_for_missing_ffmpeg(result):
    text = "\n".join([result.stdout or "", result.stderr or ""]).lower()
    patterns = [
        "no match for argument: ffmpeg",
        "unable to find a match: ffmpeg",
        "no package ffmpeg available",
        "nothing provides ffmpeg",
    ]
    return any(pattern in text for pattern in patterns)


def build_install_plan():
    ffmpeg_path = shutil.which("ffmpeg")
    ffprobe_path = shutil.which("ffprobe")
    if ffmpeg_path and ffprobe_path:
        return {
            "status": "already_available",
            "platform": platform.system(),
            "ffmpeg_path": ffmpeg_path,
            "ffprobe_path": ffprobe_path,
            "source_policy": "package_manager_only",
        }

    system = platform.system()

    if system == "Darwin":
        if has_command("brew"):
            return {
                "status": "installable",
                "platform": system,
                "package_manager": "brew",
                "source_policy": "package_manager_only",
                "avoid": ["github-direct", "npm", "manual-zip-download"],
                "commands": [["brew", "install", "ffmpeg"]],
            }
        return {
            "status": "blocked",
            "platform": system,
            "reason": "homebrew_not_found",
            "message": (
                "No supported local package manager was found. "
                "Please install Homebrew first: https://brew.sh"
            ),
            "source_policy": "package_manager_only",
        }

    if system == "Windows":
        if has_command("winget"):
            return {
                "status": "installable",
                "platform": system,
                "package_manager": "winget",
                "source_policy": "package_manager_only",
                "avoid": ["github-direct", "npm", "manual-zip-download"],
                "commands": [[
                    "winget", "install", "--id", "Gyan.FFmpeg", "-e",
                    "--accept-source-agreements", "--accept-package-agreements",
                ]],
            }
        if has_command("choco"):
            return {
                "status": "installable",
                "platform": system,
                "package_manager": "choco",
                "source_policy": "package_manager_only",
                "avoid": ["github-direct", "npm", "manual-zip-download"],
                "commands": [["choco", "install", "ffmpeg", "-y"]],
            }
        return {
            "status": "blocked",
            "platform": system,
            "reason": "package_manager_not_found",
            "message": (
                "No supported Windows package manager was found. "
                "Please install winget or choco first."
            ),
            "source_policy": "package_manager_only",
        }

    # Linux
    prefix = linux_privilege_prefix()
    if has_command("apt-get") and prefix is not None:
        return {
            "status": "installable",
            "platform": system,
            "package_manager": "apt-get",
            "source_policy": "package_manager_only",
            "avoid": ["github-direct", "npm", "manual-zip-download"],
            "commands": [
                prefix + ["apt-get", "update"],
                prefix + ["apt-get", "install", "-y", "ffmpeg"],
            ],
        }
    if has_command("dnf") and prefix is not None:
        return {
            "status": "installable",
            "platform": system,
            "package_manager": "dnf",
            "source_policy": "package_manager_only",
            "avoid": ["github-direct", "npm", "manual-zip-download"],
            "commands": [prefix + ["dnf", "install", "-y", "ffmpeg"]],
            "repo_fallback": dnf_yum_repo_fallback("dnf", prefix),
        }
    if has_command("yum") and prefix is not None:
        return {
            "status": "installable",
            "platform": system,
            "package_manager": "yum",
            "source_policy": "package_manager_only",
            "avoid": ["github-direct", "npm", "manual-zip-download"],
            "commands": [prefix + ["yum", "install", "-y", "ffmpeg"]],
            "repo_fallback": dnf_yum_repo_fallback("yum", prefix),
        }
    if has_command("zypper") and prefix is not None:
        return {
            "status": "installable",
            "platform": system,
            "package_manager": "zypper",
            "source_policy": "package_manager_only",
            "avoid": ["github-direct", "npm", "manual-zip-download"],
            "commands": [prefix + ["zypper", "--non-interactive", "install", "ffmpeg"]],
        }

    return {
        "status": "blocked",
        "platform": system,
        "reason": "package_manager_not_found_or_no_privilege_path",
        "message": (
            "No supported package manager path is available for autonomous installation. "
            "Please install ffmpeg manually."
        ),
        "source_policy": "package_manager_only",
    }


def execute_step(command_spec, outputs):
    optional = False
    command = command_spec
    if isinstance(command_spec, dict):
        optional = command_spec.get("optional", False)
        command = command_spec["command"]

    result = run_command(command)
    outputs.append({
        "command": as_command_text(command_spec),
        "returncode": result.returncode,
        "stdout": result.stdout[-4000:],
        "stderr": result.stderr[-4000:],
        "optional": optional,
    })
    return result, optional


def execute_repo_fallback(plan, outputs):
    fallback = plan.get("repo_fallback")
    if not fallback:
        return None

    for command_spec in fallback.get("commands", []):
        result, optional = execute_step(command_spec, outputs)
        if result.returncode != 0 and not optional:
            return {
                "status": "failed",
                "platform": plan.get("platform"),
                "package_manager": plan.get("package_manager"),
                "source_policy": plan.get("source_policy"),
                "message": (
                    "Tried enabling EPEL / RPM Fusion automatically, "
                    "but ffmpeg is still unavailable."
                ),
                "repo_fallback_attempted": True,
                "repo_fallback_reason": fallback.get("reason"),
                "steps": outputs,
            }
    return None


def execute_plan(plan):
    steps = plan.get("commands", [])
    outputs = []
    for index, command in enumerate(steps):
        result, optional = execute_step(command, outputs)
        if result.returncode != 0:
            if (
                not optional
                and plan.get("package_manager") in {"dnf", "yum"}
                and index == len(steps) - 1
                and command_failed_for_missing_ffmpeg(result)
            ):
                fallback_result = execute_repo_fallback(plan, outputs)
                if fallback_result is not None:
                    return fallback_result
                break
            return {
                "status": "failed",
                "platform": plan.get("platform"),
                "package_manager": plan.get("package_manager"),
                "source_policy": plan.get("source_policy"),
                "repo_fallback_available": bool(plan.get("repo_fallback")),
                "steps": outputs,
            }

    ffmpeg_path = shutil.which("ffmpeg")
    ffprobe_path = shutil.which("ffprobe")
    if ffmpeg_path and ffprobe_path:
        return {
            "status": "installed",
            "platform": plan.get("platform"),
            "package_manager": plan.get("package_manager"),
            "source_policy": plan.get("source_policy"),
            "ffmpeg_path": ffmpeg_path,
            "ffprobe_path": ffprobe_path,
            "steps": outputs,
        }

    return {
        "status": "failed",
        "platform": plan.get("platform"),
        "package_manager": plan.get("package_manager"),
        "source_policy": plan.get("source_policy"),
        "message": "Installation commands finished but ffmpeg/ffprobe are still unavailable.",
        "repo_fallback_available": bool(plan.get("repo_fallback")),
        "steps": outputs,
    }


def main():
    execute = "--execute" in sys.argv[1:]

    plan = build_install_plan()
    if not execute:
        if plan.get("status") == "installable":
            plan["commands_text"] = [as_command_text(cmd) for cmd in plan["commands"]]
        print_json(plan, exit_code=0 if plan.get("status") != "blocked" else 1)
        return

    if plan.get("status") == "already_available":
        print_json(plan)
        return
    if plan.get("status") != "installable":
        print_json(plan, exit_code=1)
        return

    result = execute_plan(plan)
    print_json(result, exit_code=0 if result.get("status") == "installed" else 1)


if __name__ == "__main__":
    main()

# -*- coding: utf-8 -*-
"""
inspect_audio.py — 探测音频文件的时长、采样率、声道数等元信息。

用法:
  python3 inspect_audio.py <audio_file_or_url>

探测优先级:
  1. ffprobe（最全面，支持本地文件和 URL）
  2. macOS afinfo（仅本地文件，仅 macOS）
  3. Python wave 模块（仅 .wav 文件）

返回 JSON:
  成功: {duration_seconds, sample_rate, channels, codec_name, container_format, probe_tool}
  失败: {error, message, recommended_next_step?}

"""

import json
import os
import platform
import re
import subprocess
import sys
import wave


FFPROBE_URL_TIMEOUT_SECONDS = 15


def print_json(payload, exit_code=0):
    print(json.dumps(payload, ensure_ascii=False, indent=2))
    if exit_code:
        sys.exit(exit_code)


def is_http_url(input_value):
    return input_value.startswith("http://") or input_value.startswith("https://")


def run_command(command, timeout_seconds=None):
    try:
        return subprocess.run(
            command,
            check=True,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            universal_newlines=True,
            timeout=timeout_seconds,
        )
    except FileNotFoundError:
        return None
    except subprocess.CalledProcessError as e:
        return e
    except subprocess.TimeoutExpired as e:
        return e


def inspect_with_ffprobe(input_value):
    """使用 ffprobe 探测音频文件元信息。"""
    timeout_seconds = FFPROBE_URL_TIMEOUT_SECONDS if is_http_url(input_value) else None
    command = [
        "ffprobe", "-v", "error",
        "-print_format", "json",
        "-show_streams", "-show_format",
        input_value,
    ]
    result = run_command(command, timeout_seconds=timeout_seconds)
    if result is None:
        return None
    if isinstance(result, subprocess.CalledProcessError):
        error_result = {
            "error": "FFPROBE_FAILED",
            "message": f"ffprobe execution failed. stderr: {result.stderr.strip()}",
            "probe_tool": "ffprobe",
        }
        if is_http_url(input_value):
            error_result["recommended_next_step"] = (
                "Verify that the URL is publicly reachable. If the URL is known-good and you plan "
                "to use standard async recognition, you may still try asr_standard.py directly."
            )
        return error_result
    if isinstance(result, subprocess.TimeoutExpired):
        error_result = {
            "error": "FFPROBE_TIMEOUT",
            "message": f"ffprobe timed out after {FFPROBE_URL_TIMEOUT_SECONDS} seconds.",
            "probe_tool": "ffprobe",
            "recommended_next_step": (
                "If the input is already a public URL and you plan to use standard async recognition, "
                "you may call asr_standard.py directly instead of blocking on local probing."
            ),
        }
        if is_http_url(input_value):
            error_result["direct_url_async_allowed"] = True
        return error_result

    payload = json.loads(result.stdout)
    streams = payload.get("streams", [])
    audio_stream = next(
        (stream for stream in streams if stream.get("codec_type") == "audio"), None
    )
    if not audio_stream:
        return {
            "error": "NO_AUDIO_STREAM",
            "message": "No audio stream found in input.",
            "probe_tool": "ffprobe",
        }

    format_info = payload.get("format", {})
    duration = audio_stream.get("duration") or format_info.get("duration")
    sample_rate = audio_stream.get("sample_rate")
    channels = audio_stream.get("channels")
    file_size = format_info.get("size")

    inspected = {
        "duration_seconds": float(duration) if duration is not None else None,
        "sample_rate": int(sample_rate) if sample_rate else None,
        "channels": int(channels) if channels else None,
        "codec_name": audio_stream.get("codec_name"),
        "container_format": format_info.get("format_name"),
        "file_size_bytes": int(file_size) if file_size else None,
        "probe_tool": "ffprobe",
    }
    return inspected


def inspect_with_afinfo(input_value):
    """macOS 专用：通过 afinfo 探测音频文件。"""
    if platform.system() != "Darwin" or not os.path.isfile(input_value):
        return None

    result = run_command(["afinfo", input_value])
    if result is None or isinstance(
        result, (subprocess.CalledProcessError, subprocess.TimeoutExpired)
    ):
        return None

    stdout = result.stdout
    duration_match = re.search(r"estimated duration:\s*([0-9.]+)", stdout)
    sample_rate_match = re.search(r"([0-9.]+)\s*Hz", stdout)
    channels_match = re.search(r"([0-9]+)\s*channel", stdout)

    inspected = {
        "duration_seconds": float(duration_match.group(1)) if duration_match else None,
        "sample_rate": int(float(sample_rate_match.group(1))) if sample_rate_match else None,
        "channels": int(channels_match.group(1)) if channels_match else None,
        "codec_name": None,
        "container_format": os.path.splitext(input_value)[1].lstrip(".").lower() or None,
        "probe_tool": "afinfo",
    }
    return inspected


def inspect_wav_with_wave(input_value):
    """通过 Python 标准库 wave 模块探测 .wav 文件。"""
    if not os.path.isfile(input_value) or not input_value.lower().endswith(".wav"):
        return None

    try:
        with wave.open(input_value, "rb") as wav_file:
            frames = wav_file.getnframes()
            sample_rate = wav_file.getframerate()
            channels = wav_file.getnchannels()
    except wave.Error:
        return None

    inspected = {
        "duration_seconds": frames / float(sample_rate) if sample_rate else None,
        "sample_rate": sample_rate,
        "channels": channels,
        "codec_name": "pcm",
        "container_format": "wav",
        "probe_tool": "python-wave",
    }
    return inspected


def main():
    if len(sys.argv) != 2:
        print_json(
            {
                "error": "NO_INPUT",
                "message": "Usage: python3 inspect_audio.py <audio_file_or_url>",
            },
            exit_code=1,
        )

    input_value = sys.argv[1]

    # 如果是本地文件，补充文件大小
    file_size_bytes = None
    if os.path.isfile(input_value):
        file_size_bytes = os.path.getsize(input_value)

    inspected = inspect_with_ffprobe(input_value)
    if inspected:
        if file_size_bytes and not inspected.get("file_size_bytes"):
            inspected["file_size_bytes"] = file_size_bytes
        exit_code = 1 if inspected.get("error") else 0
        print_json(inspected, exit_code=exit_code)

    inspected = inspect_with_afinfo(input_value)
    if inspected:
        if file_size_bytes:
            inspected["file_size_bytes"] = file_size_bytes
        print_json(inspected)

    inspected = inspect_wav_with_wave(input_value)
    if inspected:
        if file_size_bytes:
            inspected["file_size_bytes"] = file_size_bytes
        print_json(inspected)

    print_json(
        {
            "error": "REQUIRES_FFPROBE",
            "message": (
                "Unable to reliably inspect duration, sample rate, and channels "
                "with available built-in tools."
            ),
            "file_size_bytes": file_size_bytes,
            "recommended_next_step": (
                "Run ensure_ffmpeg.py --execute first, then rerun inspect_audio.py. "
                "If the input is already a public URL and you plan to use standard async "
                "recognition, you may call asr_standard.py directly."
            ),
            "direct_url_async_allowed": is_http_url(input_value),
        },
        exit_code=1,
    )


if __name__ == "__main__":
    main()