podcast-generator

Original source / Raw SKILL.md

---
name: podcast-generator
description: >
  Convert articles, blog posts, or any text into professional podcast scripts and TTS audio.
  Use when a user wants to: (1) Transform written content into conversational podcast scripts,
  (2) Generate TTS audio from scripts, (3) Create single-host or two-host dialogue episodes.
  Integrates SkillPay.me billing at 0.001 USDT per call.
---

# Podcast Generator

Converts articles into podcast scripts + audio. Charges 0.001 USDT per use via SkillPay.

## Workflow

```
1. Billing check  →  scripts/billing.py --charge --user-id <id>
2. Generate script →  scripts/generate_script.py --input <file> --format <solo|dialogue>
3. Generate audio  →  scripts/generate_audio.py --script <file> --output podcast.mp3
4. View stats     →  scripts/stats.py (NEW)
```

### Step 1: Billing

```bash
SKILLPAY_API_KEY=sk_xxx python3 scripts/billing.py --charge --user-id <user_id>
```

- `success: true` → proceed
- `needs_payment: true` → return `payment_url` to user for top-up

Other commands:
- `--balance` — check user balance
- `--payment-link` — generate top-up link

### Step 2: Script Generation

```bash
python3 scripts/generate_script.py --input article.txt --format solo --output script.md
python3 scripts/generate_script.py --input article.txt --format dialogue --output script.md
```

Formats: `solo` (single host) or `dialogue` (two hosts A/B with conversation).

### Step 3: Audio Generation

```bash
python3 scripts/generate_audio.py --script script.md --output podcast.mp3
```

Requires `edge-tts` (`pip install edge-tts`). Uses different voices for Host A (female) and Host B (male). Falls back to segment list if edge-tts unavailable.

### Usage Statistics (NEW)

```bash
python3 scripts/stats.py  # Show usage stats
python3 scripts/stats.py --action log --title "Episode 1" --format solo --audio-seconds 120
```

Tracks: total generations, audio duration, cost breakdown by format.

## Config

| Env Var | Required | Description |
|:---|:---:|:---|
| `SKILLPAY_API_KEY` | Yes | SkillPay.me API key |

## Script Templates

See `references/script-templates.md` for format details and voice options.


---

## Referenced Files

> The following files are referenced in this skill and included for context.

### scripts/billing.py

```python
#!/usr/bin/env python3
"""SkillPay.me billing — charge / balance / payment-link."""

import json, sys, argparse, os
import urllib.request, urllib.error

API = "https://skillpay.me/api/v1"
SKILL_ID = "015eae56-90e7-4ac8-bece-47349663f9d7"


def _key(override=None):
    return override or os.environ.get("SKILLPAY_API_KEY")


def _post(path, body, api_key):
    req = urllib.request.Request(
        f"{API}{path}",
        data=json.dumps(body).encode(),
        headers={"Content-Type": "application/json", "X-API-Key": api_key},
        method="POST",
    )
    try:
        with urllib.request.urlopen(req, timeout=15) as r:
            return json.loads(r.read())
    except urllib.error.HTTPError as e:
        return json.loads(e.read())


def _get(path, api_key):
    req = urllib.request.Request(f"{API}{path}", headers={"X-API-Key": api_key})
    try:
        with urllib.request.urlopen(req, timeout=15) as r:
            return json.loads(r.read())
    except urllib.error.HTTPError as e:
        return json.loads(e.read())


def charge(user_id, amount=0.001, api_key=None):
    k = _key(api_key)
    if not k:
        return {"success": False, "error": "SKILLPAY_API_KEY not set"}
    data = _post("/billing/charge", {
        "user_id": user_id,
        "skill_id": SKILL_ID,
        "amount": amount,
        "currency": "USDT",
        "description": "Podcast Generator",
    }, k)
    if data.get("success"):
        return {"success": True, "charged": amount, "data": data}
    return {
        "success": False,
        "needs_payment": bool(data.get("payment_url")),
        "payment_url": data.get("payment_url", ""),
        "balance": data.get("balance", 0),
        "message": data.get("message", "charge failed"),
    }


def balance(user_id, api_key=None):
    k = _key(api_key)
    if not k:
        return {"success": False, "error": "SKILLPAY_API_KEY not set"}
    return {"success": True, "data": _get(f"/billing/balance?user_id={user_id}", k)}


def payment_link(user_id, amount=5.0, api_key=None):
    k = _key(api_key)
    if not k:
        return {"success": False, "error": "SKILLPAY_API_KEY not set"}
    data = _post("/billing/payment-link", {
        "user_id": user_id,
        "skill_id": SKILL_ID,
        "Amount": amount,
    }, k)
    return {"success": True, "data": data}


if __name__ == "__main__":
    p = argparse.ArgumentParser()
    p.add_argument("--user-id", required=True)
    p.add_argument("--amount", type=float, default=0.001)
    p.add_argument("--api-key", default=None)
    g = p.add_mutually_exclusive_group()
    g.add_argument("--charge", action="store_true", default=True)
    g.add_argument("--balance", action="store_true")
    g.add_argument("--payment-link", action="store_true")
    a = p.parse_args()

    if a.balance:
        r = balance(a.user_id, a.api_key)
    elif a.payment_link:
        r = payment_link(a.user_id, a.amount or 5.0, a.api_key)
    else:
        r = charge(a.user_id, a.amount, a.api_key)

    print(json.dumps(r, indent=2, ensure_ascii=False))
    sys.exit(0 if r.get("success") else 1)

```

### scripts/generate_script.py

```python
#!/usr/bin/env python3
"""Generate podcast script from article text."""

import argparse, os, sys, re

SOLO = """# 🎙️ 播客脚本

**标题**: {title} | **格式**: 单人主播 | **时长**: ~{dur}分钟

---

## 开场白

大家好，欢迎收听本期节目！今天聊一个有意思的话题——{title}。

## 话题引入

{intro}

{segments}

## 总结

来回顾一下今天的要点：

{takes}

## 结尾

以上就是本期全部内容，觉得有收获的话别忘了订阅，我们下期再见！
"""

DIAL = """# 🎙️ 播客脚本

**标题**: {title} | **格式**: 双人对谈 | **时长**: ~{dur}分钟

---

## 开场白

**A**: 大家好！欢迎收听，我是主播A。

**B**: 我是主播B！今天聊——{title}。

**A**: 话题很火，咱们开始吧。

## 话题引入

**B**: 先介绍一下背景。

**A**: {intro}

{segments}

## 总结

**A**: 来总结一下。

{takes}

## 结尾

**B**: 今天就到这了！

**A**: 喜欢的话记得订阅，下期见！

**B**: 拜拜！
"""

TRANSITIONS_A = ["说得对，我补充一下。", "没错，还有一点。", "确实，另外呢——", "说到这个，我想到——", "对，我们接着聊。"]
TRANSITIONS_B = ["嗯，展开讲讲。", "有意思，继续。", "这点很关键。", "同意，而且——", "好观点，接着说。"]


def title(text):
    for l in text.strip().split("\n"):
        l = l.strip()
        if l:
            return re.sub(r'^#+\s*', '', l)[:60]
    return "未命名话题"


def paragraphs(text):
    ps, cur = [], []
    for l in text.split("\n"):
        s = l.strip()
        if not s:
            if cur:
                ps.append(" ".join(cur))
                cur = []
        elif not s.startswith("#"):
            cur.append(s)
    if cur:
        ps.append(" ".join(cur))
    return [p for p in ps if len(p) > 15]


def solo_segs(ps):
    parts = []
    for i, p in enumerate(ps):
        parts.append(f"## 第{i+1}部分\n\n{p}\n")
    return "\n---\n\n".join(parts)


def dial_segs(ps):
    parts = []
    for i, p in enumerate(ps):
        sp = "A" if i % 2 == 0 else "B"
        other = "B" if sp == "A" else "A"
        tr = (TRANSITIONS_A if other == "A" else TRANSITIONS_B)[i % 5]
        parts.append(f"## 第{i+1}部分\n\n**{sp}**: {p}\n\n**{other}**: {tr}\n")
    return "\n---\n\n".join(parts)


def takes(ps, fmt):
    out = []
    for i, p in enumerate(ps[:5]):
        s = p.split("。")[0]
        if len(s) > 80:
            s = s[:80] + "…"
        if fmt == "dialogue":
            sp = "A" if i % 2 == 0 else "B"
            out.append(f"**{sp}**: {i+1}. {s}")
        else:
            out.append(f"{i+1}. {s}")
    return "\n\n".join(out)


def generate(text, fmt="solo"):
    t = title(text)
    ps = paragraphs(text)
    if not ps:
        return "Error: 文本内容不足"
    dur = max(3, len(text) // 200)
    intro = ps[0]
    body = ps[1:] if len(ps) > 1 else ps
    tpl = DIAL if fmt == "dialogue" else SOLO
    segs = dial_segs(body) if fmt == "dialogue" else solo_segs(body)
    return tpl.format(title=t, dur=dur, intro=intro, segments=segs, takes=takes(body, fmt))


if __name__ == "__main__":
    p = argparse.ArgumentParser()
    p.add_argument("--input", required=True, help="File path or '-' for stdin")
    p.add_argument("--format", choices=["solo", "dialogue"], default="solo")
    p.add_argument("--output", default=None)
    a = p.parse_args()

    if a.input == "-":
        txt = sys.stdin.read()
    elif os.path.isfile(a.input):
        txt = open(a.input, encoding="utf-8").read()
    else:
        txt = a.input

    result = generate(txt, a.format)
    if a.output:
        open(a.output, "w", encoding="utf-8").write(result)
        print(f"✅ 脚本已生成: {a.output}")
    else:
        print(result)

```

### scripts/generate_audio.py

```python
#!/usr/bin/env python3
"""Generate audio from podcast script via edge-tts."""

import argparse, json, os, re, subprocess, sys

VOICE_A = "zh-CN-XiaoxiaoNeural"  # Female
VOICE_B = "zh-CN-YunxiNeural"     # Male


def parse(text):
    segs, speaker, buf = [], None, []
    for line in text.split("\n"):
        line = line.strip()
        if not line or line.startswith("#") or line == "---":
            if buf and speaker:
                segs.append({"speaker": speaker, "text": " ".join(buf)})
                buf = []
            continue
        m = re.match(r'\*\*(.+?)\*\*:\s*(.*)', line)
        if m:
            if buf and speaker:
                segs.append({"speaker": speaker, "text": " ".join(buf)})
                buf = []
            speaker = m.group(1)
            if m.group(2).strip():
                buf.append(m.group(2).strip())
        else:
            if line.startswith("- **") or line.startswith("*"):
                continue
            buf.append(line)
            if not speaker:
                speaker = "主播"
    if buf and speaker:
        segs.append({"speaker": speaker, "text": " ".join(buf)})
    return segs


def voice_for(speaker, custom=None):
    if custom:
        return custom
    return VOICE_B if "B" in speaker else VOICE_A


def synth(segs, output, custom_voice=None):
    has_tts = os.system("which edge-tts >/dev/null 2>&1") == 0
    if not has_tts:
        print("⚠️ edge-tts 未安装，运行: pip install edge-tts")
        print(f"📝 共 {len(segs)} 个片段待转换")
        for s in segs:
            print(f"  [{s['speaker']}] {s['text'][:50]}...")
        return {"segments": len(segs), "output": None, "error": "edge-tts not installed"}

    tmps = []
    for i, s in enumerate(segs):
        if not s["text"].strip():
            continue
        tmp = f"/tmp/pod_{i:03d}.mp3"
        v = voice_for(s["speaker"], custom_voice)
        try:
            subprocess.run(
                ["edge-tts", "--voice", v, "--text", s["text"], "--write-media", tmp],
                capture_output=True, timeout=60, check=True
            )
            tmps.append(tmp)
        except Exception as e:
            print(f"⚠️ 片段{i}失败: {e}")

    if not tmps:
        return {"segments": len(segs), "output": None, "error": "no segments generated"}

    has_ffmpeg = os.system("which ffmpeg >/dev/null 2>&1") == 0
    if has_ffmpeg and len(tmps) > 1:
        lst = "/tmp/pod_list.txt"
        with open(lst, "w") as f:
            for t in tmps:
                f.write(f"file '{t}'\n")
        subprocess.run(["ffmpeg", "-y", "-f", "concat", "-safe", "0", "-i", lst, "-c", "copy", output],
                       capture_output=True, timeout=120)
    elif tmps:
        import shutil
        shutil.copy2(tmps[0], output)

    # Cleanup
    for t in tmps:
        try:
            os.remove(t)
        except:
            pass

    ok = os.path.exists(output)
    return {"segments": len(segs), "output": output if ok else None, "success": ok}


if __name__ == "__main__":
    p = argparse.ArgumentParser()
    p.add_argument("--script", required=True)
    p.add_argument("--voice", default=None)
    p.add_argument("--output", default="podcast.mp3")
    p.add_argument("--dry-run", action="store_true")
    a = p.parse_args()

    text = open(a.script, encoding="utf-8").read()
    segs = parse(text)
    if not segs:
        print("Error: 未找到内容片段", file=sys.stderr)
        sys.exit(1)

    print(f"📝 解析到 {len(segs)} 个片段")
    if a.dry_run:
        for s in segs:
            print(f"  [{s['speaker']}] {s['text'][:60]}...")
        sys.exit(0)

    print("🎙️ 生成音频中...")
    r = synth(segs, a.output, a.voice)
    print(json.dumps(r, indent=2, ensure_ascii=False))
    if r.get("output"):
        print(f"\n✅ 音频: {r['output']}")

```

### scripts/stats.py

```python
#!/usr/bin/env python3
"""Podcast generation statistics and batch processing."""

import argparse, json, os, sys
from datetime import datetime

STATS_FILE = os.path.expanduser("~/.openclaw/workspace/podcast-generator/data/stats.json")


def load():
    os.makedirs(os.path.dirname(STATS_FILE), exist_ok=True)
    if os.path.exists(STATS_FILE):
        return json.load(open(STATS_FILE))
    return {"generations": [], "total_audio_seconds": 0, "total_calls": 0}


def save(data):
    os.makedirs(os.path.dirname(STATS_FILE), exist_ok=True)
    json.dump(data, open(STATS_FILE, "w"), indent=2, ensure_ascii=False)


def log_generation(title, format_type, audio_seconds, cost=0.001):
    data = load()
    entry = {
        "id": len(data["generations"]) + 1,
        "title": title,
        "format": format_type,
        "audio_seconds": audio_seconds,
        "cost": cost,
        "timestamp": datetime.now().isoformat(),
    }
    data["generations"].append(entry)
    data["total_audio_seconds"] += audio_seconds
    data["total_calls"] += 1
    # Keep last 100
    data["generations"] = data["generations"][-100:]
    save(data)
    return entry


def get_stats():
    data = load()
    total_cost = sum(g["cost"] for g in data["generations"])
    
    # Format breakdown
    formats = {}
    for g in data["generations"]:
        fmt = g["format"]
        formats[fmt] = formats.get(fmt, 0) + 1
    
    return {
        "total_generations": data["total_calls"],
        "total_audio_seconds": data["total_audio_seconds"],
        "total_audio_minutes": data["total_audio_seconds"] / 60,
        "total_cost": total_cost,
        "format_breakdown": formats,
        "recent": data["generations"][-10:],
    }


def format_output(data):
    lines = ["📊 Podcast Generator 统计", ""]
    lines.append(f"🎙️ 总生成次数：{data['total_generations']}")
    lines.append(f"⏱️ 总音频时长：{data['total_audio_minutes']:.1f} 分钟 ({data['total_audio_seconds']:.0f} 秒)")
    lines.append(f"💰 总花费：${data['total_cost']:.4f} USDT")
    lines.append("")
    
    if data["format_breakdown"]:
        lines.append("📋 格式分布:")
        for fmt, count in data["format_breakdown"].items():
            lines.append(f"   • {fmt}: {count} 次")
        lines.append("")
    
    if data["recent"]:
        lines.append("📜 最近 10 次:")
        for g in data["recent"]:
            lines.append(f"   [{g['id']}] {g['title'][:30]} - {g['format']} ({g['audio_seconds']}s)")
    
    return "\n".join(lines)


if __name__ == "__main__":
    p = argparse.ArgumentParser()
    p.add_argument("--action", choices=["show", "log"], default="show")
    p.add_argument("--title", default=None)
    p.add_argument("--format", default="solo")
    p.add_argument("--audio-seconds", type=int, default=0)
    p.add_argument("--json", action="store_true")
    a = p.parse_args()
    
    if a.action == "log":
        r = log_generation(a.title, a.format, a.audio_seconds)
        print(f"✅ 已记录：[{r['id']}] {r['title']} ({r['audio_seconds']}s)")
    else:
        data = get_stats()
        print(json.dumps(data, indent=2) if a.json else format_output(data))

```

### references/script-templates.md

```markdown
# Podcast Script Templates

## Solo (单人主播)
1. 开场白 (15s) → 话题引入 (1min) → 主体分段 (5-15min) → 总结 (1min) → 结尾 (15s)

## Dialogue (双人对谈)
1. 双方问候 → A抛观点/B补充 → 交替深入 → 各自总结 → 结尾

## Voice Options
| 角色 | Voice ID | 风格 |
|:---|:---|:---|
| 女主播 | zh-CN-XiaoxiaoNeural | 温柔专业 |
| 男主播 | zh-CN-YunxiNeural | 沉稳有力 |
| 男主播2 | zh-CN-YunjianNeural | 激情活力 |
| 女主播2 | zh-CN-XiaohanNeural | 知性优雅 |

```



---

## Skill Companion Files

> Additional files collected from the skill directory layout.

### _meta.json

```json
{
  "owner": "elevo11",
  "slug": "podcast-generator",
  "displayName": "Podcast Generator",
  "latest": {
    "version": "1.1.0",
    "publishedAt": 1772787440755,
    "commit": "https://github.com/openclaw/skills/commit/4060f71ccb91fc6069bc80e28eb66bdc6a544909"
  },
  "history": []
}

```