dbs-video-extract

interface:
  display_name: "dbs-video-extract"
  short_description: "引导开通所需服务并提取短视频数据、语音文字稿与 Markdown"
  default_prompt: "使用 $dbs-video-extract，检查 API 配置并提取这个短视频的数据和文字稿。"

# API 购买、配置与连接测试

只有用户缺少当前任务所需的 API Key 时读取本文件。

## 两个 API 分别解决什么

先向第一次使用的用户解释：

> 想让这个 Skill 工作，需要先开通外部数据服务。API Key 是服务商在开通后提供的一串使用凭证，可以理解为这个 Skill 调用服务的专用通行证。你不需要理解技术原理，也不要把它发到聊天中。

| 服务 | 解决的问题 | 单独配置后的能力 |
| --- | --- | --- |
| TikHub | 查询抖音、小红书和微信视频号的作品／账号、统计与媒体元信息 | 可以取得数据，不能生成轻抖语音文字稿 |
| 轻抖 | 提取抖音、小红书等短视频的语音文字稿 | 可以生成文字稿，不能查询 TikHub 作品／账号数据 |

两个 Key 都没有时，明确告诉用户需要购买并配置至少一个 API，否则当前 Skill 无法使用。用户只购买一个时允许继续，同时说明只能获得对应的一半能力。

## 先让用户选择结果

不要先抛出两个链接。用结果帮助用户选择：

1. **完整功能**：同时获得视频数据和语音文字稿，需要 TikHub 与轻抖；
2. **只看数据**：只看作者、标题、点赞、评论等信息，只需要 TikHub；
3. **只要文字稿**：只把视频口播转成文字，只需要轻抖。

用户没有明确选择、又要求「完整提取」时，按完整功能引导。只配置一个服务时不阻止使用，必须明确缺少的另一半能力。

用户只调用 Skill、尚未发送链接或说明目标时，先让用户回复：

```text
你想让它帮你得到什么？

1．完整功能：视频数据＋语音文字稿，需要开通 TikHub 和轻抖；
2．只看数据：作者、标题、点赞、评论等，只需要 TikHub；
3．只要文字稿：把视频口播变成文字，只需要轻抖。

回复 1、2 或 3，我再带你完成对应服务的开通和配置。
```

这一轮不展示充值链接。收到选择后，继续下面的法律提示、充值入口和购买步骤。用户已经发送视频链接或明确说要完整提取时，可以直接按完整功能继续。

## 充值提示与入口

每次向用户展示任一充值地址时，必须先原样展示以下提示，不能缩写、省略或改写：

> dbskill 是免费开源项目，与 TikHub、轻抖无隶属、合作或利益关系。充值由用户自主决定，相关交易、服务、争议及风险由用户自行承担；法律另有规定的除外，dbskill 不承担责任。

维护者确认的充值地址：

- TikHub：<https://user.tikhub.io/dashboard/add-credit>
- 轻抖：<https://www.qingdou.vip/voice-text-api>

只展示当前缺少的 API 对应地址。两个 Key 都没有时可以同时展示两个地址。不得使用搜索结果中的非官方代理、二手 Key 或无法核实的充值入口替代。

## 购买步骤

展示对应地址后，逐步告诉用户：

1. 点击链接，在服务商页面注册或登录；
2. 查看页面当前显示的套餐、额度、有效期和计费规则，自主决定是否充值；
3. 完成充值后，进入个人中心，寻找「API Key」「密钥管理」或「开发者设置」；
4. 页面要求创建凭证时自行创建，然后复制并暂存在本机；
5. 不要把凭证发送到聊天中。回到对话只需回复「TikHub 已充值」「轻抖已充值」或「两个都已充值」；
6. Agent 收到回复后进入下一节的安全配置流程。

服务商页面名称可能变化。用户找不到入口时，让用户发送已遮住账号、余额和凭证的截图，再根据页面内容引导。不能猜测具体按钮名称、套餐价格或到账时间。

## 购买引导的完整交付结构

用户选定结果后，回复必须依次包含：

1. 「你选择了什么、需要开通哪个服务」；
2. 固定法律提示；
3. 对应充值地址；
4. 注册／登录、查看计费规则、充值、寻找凭证的步骤；
5. 「不要把凭证发给我」；
6. 「充值后只回复哪个已充值」；
7. 下一轮将使用安全配置脚本保存并检查。

缺少以上任何一项都不算完成购买引导。

## 充值后的安全配置

默认使用 Skill 自带脚本。根据用户购买的服务，只运行对应命令：

```bash
python3 scripts/configure_api_key.py tikhub
python3 scripts/configure_api_key.py qingdou
```

脚本会要求用户在终端中粘贴凭证，输入过程不回显；随后写入 `~/.config/dbs/API_Keys.md` 并把文件权限收紧为仅当前用户可读写。Agent 不读取或复述凭证。

脚本无法运行时，再使用下面的手动方案。

用户也可以在自己的终端中设置环境变量，Key 不进入对话：

```bash
export TIKHUB_API_KEY="用户自己的 Key"
export QINGDOU_API_KEY="用户自己的 Key"
```

也可以把 Key 保存到 `~/.config/dbs/API_Keys.md`：

```markdown
## TikHub API
- **Key**: 用户自己的 Key

## 轻抖 API
- **Key**: 用户自己的 Key
```

macOS 用户可以使用钥匙串服务名：

- TikHub：`dbs-tikhub-api-key`；
- 轻抖：`dbs-qingdou-api-key`。

禁止让用户把完整 Key 粘贴进对话。禁止把 Key 写入 Skill、仓库、Markdown 交付物、命令参数或日志。

## 连接测试

配置后先检查凭证发现状态：

```bash
python3 scripts/extract_video.py --check-keys
```

发现 Key 只说明本地配置存在。执行第一个真实任务时再验证鉴权、余额、权限和接口可用性。遇到 `401`、`402`、`403`、`4100` 或资源不足时停止自动重试，并把服务商返回的原因转述给用户。

# 轻抖文案提取 API

仅在执行失败、需要理解接口兼容逻辑或维护脚本时读取。

## 限制

- 默认任务并发路数：1。
- 创建任务后需要轮询获取结果。

## 创建任务

- 请求：`POST https://www.qingdou.vip/web/api/commitGetTextTask`
- Header：`x-api-key: {API Key}`、`Content-Type: application/json`
- Body：`userInputList`

用户提供的文档出现两种 Body 形式：

```json
{"userInputList":"完整分享文案或视频链接"}
```

```json
{"userInputList":[{"numberIndex":0,"url":"完整分享文案或视频链接"}]}
```

脚本优先使用字符串形式；仅在业务状态 `4001` 时改用数组形式重试。

成功状态为 `status.code = 1001`。文档中的 `result` 既出现过数字 `batchId`，也出现过包含 `batchId` 的对象；脚本兼容两种形式。

## 获取结果

- 请求：`GET https://www.qingdou.vip/web/api/getTaskResult?batchId={batchId}`
- Header：`x-api-key: {API Key}`
- `batchStatus = 0` 或 `1`：处理中。
- `batchStatus = 2`：批任务结束。

成功条目包含：

- `originLink`：原始链接；
- `platformName`：平台；
- `awemeId`：视频 ID；
- `status`：条目状态；
- `videoTitle`：视频标题；
- `videoContent`：识别文字；
- `videoCover`：封面链接；
- `videoTime`：视频时长。

用户提供的返回样例没有作者字段。脚本兼容 `authorName`、`author_name`、`authorNickname`、`nickname`、`userName`、`accountName`，以及 `author`、`user`、`owner` 对象中的对应字段。没有明确字段时归入 `_未识别作者`。

## 状态码

### 接口状态

| 状态 | 含义 |
| --- | --- |
| `1001` | 请求成功 |
| `4001` | 参数错误 |
| `4100` | API Key 异常 |

### 条目状态

| 状态 | 含义 |
| --- | --- |
| `0` | 处理中 |
| `1000` | 文案提取成功 |
| `1001` | 获取视频唯一状态码失败 |
| `1010` | 链接获取失败 |
| `1011` | QPS 上限 |
| `1100` | 视频链接为空 |
| `1101` | 视频结果获取失败 |
| `1110` | 文案提取错误 |
| `1111` | 其他错误 |
| `1112` | 视频过大或视频不是 MP4 |
| `1113` | 用户资源不足 |
| `1114` | 视频时长获取失败或时长为 0 |
| `1115` | 获取链接信息失败 |
| `1116` | 预扣费异常 |
| `1117` | 获取视频链接异常 |
| `1118` | 超时失败 |

# TikHub API 接口参考

仅在新增平台查询、确认接口参数、处理错误或检查费用时读取。

## 基础配置

- 中国大陆：`https://api.tikhub.dev`
- 中国大陆以外：`https://api.tikhub.io`
- 鉴权请求头：`Authorization: Bearer API_KEY`
- 中文文档：<https://docs.tikhub.io/doc-4579297>
- Swagger：<https://api.tikhub.io>

所有 API 路径和参数在两个域名之间保持一致。

## MCP 配置

- 官方说明：<https://tikhub.io/mcp>
- Streamable HTTP：`https://mcp.tikhub.io/{platform}/mcp`
- 抖音平台：`https://mcp.tikhub.io/douyin/mcp`
- 小红书平台：`https://mcp.tikhub.io/xiaohongshu/mcp`
- 微信平台：`https://mcp.tikhub.io/wechat/mcp`
- 鉴权：`Authorization: Bearer API_KEY`
- 协议版本：`2024-11-05`
- 内容类型：`application/json`
- Accept：`application/json, text/event-stream`

MCP 调用顺序：

1. `initialize`；
2. 保存响应头 `Mcp-Session-Id`；
3. 使用同一鉴权头和 Session ID 调用 `tools/list` 或 `tools/call`。

`https://api.tikhub.dev/mcp` 和 `https://api.tikhub.io/mcp` 不提供 MCP 服务。REST API 返回的 `api_key_scopes` 中可能出现 `/mcp`，它表示权限范围，不能当成 MCP URL。

抖音常用 MCP 工具：

- `douyin_app_v3_handler_user_profile`：用 `sec_user_id` 查询用户资料；
- `douyin_app_v3_fetch_one_video_by_share_url`：用 `share_url` 查询作品；
- `douyin_web_fetch_one_video_by_share_url`：作品 App V3 失败时回退。

小红书视频笔记：

- `xiaohongshu_app_v2_get_video_note_detail`：参数使用 `share_text` 传入完整分享链接或文案；响应 `data.data` 的首条为目标笔记，后续条目可能是推荐内容。

微信视频号作品：

- `wechat_channels_v2_fetch_video_detail`：参数使用 `share_url`，并设置 `raw: false` 获取稳定字段摘要。

抖音分享短链接可能指向用户主页。先跟随公开跳转：

- `/share/user/` 或查询参数含 `sec_uid`：提取用户 ID，调用用户资料工具；
- `/video/`、`/note/` 或 `/slides/`：调用作品工具。

## 内置路由

### 当前账户信息

```text
GET /api/v1/tikhub/user/get_user_info
```

不需要查询参数。返回 `api_key_data` 和 `user_data`。脚本的 `account` 命令会删除邮箱字段后再输出。

### 抖音 App V3 分享链接解析

```text
GET /api/v1/douyin/app/v3/fetch_one_video_by_share_url
query: share_url=<抖音分享链接>
```

返回字段较丰富。内容受版权、删除、隐私或可见范围限制时，可能没有有效作品数据。

### 抖音 Web 分享链接解析

```text
GET /api/v1/douyin/web/fetch_one_video_by_share_url
query: share_url=<抖音分享链接>
```

Web 返回字段较少，视频画质通常更高。App V3 失败或没有作品数据时可回退 1 次。

### 小红书 App V2 视频笔记详情

```text
MCP xiaohongshu_app_v2_get_video_note_detail
arguments: share_text=<小红书分享链接或完整分享文案>
```

摘要保留笔记 ID、作者、标题、发布时间、时长、IP 属地、标签，以及点赞、收藏、评论、分享和播放字段。接口可能同时返回推荐笔记，当前只读取首条目标笔记。

### 微信视频号 Channels V2 作品详情

```text
MCP wechat_channels_v2_fetch_video_detail
arguments: share_url=<视频号分享链接>, raw=false
```

摘要保留作品 ID、作者、标题、发布时间、时长，以及阅读、点赞、收藏、评论和转发字段。

## 字幕与语音转写边界

TikHub 提供部分字幕接口：

- Bilibili Web：获取视频已有字幕信息；
- YouTube Web／Web V2：获取视频已有字幕，可输出 SRT、XML、JSON3 或纯文本。

YouTube Web V2 文档明确说明不会为无字幕视频执行 AI 语音转写。当前文档没有抖音或小红书通用 AI 语音转写端点。

## HTTP 状态码

- `400 Bad Request`：请求格式或参数错误；
- `401 Unauthorized`：Key 缺失、无效、未激活或过期；
- `402 Payment Required`：余额或免费额度不足；
- `403 Forbidden`：账户、邮箱验证、路由或 Key 权限问题；
- `404 Not Found`：路由或数据不存在；
- `429 Too Many Requests`：超过速率限制；
- `500 Internal Server Error`：TikHub 服务端错误。

出现 `401`、`402`、`403`、`429` 时停止自动重试。`500` 可以在用户仍需要该结果时进行 1 次人工确认后的重试。

默认单次数据查询 15 秒超时，不自动重试。TikHub 单侧超时或失败时，整合脚本继续执行轻抖文字稿步骤。

## 新增路由原则

1. 在官方文档中确认路由属于读取公开数据。
2. 记录路径、参数、返回结构、价格和限制。
3. 优先通过通用 `get` 验证一次真实响应。
4. 只有重复使用且字段稳定时，才为该路由增加专用子命令。
5. 不把完整 API Key、Cookie 或其他平台凭证写入参考文件。

#!/usr/bin/env python3
"""在本地隐蔽读取并保存 dbs-video-extract 的服务凭证。"""

from __future__ import annotations

import argparse
import getpass
import os
import re
import tempfile
from pathlib import Path


DEFAULT_CONFIG_PATH = Path.home() / ".config" / "dbs" / "API_Keys.md"
PROVIDERS = {
    "tikhub": {
        "display_name": "TikHub",
        "section": "TikHub API",
    },
    "qingdou": {
        "display_name": "轻抖",
        "section": "轻抖 API",
    },
}


class ConfigureError(RuntimeError):
    """可以安全展示给用户的配置错误。"""


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="安全保存 TikHub 或轻抖的 API 使用凭证。"
    )
    parser.add_argument("provider", choices=tuple(PROVIDERS))
    parser.add_argument(
        "--file",
        type=Path,
        default=DEFAULT_CONFIG_PATH,
        help="本地私密配置文件，默认 ~/.config/dbs/API_Keys.md",
    )
    return parser.parse_args()


def replace_section(text: str, section: str, secret: str) -> str:
    block = f"## {section}\n- **Key**: {secret}\n"
    pattern = re.compile(
        rf"(?ms)^##\s+{re.escape(section)}\s*$\n.*?(?=^##\s|\Z)"
    )
    if pattern.search(text):
        return pattern.sub(lambda _match: block, text, count=1).rstrip() + "\n"
    prefix = text.rstrip()
    return f"{prefix}\n\n{block}" if prefix else block


def write_private_config(path: Path, section: str, secret: str) -> None:
    target = path.expanduser().resolve()
    target.parent.mkdir(parents=True, exist_ok=True)
    existing = target.read_text(encoding="utf-8") if target.is_file() else ""
    updated = replace_section(existing, section, secret)
    descriptor, temporary_name = tempfile.mkstemp(
        prefix=f".{target.name}.", dir=target.parent
    )
    temporary = Path(temporary_name)
    try:
        with os.fdopen(descriptor, "w", encoding="utf-8") as handle:
            handle.write(updated)
        temporary.chmod(0o600)
        os.replace(temporary, target)
        target.chmod(0o600)
    except Exception:
        temporary.unlink(missing_ok=True)
        raise


def main() -> int:
    args = parse_args()
    provider = PROVIDERS[args.provider]
    display_name = provider["display_name"]
    print(
        f"请粘贴 {display_name} 提供的 API 使用凭证。输入不会显示在屏幕上，"
        "按回车保存。"
    )
    secret = getpass.getpass("使用凭证：").strip()
    if not secret:
        raise ConfigureError("没有收到使用凭证，未修改任何文件。")
    if any(character.isspace() for character in secret):
        raise ConfigureError("使用凭证中包含空格或换行，未修改任何文件。")
    try:
        write_private_config(args.file, provider["section"], secret)
    except OSError as error:
        raise ConfigureError(f"保存失败：{error}") from error
    print(
        f"{display_name} 使用凭证已保存到本地私密配置。"
        "现在可以运行 extract_video.py --check-keys 检查状态。"
    )
    return 0


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except ConfigureError as error:
        print(error)
        raise SystemExit(2) from None

#!/usr/bin/env python3
"""调用轻抖文案提取 API，并把短视频文字稿按作者和标题保存为 Markdown。"""

from __future__ import annotations

import argparse
import json
import os
import re
import subprocess
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from datetime import datetime, timezone
from pathlib import Path
from typing import Any


DEFAULT_BASE_URL = "https://www.qingdou.vip"
KEYCHAIN_SERVICE = "dbs-qingdou-api-key"
DEFAULT_API_KEYS_FILE = Path.home() / ".config" / "dbs" / "API_Keys.md"
SUCCESS_CODE = 1001
ITEM_STATUS = {
    0: "任务处理中",
    1000: "文案提取成功",
    1001: "获取视频唯一状态码失败",
    1010: "链接获取失败",
    1011: "QPS 上限",
    1100: "视频链接为空",
    1101: "视频结果获取失败",
    1110: "文案提取错误",
    1111: "其他错误",
    1112: "视频过大或视频不是 MP4",
    1113: "用户资源不足",
    1114: "视频时长获取失败或时长为 0",
    1115: "获取链接信息失败",
    1116: "预扣费异常",
    1117: "获取视频链接异常",
    1118: "超时失败",
}
AUTHOR_KEYS = (
    "authorName",
    "author_name",
    "authorNickname",
    "author_nickname",
    "nickname",
    "userName",
    "user_name",
    "accountName",
    "account_name",
)


class ExtractError(RuntimeError):
    """可向调用方展示的提取失败。"""


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="把抖音、小红书等短视频链接提取为 Markdown 文字稿。"
    )
    parser.add_argument("inputs", nargs="*", help="视频链接或完整分享文案")
    parser.add_argument(
        "--stdin", action="store_true", help="从标准输入读取一个完整分享文案"
    )
    parser.add_argument(
        "--input-file", type=Path, help="从 UTF-8 文件读取，每个非空行作为一个输入"
    )
    parser.add_argument(
        "--output-dir",
        type=Path,
        default=Path.cwd() / "短视频文字稿",
        help="Markdown 输出根目录",
    )
    parser.add_argument(
        "--poll-interval", type=float, default=2.0, help="轮询间隔秒数，默认 2"
    )
    parser.add_argument(
        "--timeout", type=float, default=900.0, help="每条输入最长等待秒数，默认 900"
    )
    parser.add_argument(
        "--overwrite", action="store_true", help="覆盖已经存在的同源 Markdown"
    )
    return parser.parse_args()


def find_api_key() -> str:
    value = os.environ.get("QINGDOU_API_KEY", "").strip()
    if value:
        return value
    configured_path = os.environ.get("QINGDOU_API_KEYS_FILE", "").strip()
    api_keys_path = Path(configured_path).expanduser() if configured_path else DEFAULT_API_KEYS_FILE
    if api_keys_path.is_file():
        try:
            text = api_keys_path.read_text(encoding="utf-8")
        except OSError:
            pass
        else:
            section = re.search(
                r"(?ms)^##\s+轻抖 API\s*$\n(.*?)(?=^##\s|\Z)", text
            )
            if section:
                key_line = re.search(
                    r"(?m)^-\s+\*\*Key\*\*:\s*(\S+)\s*$", section.group(1)
                )
                if key_line:
                    return key_line.group(1)
    if (
        sys.platform == "darwin"
        and os.environ.get("DBS_VIDEO_EXTRACT_DISABLE_KEYCHAIN") != "1"
    ):
        try:
            result = subprocess.run(
                ["security", "find-generic-password", "-s", KEYCHAIN_SERVICE, "-w"],
                check=True,
                capture_output=True,
                text=True,
            )
        except (FileNotFoundError, subprocess.CalledProcessError):
            pass
        else:
            value = result.stdout.strip()
            if value:
                return value
    return ""


def get_api_key() -> str:
    value = find_api_key()
    if value:
        return value
    raise ExtractError(
        "缺少轻抖 API Key：请在 API_Keys.md 中添加轻抖 API 条目，设置环境变量 "
        "QINGDOU_API_KEY，或在 macOS 钥匙串中创建服务名为 "
        f"{KEYCHAIN_SERVICE} 的通用密码。轻抖只负责语音文字稿；没有这个 Key "
        "时仍可单独使用 TikHub 查询作品或账号数据。"
    )


def request_json(
    method: str, url: str, api_key: str, payload: dict[str, Any] | None = None
) -> dict[str, Any]:
    body = None if payload is None else json.dumps(payload).encode("utf-8")
    request = urllib.request.Request(
        url,
        data=body,
        method=method,
        headers={
            "Accept": "application/json",
            "Content-Type": "application/json;charset=UTF-8",
            "User-Agent": "dbs-video-extract/1.0",
            "x-api-key": api_key,
        },
    )
    try:
        with urllib.request.urlopen(request, timeout=30) as response:
            raw = response.read().decode("utf-8")
    except urllib.error.HTTPError as error:
        detail = error.read().decode("utf-8", errors="replace")[:500]
        detail = detail.replace(api_key, "[REDACTED]")
        raise ExtractError(f"HTTP {error.code}：{detail}") from error
    except urllib.error.URLError as error:
        raise ExtractError(f"网络请求失败：{error.reason}") from error
    try:
        data = json.loads(raw)
    except json.JSONDecodeError as error:
        raise ExtractError("接口没有返回合法 JSON。") from error
    if not isinstance(data, dict):
        raise ExtractError("接口返回的 JSON 顶层不是对象。")
    return data


def business_status(data: dict[str, Any]) -> tuple[int | None, str]:
    status = data.get("status")
    if not isinstance(status, dict):
        return None, "接口缺少 status"
    code = status.get("code")
    try:
        normalized_code = int(code)
    except (TypeError, ValueError):
        normalized_code = None
    return normalized_code, str(status.get("msg") or "未知错误")


def extract_batch_id(data: dict[str, Any]) -> str:
    result = data.get("result")
    batch_id = result.get("batchId") if isinstance(result, dict) else result
    if batch_id is None or str(batch_id).strip() == "":
        raise ExtractError("创建任务成功，但返回结果中没有 batchId。")
    return str(batch_id)


def commit_task(base_url: str, api_key: str, user_input: str) -> str:
    url = f"{base_url.rstrip('/')}/web/api/commitGetTextTask"
    payloads = (
        {"userInputList": user_input},
        {"userInputList": [{"numberIndex": 0, "url": user_input}]},
    )
    for index, payload in enumerate(payloads):
        data = request_json("POST", url, api_key, payload)
        code, message = business_status(data)
        if code == SUCCESS_CODE:
            return extract_batch_id(data)
        if code != 4001 or index == len(payloads) - 1:
            raise ExtractError(f"创建任务失败：业务状态 {code}，{message}")
    raise ExtractError("创建任务失败。")


def poll_task(
    base_url: str,
    api_key: str,
    batch_id: str,
    poll_interval: float,
    timeout: float,
) -> dict[str, Any]:
    query = urllib.parse.urlencode({"batchId": batch_id})
    url = f"{base_url.rstrip('/')}/web/api/getTaskResult?{query}"
    deadline = time.monotonic() + timeout
    while True:
        data = request_json("GET", url, api_key)
        code, message = business_status(data)
        if code != SUCCESS_CODE:
            raise ExtractError(f"获取任务失败：业务状态 {code}，{message}")
        result = data.get("result")
        if not isinstance(result, dict):
            raise ExtractError("任务结果缺少 result 对象。")
        try:
            batch_status = int(result.get("batchStatus"))
        except (TypeError, ValueError):
            batch_status = None
        if batch_status == 2:
            return result
        if batch_status not in (0, 1):
            raise ExtractError(f"未知 batchStatus：{batch_status}")
        if time.monotonic() >= deadline:
            raise ExtractError("轮询超时。")
        remain = result.get("remainInterval")
        try:
            server_interval = float(remain)
        except (TypeError, ValueError):
            server_interval = 0.0
        wait_seconds = max(poll_interval, min(server_interval, 30.0))
        time.sleep(min(wait_seconds, max(0.0, deadline - time.monotonic())))


def first_text(mapping: dict[str, Any], keys: tuple[str, ...]) -> str:
    for key in keys:
        value = mapping.get(key)
        if isinstance(value, str) and value.strip():
            return value.strip()
    return ""


def find_author(item: dict[str, Any]) -> str:
    direct = first_text(item, AUTHOR_KEYS)
    if direct:
        return direct
    for parent_key in ("author", "user", "owner", "account"):
        nested = item.get(parent_key)
        if isinstance(nested, dict):
            value = first_text(nested, AUTHOR_KEYS + ("name",))
            if value:
                return value
    return ""


def safe_name(value: str, fallback: str, max_length: int = 100) -> str:
    cleaned = re.sub(r"[\x00-\x1f/\\:*?\"<>|]", " ", value)
    cleaned = re.sub(r"\s+", " ", cleaned).strip(" .")
    if not cleaned or cleaned in {".", ".."}:
        cleaned = fallback
    return cleaned[:max_length].rstrip(" .") or fallback


def yaml_string(value: Any) -> str:
    return json.dumps("" if value is None else str(value), ensure_ascii=False)


def normalize_transcript(value: Any) -> str:
    if not isinstance(value, str):
        return ""
    return re.sub(r"\s+", " ", value).strip()


def render_markdown(item: dict[str, Any], title: str, author: str) -> str:
    transcript = normalize_transcript(item.get("videoContent"))
    fields = {
        "title": title,
        "author": author or "未识别作者",
        "platform": item.get("platformName"),
        "source_url": item.get("originLink"),
        "video_id": item.get("awemeId") or item.get("videoId"),
        "duration_seconds": item.get("videoTime"),
        "cover_url": item.get("videoCover"),
        "extracted_at": datetime.now(timezone.utc).astimezone().isoformat(timespec="seconds"),
    }
    frontmatter = "\n".join(f"{key}: {yaml_string(value)}" for key, value in fields.items())
    return f"---\n{frontmatter}\n---\n\n# {title}\n\n{transcript}\n"


def source_url_in_file(path: Path, source_url: str) -> bool:
    try:
        head = path.read_text(encoding="utf-8")[:5000]
    except OSError:
        return False
    return f"source_url: {yaml_string(source_url)}" in head


def choose_output_path(
    output_dir: Path,
    author: str,
    title: str,
    video_id: str,
    source_url: str,
    overwrite: bool,
) -> tuple[Path, str]:
    folder_name = safe_name(author, "_未识别作者") if author else "_未识别作者"
    title_name = safe_name(title, f"未命名视频-{video_id or '未知ID'}")
    folder = output_dir.expanduser().resolve() / folder_name
    candidate = folder / f"{title_name}.md"
    if not candidate.exists():
        return candidate, "created"
    if source_url_in_file(candidate, source_url):
        return candidate, "overwritten" if overwrite else "skipped_existing"
    suffix = safe_name(video_id, "", 40)
    if suffix:
        candidate = folder / f"{title_name}-{suffix}.md"
        if not candidate.exists() or source_url_in_file(candidate, source_url):
            action = "overwritten" if candidate.exists() and overwrite else (
                "skipped_existing" if candidate.exists() else "created"
            )
            return candidate, action
    number = 2
    while True:
        candidate = folder / f"{title_name}-{number}.md"
        if not candidate.exists():
            return candidate, "created"
        if source_url_in_file(candidate, source_url):
            return candidate, "overwritten" if overwrite else "skipped_existing"
        number += 1


def save_item(item: dict[str, Any], output_dir: Path, overwrite: bool) -> dict[str, Any]:
    try:
        status = int(item.get("status"))
    except (TypeError, ValueError):
        status = None
    if status != 1000:
        return {
            "ok": False,
            "status": status,
            "error": ITEM_STATUS.get(status, "未知条目状态"),
            "source_url": str(item.get("originLink") or ""),
        }
    video_id = str(item.get("awemeId") or item.get("videoId") or "").strip()
    raw_title = str(item.get("videoTitle") or "").strip()
    title = raw_title or f"未命名视频-{video_id or '未知ID'}"
    author = find_author(item)
    source_url = str(item.get("originLink") or "").strip()
    transcript = normalize_transcript(item.get("videoContent"))
    if not transcript:
        return {
            "ok": False,
            "status": status,
            "error": "接口返回成功，但文字稿为空",
            "source_url": source_url,
        }
    path, action = choose_output_path(
        output_dir, author, title, video_id, source_url, overwrite
    )
    if action != "skipped_existing":
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(render_markdown(item, title, author), encoding="utf-8")
    return {
        "ok": True,
        "action": action,
        "path": str(path),
        "title": title,
        "author": author or None,
        "author_identified": bool(author),
        "source_url": source_url,
    }


def collect_inputs(args: argparse.Namespace) -> list[str]:
    inputs = [value.strip() for value in args.inputs if value.strip()]
    if args.stdin:
        value = sys.stdin.read().strip()
        if value:
            inputs.append(value)
    if args.input_file:
        content = args.input_file.expanduser().read_text(encoding="utf-8")
        inputs.extend(line.strip() for line in content.splitlines() if line.strip())
    if not inputs:
        raise ExtractError("没有收到视频链接或分享文案。")
    return inputs


def main() -> int:
    args = parse_args()
    try:
        inputs = collect_inputs(args)
        api_key = get_api_key()
    except (ExtractError, OSError) as error:
        print(json.dumps({"ok": False, "error": str(error)}, ensure_ascii=False))
        return 2

    base_url = os.environ.get("QINGDOU_BASE_URL", DEFAULT_BASE_URL).strip()
    results: list[dict[str, Any]] = []
    for user_input in inputs:
        try:
            batch_id = commit_task(base_url, api_key, user_input)
            result = poll_task(
                base_url,
                api_key,
                batch_id,
                max(0.2, args.poll_interval),
                max(1.0, args.timeout),
            )
            items = result.get("list")
            if not isinstance(items, list) or not items:
                raise ExtractError("批任务没有返回条目。")
            item_results = [
                save_item(item, args.output_dir, args.overwrite)
                for item in items
                if isinstance(item, dict)
            ]
            results.append(
                {
                    "input": user_input,
                    "items": item_results,
                }
            )
        except (ExtractError, OSError) as error:
            results.append({"input": user_input, "ok": False, "error": str(error)})

    success_count = sum(
        1
        for result in results
        for item in result.get("items", [])
        if item.get("ok")
    )
    failure_count = sum(
        1
        for result in results
        if result.get("ok") is False
    ) + sum(
        1
        for result in results
        for item in result.get("items", [])
        if not item.get("ok")
    )
    summary = {
        "ok": failure_count == 0,
        "success_count": success_count,
        "failure_count": failure_count,
        "results": results,
    }
    print(json.dumps(summary, ensure_ascii=False, indent=2))
    return 0 if failure_count == 0 else 1


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env python3
"""一次执行 TikHub 数据查询和轻抖短视频文稿提取。"""

from __future__ import annotations

import argparse
import json
import os
import sys
from pathlib import Path
from typing import Any

import extract_video_transcript as transcript
import tikhub_api as tikhub


class CombinedError(RuntimeError):
    """可向调用方展示的整合流程错误。"""


LEGAL_NOTICE = (
    "dbskill 是免费开源项目，与 TikHub、轻抖无隶属、合作或利益关系。"
    "充值由用户自主决定，相关交易、服务、争议及风险由用户自行承担；"
    "法律另有规定的除外，dbskill 不承担责任。"
)
PURCHASE_URLS = {
    "TikHub": "https://user.tikhub.io/dashboard/add-credit",
    "轻抖": "https://www.qingdou.vip/voice-text-api",
}
API_KEY_EXPLANATION = (
    "想让这个 Skill 工作，需要先开通外部数据服务。API Key 是服务商在开通后"
    "提供的一串使用凭证，可以理解为这个 Skill 调用服务的专用通行证。"
    "你不需要理解技术原理，也不要把它发到聊天中。"
)
PURCHASE_CHOICES = {
    "完整功能": {
        "providers": ["TikHub", "轻抖"],
        "result": "同时获得作品／账号数据和语音文字稿",
    },
    "只看数据": {
        "providers": ["TikHub"],
        "result": "获得作者、标题、点赞、评论等作品／账号数据",
    },
    "只要文字稿": {
        "providers": ["轻抖"],
        "result": "把短视频口播提取成文字稿",
    },
}
PURCHASE_STEPS = [
    "按需要的结果选择完整功能、只看数据或只要文字稿。",
    "打开对应充值地址，在服务商页面注册或登录。",
    "查看页面当前显示的套餐、额度、有效期和计费规则，自主决定是否充值。",
    "充值后进入个人中心，寻找 API Key、密钥管理或开发者设置；需要时创建并复制凭证。",
    "不要把凭证发送到聊天中；只需回复 TikHub 已充值、轻抖已充值或两个都已充值。",
    "随后在本地终端运行安全配置脚本，并重新检查服务状态。",
]
SETUP_COMMANDS = {
    "TikHub": "python3 scripts/configure_api_key.py tikhub",
    "轻抖": "python3 scripts/configure_api_key.py qingdou",
}


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="同时查询短视频数据并提取 Markdown 文字稿。"
    )
    parser.add_argument("input", nargs="?", help="短视频链接或完整分享文案")
    parser.add_argument("--stdin", action="store_true", help="从标准输入读取分享文案")
    parser.add_argument(
        "--check-keys",
        action="store_true",
        help="只检查两个 API Key 的发现状态，不读取输入或发起网络请求",
    )
    parser.add_argument(
        "--mode",
        choices=("both", "data", "transcript"),
        default="both",
        help="默认同时查询数据和提取文字稿",
    )
    parser.add_argument(
        "--output-dir",
        type=Path,
        default=Path.cwd() / "短视频文字稿",
        help="Markdown 输出根目录",
    )
    parser.add_argument("--overwrite", action="store_true", help="覆盖同源文稿")
    parser.add_argument(
        "--source",
        choices=("auto", "app", "web"),
        default="auto",
        help="TikHub 抖音作品数据源；小红书与视频号忽略此参数",
    )
    parser.add_argument("--raw-data", action="store_true", help="保留 TikHub 完整响应")
    parser.add_argument(
        "--data-timeout", type=float, default=15.0, help="TikHub 请求超时秒数"
    )
    parser.add_argument(
        "--poll-interval", type=float, default=2.0, help="轻抖轮询间隔秒数"
    )
    parser.add_argument(
        "--transcript-timeout", type=float, default=900.0, help="轻抖最长等待秒数"
    )
    return parser.parse_args()


def collect_input(args: argparse.Namespace) -> str:
    values: list[str] = []
    if args.input and args.input.strip():
        values.append(args.input.strip())
    if args.stdin:
        value = sys.stdin.read().strip()
        if value:
            values.append(value)
    if not values:
        raise CombinedError("没有收到短视频链接或分享文案。")
    if len(values) > 1:
        raise CombinedError("请只提供 1 条短视频链接或分享文案。")
    return values[0]


def credential_summary() -> dict[str, Any]:
    has_tikhub = bool(tikhub.find_api_key())
    has_qingdou = bool(transcript.find_api_key())
    if has_tikhub and has_qingdou:
        state = "complete"
        message = "TikHub 与轻抖 API Key 均已配置，可以查询数据并提取文字稿。"
    elif has_tikhub:
        state = "data_only"
        message = (
            "TikHub 已经可以使用：能查询作品或账号数据。轻抖尚未开通，"
            "所以暂时不能生成语音文字稿。"
        )
    elif has_qingdou:
        state = "transcript_only"
        message = (
            "轻抖已经可以使用：能生成语音文字稿。TikHub 尚未开通，"
            "所以暂时不能查询作品或账号数据。"
        )
    else:
        state = "unavailable"
        message = (
            "当前两个外部服务都没有开通。如果不购买并配置至少一个，"
            "这个 Skill 无法使用。"
        )
    summary: dict[str, Any] = {
        "credential_state": state,
        "tikhub_configured": has_tikhub,
        "qingdou_configured": has_qingdou,
        "message": message,
    }
    missing_providers: list[str] = []
    if not has_tikhub:
        missing_providers.append("TikHub")
    if not has_qingdou:
        missing_providers.append("轻抖")
    if missing_providers:
        summary["purchase_required_to_use"] = not has_tikhub and not has_qingdou
        summary["purchase_required_for_full_functionality"] = True
        summary["api_key_explanation"] = API_KEY_EXPLANATION
        summary["purchase_choices"] = PURCHASE_CHOICES
        summary["legal_notice"] = LEGAL_NOTICE
        summary["purchase_urls"] = {
            provider: PURCHASE_URLS[provider] for provider in missing_providers
        }
        summary["purchase_steps"] = PURCHASE_STEPS
        summary["setup_commands"] = {
            provider: SETUP_COMMANDS[provider] for provider in missing_providers
        }
    return summary


def missing_credential_part(provider: str, capability: str) -> dict[str, Any]:
    return {
        "ok": False,
        "missing_api_key": True,
        "provider": provider,
        "error": f"缺少 {provider} API Key，当前无法{capability}。",
    }


def run_data(user_input: str, args: argparse.Namespace) -> dict[str, Any]:
    try:
        share_url = tikhub.extract_share_url(user_input)
    except tikhub.TikHubError as error:
        return {"ok": False, "error": str(error)}
    if not tikhub.detect_platform(share_url):
        return {
            "ok": None,
            "skipped": True,
            "reason": "当前 TikHub 数据解析只支持抖音、小红书和微信视频号链接。",
        }
    try:
        return tikhub.fetch_supported_link_mcp(
            tikhub.get_api_key(),
            share_url,
            args.source,
            max(1.0, args.data_timeout),
            args.raw_data,
        )
    except (tikhub.TikHubError, OSError) as error:
        return {"ok": False, "error": str(error)}


def apply_data_fallbacks(
    item: dict[str, Any], data_result: dict[str, Any] | None
) -> dict[str, Any]:
    enriched = dict(item)
    if not data_result or data_result.get("link_type") != "video":
        return enriched
    summary = data_result.get("response")
    if not isinstance(summary, dict) or "response" in summary:
        return enriched
    if not transcript.find_author(enriched) and summary.get("author"):
        enriched["authorName"] = summary["author"]
    if not str(enriched.get("videoTitle") or "").strip() and summary.get("description"):
        enriched["videoTitle"] = summary["description"]
    if not (enriched.get("awemeId") or enriched.get("videoId")):
        video_id = summary.get("aweme_id") or summary.get("video_id")
        if video_id:
            enriched["videoId"] = video_id
    return enriched


def run_transcript(
    user_input: str,
    args: argparse.Namespace,
    data_result: dict[str, Any] | None = None,
) -> dict[str, Any]:
    try:
        api_key = transcript.get_api_key()
        base_url = os.environ.get(
            "QINGDOU_BASE_URL", transcript.DEFAULT_BASE_URL
        ).strip()
        batch_id = transcript.commit_task(base_url, api_key, user_input)
        result = transcript.poll_task(
            base_url,
            api_key,
            batch_id,
            max(0.2, args.poll_interval),
            max(1.0, args.transcript_timeout),
        )
        items = result.get("list")
        if not isinstance(items, list) or not items:
            raise transcript.ExtractError("批任务没有返回文稿条目。")
        item_results = [
            transcript.save_item(
                apply_data_fallbacks(item, data_result),
                args.output_dir,
                args.overwrite,
            )
            for item in items
            if isinstance(item, dict)
        ]
        if not item_results:
            raise transcript.ExtractError("批任务没有返回可处理的文稿条目。")
        success_count = sum(1 for item in item_results if item.get("ok"))
        return {
            "ok": success_count == len(item_results),
            "success_count": success_count,
            "failure_count": len(item_results) - success_count,
            "items": item_results,
        }
    except (transcript.ExtractError, OSError) as error:
        return {"ok": False, "error": str(error)}


def overall_status(parts: list[dict[str, Any]]) -> tuple[bool, bool]:
    attempted = [part for part in parts if not part.get("skipped")]
    successes = [part for part in attempted if part.get("ok") is True]
    failures = [part for part in attempted if part.get("ok") is False]
    return bool(successes) and not failures, bool(successes) and bool(failures)


def main() -> int:
    args = parse_args()
    credentials = credential_summary()
    if args.check_keys:
        print(json.dumps(credentials, ensure_ascii=False, indent=2))
        return 0 if credentials["credential_state"] != "unavailable" else 2

    needs_data = args.mode in ("both", "data")
    needs_transcript = args.mode in ("both", "transcript")
    has_required_key = (
        (needs_data and credentials["tikhub_configured"])
        or (needs_transcript and credentials["qingdou_configured"])
    )
    if not has_required_key:
        print(
            json.dumps(
                {
                    "ok": False,
                    "mode": args.mode,
                    **credentials,
                    "next_step": (
                        "先按需要的结果选择服务并完成充值。充值后不要发送凭证，"
                        "只需回复已充值，再按 setup_commands 在本地安全保存并复查状态。"
                    ),
                },
                ensure_ascii=False,
                indent=2,
            )
        )
        return 2

    try:
        user_input = collect_input(args)
    except CombinedError as error:
        print(json.dumps({"ok": False, "error": str(error)}, ensure_ascii=False))
        return 2

    result: dict[str, Any] = {"mode": args.mode, **credentials}
    parts: list[dict[str, Any]] = []
    data_result: dict[str, Any] | None = None
    if args.mode in ("both", "data"):
        if credentials["tikhub_configured"]:
            data_result = run_data(user_input, args)
        else:
            data_result = missing_credential_part(
                "TikHub", "查询作品或账号数据"
            )
        result["data"] = data_result
        parts.append(data_result)
    if args.mode in ("both", "transcript"):
        if not credentials["qingdou_configured"]:
            result["transcript"] = missing_credential_part(
                "轻抖", "提取语音文字稿"
            )
        elif args.mode == "both" and data_result and data_result.get("link_type") == "user":
            result["transcript"] = {
                "ok": None,
                "skipped": True,
                "reason": "链接指向用户主页，没有可提取的单条视频文稿。",
            }
        else:
            result["transcript"] = run_transcript(user_input, args, data_result)
        parts.append(result["transcript"])

    ok, partial_success = overall_status(parts)
    result["ok"] = ok
    result["partial_success"] = partial_success
    print(json.dumps(result, ensure_ascii=False, indent=2))
    return 0 if ok else 1


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env python3
"""TikHub 只读 API 客户端：账户检查、多平台作品解析和受限 GET。"""

from __future__ import annotations

import argparse
import json
import os
import re
import subprocess
import sys
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path
from typing import Any


DEFAULT_BASE_URL = "https://api.tikhub.dev"
MCP_URLS = {
    "douyin": "https://mcp.tikhub.io/douyin/mcp",
    "xiaohongshu": "https://mcp.tikhub.io/xiaohongshu/mcp",
    "wechat": "https://mcp.tikhub.io/wechat/mcp",
}
MCP_PROTOCOL_VERSION = "2024-11-05"
DEFAULT_API_KEYS_FILE = Path.home() / ".config" / "dbs" / "API_Keys.md"
KEYCHAIN_SERVICE = "dbs-tikhub-api-key"
ACCOUNT_PATH = "/api/v1/tikhub/user/get_user_info"


class TikHubError(RuntimeError):
    """可向调用方展示、且不应包含 API Key 的错误。"""


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="调用 TikHub 只读 API，不在命令参数或日志中暴露 API Key。"
    )
    parser.add_argument(
        "--base-url",
        default=os.environ.get("TIKHUB_BASE_URL", DEFAULT_BASE_URL),
        help=f"TikHub 基础域名，默认 {DEFAULT_BASE_URL}",
    )
    parser.add_argument("--timeout", type=float, default=15.0, help="请求超时秒数")
    parser.add_argument("--output", type=Path, help="把 JSON 保存到指定文件")

    subparsers = parser.add_subparsers(dest="command", required=True)
    subparsers.add_parser("account", help="检查鉴权并返回脱敏账户摘要")

    douyin = subparsers.add_parser(
        "douyin-link",
        aliases=("douyin-video",),
        help="通过 TikHub MCP 自动识别并解析抖音作品或用户主页链接",
    )
    douyin.add_argument("input", nargs="?", help="抖音链接或完整分享文案")
    douyin.add_argument("--stdin", action="store_true", help="从标准输入读取分享文案")
    douyin.add_argument(
        "--source",
        choices=("auto", "app", "web"),
        default="auto",
        help="作品链接默认 App V3 无有效数据时回退 Web；用户主页不使用此参数",
    )
    douyin.add_argument(
        "--raw", action="store_true", help="返回完整 TikHub 响应，而非精简摘要"
    )

    xiaohongshu = subparsers.add_parser(
        "xiaohongshu-link",
        help="通过 TikHub MCP 解析小红书视频笔记链接",
    )
    xiaohongshu.add_argument("input", nargs="?", help="小红书链接或完整分享文案")
    xiaohongshu.add_argument("--stdin", action="store_true", help="从标准输入读取分享文案")
    xiaohongshu.add_argument(
        "--raw", action="store_true", help="返回完整 TikHub 响应，而非精简摘要"
    )

    wechat = subparsers.add_parser(
        "wechat-link",
        help="通过 TikHub MCP 解析微信视频号作品链接",
    )
    wechat.add_argument("input", nargs="?", help="视频号链接或完整分享文案")
    wechat.add_argument("--stdin", action="store_true", help="从标准输入读取分享文案")
    wechat.add_argument(
        "--raw", action="store_true", help="返回完整 TikHub 响应，而非精简摘要"
    )

    generic = subparsers.add_parser("get", help="调用 /api/v1/ 下的只读 GET 路由")
    generic.add_argument("--path", required=True, help="以 /api/v1/ 开头的接口路径")
    generic.add_argument(
        "--param",
        action="append",
        default=[],
        metavar="KEY=VALUE",
        help="查询参数，可重复传入",
    )
    return parser.parse_args()


def key_from_document(path: Path) -> str:
    try:
        text = path.read_text(encoding="utf-8")
    except OSError:
        return ""
    section = re.search(r"(?ms)^##\s+TikHub API\s*$\n(.*?)(?=^##\s|\Z)", text)
    if not section:
        return ""
    match = re.search(r"(?m)^-\s+\*\*Key\*\*:\s*(\S+)\s*$", section.group(1))
    return match.group(1) if match else ""


def find_api_key() -> str:
    value = os.environ.get("TIKHUB_API_KEY", "").strip()
    if value:
        return value

    configured_path = os.environ.get("TIKHUB_API_KEYS_FILE", "").strip()
    key_path = Path(configured_path).expanduser() if configured_path else DEFAULT_API_KEYS_FILE
    value = key_from_document(key_path) if key_path.is_file() else ""
    if value:
        return value

    if (
        sys.platform == "darwin"
        and os.environ.get("DBS_VIDEO_EXTRACT_DISABLE_KEYCHAIN") != "1"
    ):
        try:
            result = subprocess.run(
                ["security", "find-generic-password", "-s", KEYCHAIN_SERVICE, "-w"],
                check=True,
                capture_output=True,
                text=True,
            )
        except (FileNotFoundError, subprocess.CalledProcessError):
            pass
        else:
            value = result.stdout.strip()
            if value:
                return value

    return ""


def get_api_key() -> str:
    value = find_api_key()
    if value:
        return value
    raise TikHubError(
        "缺少 TikHub API Key：请配置 TIKHUB_API_KEY、API_Keys.md，"
        f"或 macOS 钥匙串服务 {KEYCHAIN_SERVICE}。TikHub 只负责作品、"
        "账号和统计数据；没有这个 Key 时仍可单独使用轻抖提取文字稿。"
    )


def normalize_base_url(value: str) -> str:
    value = value.strip().rstrip("/")
    parsed = urllib.parse.urlparse(value)
    if parsed.scheme != "https" or not parsed.netloc or parsed.path not in ("", "/"):
        raise TikHubError("Base URL 必须是只有域名的 HTTPS 地址。")
    return value


def normalize_api_path(value: str) -> str:
    value = value.strip()
    parsed = urllib.parse.urlparse(value)
    if parsed.scheme or parsed.netloc or parsed.query or parsed.fragment:
        raise TikHubError("接口路径不能包含域名、查询字符串或片段。")
    if not value.startswith("/api/v1/"):
        raise TikHubError("只允许调用以 /api/v1/ 开头的只读接口。")
    return value


def parse_params(values: list[str]) -> dict[str, str]:
    params: dict[str, str] = {}
    for value in values:
        key, separator, item = value.partition("=")
        key = key.strip()
        if not separator or not key:
            raise TikHubError(f"查询参数格式错误：{value!r}，应为 KEY=VALUE。")
        params[key] = item
    return params


def request_json(
    base_url: str,
    path: str,
    api_key: str,
    params: dict[str, str] | None,
    timeout: float,
) -> dict[str, Any]:
    normalized_base = normalize_base_url(base_url)
    normalized_path = normalize_api_path(path)
    query = urllib.parse.urlencode(params or {})
    url = f"{normalized_base}{normalized_path}"
    if query:
        url = f"{url}?{query}"
    request = urllib.request.Request(
        url,
        method="GET",
        headers={
            "Accept": "application/json",
            "Authorization": f"Bearer {api_key}",
            "User-Agent": "dbs-video-extract/1.0",
        },
    )
    try:
        with urllib.request.urlopen(request, timeout=max(1.0, timeout)) as response:
            raw = response.read().decode("utf-8")
            http_status = response.status
    except urllib.error.HTTPError as error:
        detail = error.read().decode("utf-8", errors="replace")[:1000]
        detail = detail.replace(api_key, "[REDACTED]")
        raise TikHubError(f"HTTP {error.code}：{detail}") from error
    except urllib.error.URLError as error:
        raise TikHubError(f"网络请求失败：{error.reason}") from error
    except TimeoutError as error:
        raise TikHubError("网络请求超时。") from error

    try:
        data = json.loads(raw)
    except json.JSONDecodeError as error:
        raise TikHubError(f"HTTP {http_status}，接口没有返回合法 JSON。") from error
    if not isinstance(data, dict):
        raise TikHubError(f"HTTP {http_status}，JSON 顶层不是对象。")
    return data


def parse_sse_json(body: str) -> dict[str, Any]:
    data_lines = [
        line[6:] for line in body.splitlines() if line.startswith("data: ")
    ]
    payload = "\n".join(data_lines) if data_lines else body
    try:
        data = json.loads(payload)
    except json.JSONDecodeError as error:
        raise TikHubError("MCP 没有返回合法 JSON 或 SSE data。") from error
    if not isinstance(data, dict):
        raise TikHubError("MCP 返回的 JSON 顶层不是对象。")
    return data


def read_mcp_response(response: Any, allow_empty: bool = False) -> dict[str, Any]:
    content_type = str(response.headers.get("Content-Type", "")).lower()
    if "text/event-stream" not in content_type:
        raw = response.read().decode("utf-8")
        if allow_empty and not raw.strip():
            return {}
        return parse_sse_json(raw)

    data_lines: list[str] = []
    while True:
        raw_line = response.readline()
        if not raw_line:
            break
        line = raw_line.decode("utf-8", errors="replace").rstrip("\r\n")
        if line.startswith("data:"):
            data_lines.append(line[5:].lstrip())
        elif not line and data_lines:
            return parse_sse_json("\n".join(data_lines))
    if data_lines:
        return parse_sse_json("\n".join(data_lines))
    if allow_empty:
        return {}
    raise TikHubError("MCP SSE 响应中没有 data 事件。")


def mcp_post(
    api_key: str,
    payload: dict[str, Any],
    timeout: float,
    session_id: str = "",
    platform: str = "douyin",
    allow_empty: bool = False,
) -> tuple[str, dict[str, Any]]:
    headers = {
        "Accept": "application/json, text/event-stream",
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json",
        "User-Agent": "dbs-video-extract/1.0",
    }
    if session_id:
        headers["Mcp-Session-Id"] = session_id
    request = urllib.request.Request(
        MCP_URLS.get(platform, ""),
        data=json.dumps(payload).encode("utf-8"),
        method="POST",
        headers=headers,
    )
    try:
        with urllib.request.urlopen(request, timeout=max(1.0, timeout)) as response:
            returned_session = response.headers.get("Mcp-Session-Id", "")
            parsed_response = read_mcp_response(response, allow_empty)
    except urllib.error.HTTPError as error:
        detail = error.read().decode("utf-8", errors="replace")[:1000]
        detail = detail.replace(api_key, "[REDACTED]")
        raise TikHubError(f"MCP HTTP {error.code}：{detail}") from error
    except urllib.error.URLError as error:
        raise TikHubError(f"MCP 网络请求失败：{error.reason}") from error
    except TimeoutError as error:
        raise TikHubError(f"TikHub {platform} MCP 请求超时。") from error
    return returned_session, parsed_response


def mcp_initialize(api_key: str, timeout: float, platform: str = "douyin") -> str:
    if platform not in MCP_URLS:
        raise TikHubError(f"暂不支持 TikHub MCP 平台：{platform}")
    session_id, response = mcp_post(
        api_key,
        {
            "jsonrpc": "2.0",
            "id": 1,
            "method": "initialize",
            "params": {
                "protocolVersion": MCP_PROTOCOL_VERSION,
                "capabilities": {},
                "clientInfo": {"name": "dbs-video-extract", "version": "1.0"},
            },
        },
        timeout,
        platform=platform,
    )
    if not session_id:
        raise TikHubError("MCP initialize 成功，但响应缺少 Mcp-Session-Id。")
    if "error" in response:
        raise TikHubError(f"MCP initialize 失败：{response['error']}")
    mcp_post(
        api_key,
        {
            "jsonrpc": "2.0",
            "method": "notifications/initialized",
            "params": {},
        },
        timeout,
        session_id,
        platform,
        allow_empty=True,
    )
    return session_id


def mcp_tool_call(
    api_key: str,
    tool_name: str,
    arguments: dict[str, Any],
    timeout: float,
    platform: str = "douyin",
) -> dict[str, Any]:
    session_id = mcp_initialize(api_key, timeout, platform)
    _, response = mcp_post(
        api_key,
        {
            "jsonrpc": "2.0",
            "id": 2,
            "method": "tools/call",
            "params": {"name": tool_name, "arguments": arguments},
        },
        timeout,
        session_id,
        platform,
    )
    if "error" in response:
        raise TikHubError(f"MCP tools/call 失败：{response['error']}")
    result = response.get("result")
    if not isinstance(result, dict):
        raise TikHubError("MCP tools/call 响应缺少 result。")

    structured = result.get("structuredContent")
    candidate: Any = structured.get("result") if isinstance(structured, dict) else None
    if candidate is None:
        content = result.get("content")
        if isinstance(content, list):
            for item in content:
                if isinstance(item, dict) and item.get("type") == "text":
                    candidate = item.get("text")
                    break
    if isinstance(candidate, str):
        try:
            candidate = json.loads(candidate)
        except json.JSONDecodeError:
            return {"text": candidate, "mcp_is_error": bool(result.get("isError"))}
    if not isinstance(candidate, dict):
        raise TikHubError("MCP 工具没有返回可解析的对象。")
    return candidate


def account_summary(response: dict[str, Any]) -> dict[str, Any]:
    user_data = response.get("user_data")
    api_key_data = response.get("api_key_data")
    safe_user = dict(user_data) if isinstance(user_data, dict) else {}
    safe_user.pop("email", None)
    safe_key = dict(api_key_data) if isinstance(api_key_data, dict) else {}
    for field in tuple(safe_key):
        if "key" in field.lower() and field not in {
            "api_key_name",
            "api_key_scopes",
            "api_key_status",
        }:
            safe_key.pop(field, None)
    return {
        "ok": response.get("code") == 200,
        "code": response.get("code"),
        "router": response.get("router"),
        "api_key_data": safe_key,
        "user_data": safe_user,
    }


def extract_share_url(value: str) -> str:
    match = re.search(r"https?://[^\s<>\"']+", value)
    if not match:
        raise TikHubError("输入中没有找到 HTTP 或 HTTPS 分享链接。")
    return match.group(0).rstrip("，。！？；：、,.!?;:)]}）】》")


def resolve_douyin_url(url: str, timeout: float) -> str:
    request = urllib.request.Request(
        url,
        method="GET",
        headers={
            "User-Agent": (
                "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
                "AppleWebKit/537.36 Chrome/131.0 Safari/537.36"
            )
        },
    )
    try:
        with urllib.request.urlopen(request, timeout=max(1.0, timeout)) as response:
            return response.geturl()
    except (urllib.error.HTTPError, urllib.error.URLError) as error:
        raise TikHubError(f"抖音短链接跳转解析失败：{error}") from error


def classify_douyin_url(resolved_url: str) -> tuple[str, str]:
    parsed = urllib.parse.urlparse(resolved_url)
    query = urllib.parse.parse_qs(parsed.query)
    path_parts = [part for part in parsed.path.split("/") if part]
    sec_user_id = (query.get("sec_uid") or query.get("sec_user_id") or [""])[0]
    if "user" in path_parts:
        if not sec_user_id:
            try:
                user_index = path_parts.index("user")
                sec_user_id = path_parts[user_index + 1]
            except (ValueError, IndexError):
                pass
        if not sec_user_id:
            raise TikHubError("链接指向用户主页，但没有找到 sec_user_id。")
        return "user", sec_user_id
    if any(part in path_parts for part in ("video", "note", "slides")):
        return "video", ""
    raise TikHubError(f"暂时无法识别抖音链接类型：{parsed.path}")


def collect_link_input(args: argparse.Namespace, platform_name: str) -> str:
    values: list[str] = []
    if args.input:
        values.append(args.input.strip())
    if args.stdin:
        stdin_value = sys.stdin.read().strip()
        if stdin_value:
            values.append(stdin_value)
    if not values:
        raise TikHubError(f"没有收到{platform_name}链接或分享文案。")
    if len(values) > 1:
        raise TikHubError(f"请只提供 1 条{platform_name}链接或分享文案。")
    return extract_share_url(values[0])


def detect_platform(url: str) -> str:
    hostname = (urllib.parse.urlparse(url).hostname or "").lower()
    if hostname == "douyin.com" or hostname.endswith(".douyin.com"):
        return "douyin"
    if hostname in {"xhslink.cn", "xhslink.com"} or hostname.endswith(
        ".xiaohongshu.com"
    ) or hostname == "xiaohongshu.com":
        return "xiaohongshu"
    if hostname == "weixin.qq.com" or hostname.endswith(".weixin.qq.com"):
        return "wechat"
    return ""


def has_video_payload(response: dict[str, Any]) -> bool:
    if response.get("code") not in (None, 200):
        return False
    data = response.get("data")
    if not isinstance(data, dict) or not data:
        return False
    for key in ("aweme_detail", "aweme_info", "item", "video"):
        if isinstance(data.get(key), dict) and data[key]:
            return True
    for key in ("aweme_list", "item_list"):
        if isinstance(data.get(key), list) and data[key]:
            return True
    return not set(data).issubset({"filter_list", "status_code", "status_msg"})


def api_response_success(response: dict[str, Any]) -> bool:
    return response.get("code") == 200 and isinstance(response.get("data"), dict)


def response_charged(response: dict[str, Any]) -> bool:
    messages = f"{response.get('message', '')} {response.get('message_zh', '')}".lower()
    return "charge" in messages or "计费" in messages


def summarize_user_response(response: dict[str, Any]) -> dict[str, Any]:
    data = response.get("data")
    user = data.get("user") if isinstance(data, dict) else None
    if not isinstance(user, dict):
        return {"ok": False, "code": response.get("code"), "response": response}
    avatar = user.get("avatar_larger")
    avatar_urls = avatar.get("url_list") if isinstance(avatar, dict) else None
    return {
        "ok": True,
        "code": response.get("code"),
        "nickname": user.get("nickname"),
        "douyin_id": user.get("unique_id"),
        "uid": user.get("uid"),
        "sec_user_id": user.get("sec_uid"),
        "signature": user.get("signature"),
        "ip_location": user.get("ip_location"),
        "following_count": user.get("following_count"),
        "follower_count": user.get("follower_count"),
        "total_favorited": user.get("total_favorited"),
        "aweme_count": user.get("aweme_count"),
        "mix_count": user.get("mix_count"),
        "live_status": user.get("live_status"),
        "avatar_url": avatar_urls[0] if isinstance(avatar_urls, list) and avatar_urls else None,
        "charged": response_charged(response),
    }


def summarize_video_response(response: dict[str, Any]) -> dict[str, Any]:
    data = response.get("data")
    if not isinstance(data, dict):
        return {"ok": False, "code": response.get("code"), "response": response}
    item = data.get("aweme_detail") or data.get("aweme_info") or data.get("item")
    if not isinstance(item, dict):
        return {"ok": False, "code": response.get("code"), "response": response}
    author = item.get("author")
    statistics = item.get("statistics")
    return {
        "ok": True,
        "code": response.get("code"),
        "aweme_id": item.get("aweme_id") or item.get("item_id"),
        "description": item.get("desc") or item.get("description"),
        "author": author if isinstance(author, str) else (
            author.get("nickname") if isinstance(author, dict) else None
        ),
        "statistics": statistics if isinstance(statistics, dict) else None,
        "charged": response_charged(response),
    }


def summarize_xiaohongshu_video_response(
    response: dict[str, Any],
) -> dict[str, Any]:
    outer_data = response.get("data")
    notes = outer_data.get("data") if isinstance(outer_data, dict) else None
    item = notes[0] if isinstance(notes, list) and notes else None
    if not isinstance(item, dict):
        return {"ok": False, "code": response.get("code"), "response": response}
    user = item.get("user")
    video_info = item.get("video_info_v2")
    capa = video_info.get("capa") if isinstance(video_info, dict) else None
    share_info = item.get("share_info")
    tags = item.get("hash_tag")
    return {
        "ok": True,
        "code": response.get("code"),
        "video_id": item.get("id"),
        "description": item.get("desc") or item.get("title"),
        "title": item.get("title"),
        "author": user.get("nickname") if isinstance(user, dict) else None,
        "author_id": user.get("id") if isinstance(user, dict) else None,
        "red_id": user.get("red_id") if isinstance(user, dict) else None,
        "published_at": item.get("time"),
        "ip_location": item.get("ip_location"),
        "duration_seconds": capa.get("duration") if isinstance(capa, dict) else None,
        "statistics": {
            "like_count": item.get("liked_count"),
            "collect_count": item.get("collected_count"),
            "comment_count": item.get("comments_count"),
            "share_count": item.get("shared_count"),
            "view_count": item.get("view_count"),
        },
        "tags": [
            tag.get("name")
            for tag in tags
            if isinstance(tag, dict) and tag.get("name")
        ] if isinstance(tags, list) else [],
        "cover_url": share_info.get("image") if isinstance(share_info, dict) else None,
        "charged": response_charged(response),
    }


def summarize_wechat_video_response(response: dict[str, Any]) -> dict[str, Any]:
    item = response.get("data")
    if not isinstance(item, dict) or not item:
        return {"ok": False, "code": response.get("code"), "response": response}
    media = item.get("media")
    return {
        "ok": True,
        "code": response.get("code"),
        "video_id": item.get("id"),
        "description": item.get("description") or item.get("title"),
        "title": item.get("title") or item.get("short_title"),
        "author": item.get("nickname"),
        "author_id": item.get("username"),
        "published_at": item.get("create_time"),
        "duration_seconds": media.get("duration") if isinstance(media, dict) else None,
        "statistics": {
            "read_count": item.get("read_count"),
            "like_count": item.get("like_count"),
            "collect_count": item.get("fav_count"),
            "comment_count": item.get("comment_count"),
            "share_count": item.get("forward_count"),
        },
        "cover_url": media.get("cover_url") if isinstance(media, dict) else None,
        "charged": response_charged(response),
    }


def fetch_douyin_link_mcp(
    api_key: str,
    share_url: str,
    source: str,
    timeout: float,
    raw: bool,
) -> dict[str, Any]:
    resolved_url = resolve_douyin_url(share_url, timeout)
    link_type, identifier = classify_douyin_url(resolved_url)
    if link_type == "user":
        tool = "douyin_app_v3_handler_user_profile"
        response = mcp_tool_call(
            api_key, tool, {"sec_user_id": identifier}, timeout
        )
        payload = response if raw else summarize_user_response(response)
        return {
            "ok": api_response_success(response),
            "transport": "mcp",
            "link_type": "user",
            "tool": tool,
            "response": payload,
        }

    tools = {
        "app": "douyin_app_v3_fetch_one_video_by_share_url",
        "web": "douyin_web_fetch_one_video_by_share_url",
    }
    attempts = ("app", "web") if source == "auto" else (source,)
    previous_errors: list[dict[str, Any]] = []
    last_response: dict[str, Any] = {}
    last_provider = attempts[-1]
    for provider in attempts:
        last_provider = provider
        response = mcp_tool_call(
            api_key, tools[provider], {"share_url": share_url}, timeout
        )
        last_response = response
        if api_response_success(response) and has_video_payload(response):
            payload = response if raw else summarize_video_response(response)
            return {
                "ok": True,
                "transport": "mcp",
                "link_type": "video",
                "provider": provider,
                "fallback_used": provider == "web" and len(attempts) > 1,
                "tool": tools[provider],
                "response": payload,
                "previous_errors": previous_errors,
            }
        previous_errors.append(
            {
                "provider": provider,
                "code": response.get("code"),
                "error": response.get("error") or response.get("message"),
            }
        )
    return {
        "ok": False,
        "transport": "mcp",
        "link_type": "video",
        "provider": last_provider,
        "fallback_used": source == "auto",
        "tool": tools[last_provider],
        "response": last_response if raw else summarize_video_response(last_response),
        "previous_errors": previous_errors,
    }


def fetch_xiaohongshu_video_mcp(
    api_key: str,
    share_url: str,
    timeout: float,
    raw: bool,
) -> dict[str, Any]:
    tool = "xiaohongshu_app_v2_get_video_note_detail"
    response = mcp_tool_call(
        api_key,
        tool,
        {"share_text": share_url},
        timeout,
        platform="xiaohongshu",
    )
    summary = summarize_xiaohongshu_video_response(response)
    return {
        "ok": bool(summary.get("ok")),
        "transport": "mcp",
        "platform": "xiaohongshu",
        "link_type": "video",
        "tool": tool,
        "response": response if raw else summary,
    }


def fetch_wechat_video_mcp(
    api_key: str,
    share_url: str,
    timeout: float,
    raw: bool,
) -> dict[str, Any]:
    tool = "wechat_channels_v2_fetch_video_detail"
    response = mcp_tool_call(
        api_key,
        tool,
        {"share_url": share_url, "raw": False},
        timeout,
        platform="wechat",
    )
    summary = summarize_wechat_video_response(response)
    return {
        "ok": bool(summary.get("ok")),
        "transport": "mcp",
        "platform": "wechat",
        "link_type": "video",
        "tool": tool,
        "response": response if raw else summary,
    }


def fetch_supported_link_mcp(
    api_key: str,
    share_url: str,
    source: str,
    timeout: float,
    raw: bool,
) -> dict[str, Any]:
    platform = detect_platform(share_url)
    if platform == "douyin":
        result = fetch_douyin_link_mcp(api_key, share_url, source, timeout, raw)
        result.setdefault("platform", "douyin")
        return result
    if platform == "xiaohongshu":
        return fetch_xiaohongshu_video_mcp(api_key, share_url, timeout, raw)
    if platform == "wechat":
        return fetch_wechat_video_mcp(api_key, share_url, timeout, raw)
    raise TikHubError("当前 TikHub 数据解析只支持抖音、小红书和微信视频号链接。")


def emit_json(data: dict[str, Any], output: Path | None) -> None:
    rendered = json.dumps(data, ensure_ascii=False, indent=2)
    if output is None:
        print(rendered)
        return
    target = output.expanduser().resolve()
    target.parent.mkdir(parents=True, exist_ok=True)
    target.write_text(f"{rendered}\n", encoding="utf-8")
    print(json.dumps({"ok": True, "output": str(target)}, ensure_ascii=False))


def main() -> int:
    args = parse_args()
    try:
        api_key = get_api_key()
        base_url = normalize_base_url(args.base_url)
        if args.command == "account":
            result = account_summary(
                request_json(base_url, ACCOUNT_PATH, api_key, None, args.timeout)
            )
        elif args.command in ("douyin-link", "douyin-video"):
            result = fetch_douyin_link_mcp(
                api_key,
                collect_link_input(args, "抖音"),
                args.source,
                args.timeout,
                args.raw,
            )
        elif args.command == "xiaohongshu-link":
            result = fetch_xiaohongshu_video_mcp(
                api_key,
                collect_link_input(args, "小红书"),
                args.timeout,
                args.raw,
            )
        elif args.command == "wechat-link":
            result = fetch_wechat_video_mcp(
                api_key,
                collect_link_input(args, "视频号"),
                args.timeout,
                args.raw,
            )
        elif args.command == "get":
            path = normalize_api_path(args.path)
            result = {
                "ok": True,
                "path": path,
                "response": request_json(
                    base_url,
                    path,
                    api_key,
                    parse_params(args.param),
                    args.timeout,
                ),
            }
        else:
            raise TikHubError(f"未知命令：{args.command}")
        emit_json(result, args.output)
        return 0 if result.get("ok", True) else 1
    except (TikHubError, OSError) as error:
        print(json.dumps({"ok": False, "error": str(error)}, ensure_ascii=False))
        return 2


if __name__ == "__main__":
    raise SystemExit(main())

---
name: dbs-video-extract
description: 接收抖音、小红书、微信视频号等短视频链接或分享文案，查询作品／账号数据并提取语音文字稿，按作者和标题归档为 Markdown。用户希望完整解析短视频、取得数据或文字稿时使用；运行前检查 TikHub 与轻抖使用凭证，缺少时用新手能理解的方式引导购买和安全配置。
---

# dbs-video-extract：短视频信息提取

目标：用一个入口处理同一条短视频分享文案。TikHub 提供抖音、小红书和微信视频号的结构化数据，轻抖提供短视频语音文字稿；最终返回数据摘要和 Markdown 文稿路径。

## 先检查使用凭证

任何提取任务开始前，先运行只读预检：

```bash
python3 scripts/extract_video.py --check-keys
```

根据 `credential_state` 处理：

| 状态 | 能力 | 必须告知用户 |
| --- | --- | --- |
| `complete` | 数据与文字稿都可用 | 两部分都可以执行 |
| `data_only` | 只有 TikHub 数据查询可用 | 只能取得作品／账号数据，无法生成语音文字稿 |
| `transcript_only` | 只有轻抖文字稿可用 | 只能取得语音文字稿，无法查询 TikHub 作品／账号数据 |
| `unavailable` | 两部分都不可用 | 当前无法使用；先解释使用凭证，再带用户完成购买和配置 |

只有一个 Key 时允许继续，不要求用户补齐另一个 Key。执行前明确说明本次只能解决哪一部分，默认把模式收窄为当前可用的一侧。用户仍要求 `both` 时保留单侧结果，并将另一侧标记为缺少凭证。

两个 Key 都没有时停止提取。不要直接用「API Key」「接口鉴权」等术语作为结论。先用下面这句话解释：

> 想让这个 Skill 工作，需要先开通外部数据服务。API Key 是服务商在开通后提供的一串使用凭证，可以理解为这个 Skill 调用服务的专用通行证。你不需要理解技术原理，也不要把它发到聊天中。

随后告诉用户：「当前两个服务都没有开通。如果不购买并配置至少一个，这个 Skill 无法使用。」再按 [references/api-setup.md](references/api-setup.md) 完成需求选择、购买、凭证保存和连接测试。

不要替用户付款、接受服务协议或处理完整 Key。不要要求用户把 Key 发到对话里。购买入口、逐步引导、安全配置脚本和连接测试见 [references/api-setup.md](references/api-setup.md)。

每次向用户展示购买或充值地址时，必须先原样展示下面这条提示，不能省略或改写：

> dbskill 是免费开源项目，与 TikHub、轻抖无隶属、合作或利益关系。充值由用户自主决定，相关交易、服务、争议及风险由用户自行承担；法律另有规定的除外，dbskill 不承担责任。

用户没有进入购买或充值引导时不重复展示。

### 缺少凭证时的对话要求

缺少任一服务时，回复必须形成下面的连续路径：

1. 先用普通语言解释 API Key；
2. 让用户按结果选择：完整功能、只看数据、只要文字稿；
3. 说明每个选择需要购买哪个服务，以及只买一个会少什么；
4. 展示固定法律提示，再展示对应充值地址；
5. 逐步说明注册／登录、查看页面套餐、充值、进入个人中心或密钥管理页、取得凭证；
6. 告诉用户取得凭证后不要发到聊天中，回复「TikHub 已充值」「轻抖已充值」或「两个都已充值」；
7. 用户回复已充值后，引导其在本地终端运行安全配置脚本：

```bash
python3 scripts/configure_api_key.py tikhub
python3 scripts/configure_api_key.py qingdou
```

脚本会在终端中隐蔽读取凭证并保存到本地私密配置，不能把完整凭证显示在命令、回复或日志中。保存后重新运行 `--check-keys`，直到对应状态变为可用。

不要一次只发链接，也不要假定用户知道去哪里找凭证。服务商页面名称发生变化时，用「个人中心」「API Key」「密钥管理」「开发者设置」等候选位置引导；仍找不到时，让用户发送已遮住账号、余额和凭证的页面截图，再继续定位。

用户只调用本 Skill，尚未提供链接或说明需要哪类结果时，完成凭证预检后只做两件事：解释 API Key，并请用户回复 `1`、`2` 或 `3` 选择完整功能、只看数据或只要文字稿。收到选择后再展示法律提示、对应充值地址和购买步骤。用户已经发送视频链接或明确要求完整提取时，默认选择完整功能，无需重复询问。

## 默认执行

优先使用标准输入，避免分享文案中的特殊字符被 Shell 解释：

```bash
python3 scripts/extract_video.py \
  --output-dir "/绝对路径/短视频文字稿" \
  --stdin
```

默认模式为 `both`：

1. 从完整分享文案中提取链接；
2. 根据链接平台调用 TikHub MCP：抖音支持用户主页或单条作品，小红书与视频号支持单条视频；
3. 保留完整分享文案提交轻抖 API，轮询取得文稿；
4. 按 `{输出目录}/{作者}/{标题}.md` 保存文稿；
5. 输出 JSON 汇总，包含凭证状态、TikHub 数据摘要和文稿文件路径。

TikHub 或轻抖任一侧失败时继续完成另一侧，并在汇总中标记 `partial_success`。已经取得的结果不能因单侧失败而丢弃。

## 可选模式

只有用户明确只需要一侧结果，或预检确认只有一侧 Key 时才切换：

```bash
python3 scripts/extract_video.py --mode data --stdin
python3 scripts/extract_video.py --mode transcript --stdin
```

- `data`：只查询 TikHub 数据，支持抖音、小红书和微信视频号；
- `transcript`：只提取文字稿，支持轻抖 API 能解析的平台；
- `both`：同时执行两侧。

同一来源已经存在时默认跳过文稿写入，保护用户编辑；用户明确要求重新生成时加 `--overwrite`。需要 TikHub 完整响应时加 `--raw-data`。

## 密钥与费用

脚本按以下顺序读取 Key：

1. 环境变量：`TIKHUB_API_KEY`、`QINGDOU_API_KEY`；
2. 用户指定的本地文件：`TIKHUB_API_KEYS_FILE`、`QINGDOU_API_KEYS_FILE`；
3. `~/.config/dbs/API_Keys.md`；
4. macOS 钥匙串：`dbs-tikhub-api-key`、`dbs-qingdou-api-key`。

禁止把 Key 写入 Skill、命令参数、Markdown、Git 或日志。TikHub 查询和轻抖转写都可能计费；执行前告知用户。用户已经明确要求提取时无需重复确认。

## 数据边界

- TikHub 负责作品／账号资料、统计和媒体元信息；抖音作品优先 App V3，无有效数据时只回退 1 次 Web；小红书使用 App V2 视频笔记详情；视频号使用 Channels V2 作品详情。
- 轻抖负责语音文字稿。正文忠实保存 API 返回内容，不补写、润色或修订口误。
- 当前 TikHub 专用数据解析支持抖音、小红书和微信视频号；其他轻抖可识别的平台仍可只执行文字稿提取。
- 公开、删除、私密、版权和可见范围限制以接口返回为准，不尝试绕过。
- 不自动下载媒体、发布内容、提交 Git 或推送。

TikHub 调用细节见 [references/tikhub-api.md](references/tikhub-api.md)；轻抖状态码和兼容逻辑见 [references/qingdou-api.md](references/qingdou-api.md)。只在排错或维护对应部分时读取。

## 交付

完成后简洁报告：

- 当前配置了哪一个 API Key，以及可用能力范围；
- 数据查询是否成功、使用的 MCP 工具、是否发生 App → Web 回退；
- 关键作品／账号数据；
- 文稿是否成功、生成或跳过的 Markdown 绝对路径；
- 单侧缺少 Key 或执行失败的具体原因；
- 接口是否明确返回已计费。

不要在回复中展示 API Key、`batchId` 或无必要的内部请求标识。

完成当前任务后直接结束。只有用户明确询问下一步，且当前环境已经安装 `/dbs` 时，简短提示：「下一步不确定时，可以输入 `/dbs`。」