/~learn/s/e3a7b51d
> For Claude: Use ${SUPERPOWERS_SKILLS_ROOT}/skills/collaboration/executing-plans/SKILL.md to implement this plan task-by-task.
Goal: 为文档浏览模块添加"听书"功能 —— 底部固定播放器条，用火山引擎大模型 TTS（豆包同源精品音色）朗读文档正文，支持进度跟踪、续听、音色切换、语速调节、段落高亮跟随，跳过代码块/图表/URL 等非正文内容。
Architecture: 后端在 unified_server.py 新增 /api/tts HTTP 接口，调用火山引擎大模型 TTS（https://openspeech.bytedance.com/api/v1/tts，Bearer Token 鉴权，非流式 HTTP 一次性返回 MP3），用 SQLite (tts_cache.db) 做文本→音频缓存。前端在 DOC_PAGE 模板中嵌入悬浮播放器组件，提取清洗正文→按段落分 chunk→逐段请求/播放→进度条+段落高亮+续听（localStorage）。
Tech Stack: Python 3.12 标准库（urllib, sqlite3, json, hashlib, uuid, base64），火山引擎大模型 TTS HTTP API (v1)，原生 JavaScript + HTML5 Audio API，CSS 暗色主题（复用现有 CSS_BASE 变量）。
前置条件（需要用户手动完成一次）： 在火山引擎控制台 https://console.volcengine.com/speech/app 创建"语音合成"应用，获取 AppID 和 Access Token，填入 api_balance_config.json 的 volcengine.tts.appid 和 volcengine.tts.access_token。开通后有 20,000 字符免费试用额度。
---
| 操作 | 文件路径 | 说明 |
| ------ | --------- | ------ |
| 新建 | adata/aohan/quant/utils/tts_service.py | TTS 服务封装类（调用火山API + SQLite缓存） |
| 新建 | tests/test_tts_service.py | TTS 服务单元测试 |
| 修改 | api_balance_config.json（可选：example 文件） | 增加 tts 子配置 |
| 修改 | unified_server.py | 新增 3 个路由 + 注入播放器 JS/CSS 到 DOC_PAGE |
---
Files:
api_balance_config.example.jsontests/test_tts_service.py（基础骨架）Step 1: 在 example 配置里加入 tts 字段
在 api_balance_config.example.json 的 volcengine 对象中加入：
"tts": {
"enabled": false,
"appid": "your_appid_here",
"access_token": "your_access_token_here",
"default_voice": "BV700_V2_streaming",
"default_speed": 1.0,
"cluster": "volcano_tts"
}Step 2: 创建测试文件骨架
# tests/test_tts_service.py
import sys, os
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
import unittest
from unittest.mock import patch, MagicMockStep 3: 运行
cd /Users/aohan/PycharmProjects/adata
python -m pytest tests/test_tts_service.py -vExpected: 收集到 0 个测试，无报错。
---
Files:
adata/aohan/quant/utils/tts_service.pytests/test_tts_service.py这一步只实现缓存层（SQLite），不实现外部 API 调用。
Step 1: 写测试
在 tests/test_tts_service.py 添加：
import tempfile, sqlite3
class TestTTSServiceCache(unittest.TestCase):
def setUp(self):
self.tmpdir = tempfile.mkdtemp()
self.db_path = os.path.join(self.tmpdir, "tts_cache.db")
def _make_service(self):
from adata.aohan.quant.utils.tts_service import TTSService
return TTSService(
config={"appid": "test", "access_token": "test",
"cluster": "volcano_tts", "default_voice": "BV700_V2_streaming",
"default_speed": 1.0},
db_path=self.db_path,
)
def test_cache_key_is_deterministic(self):
svc = self._make_service()
k1 = svc._make_cache_key("你好", "BV700_V2_streaming", 1.0)
k2 = svc._make_cache_key("你好", "BV700_V2_streaming", 1.0)
k3 = svc._make_cache_key("你好", "BV701_streaming", 1.0)
self.assertEqual(k1, k2)
self.assertNotEqual(k1, k3)
self.assertIsInstance(k1, str)
self.assertEqual(len(k1), 64) # sha256 hex
def test_cache_set_and_get(self):
svc = self._make_service()
key = svc._make_cache_key("hello", "v", 1.0)
self.assertIsNone(svc._cache_get(key))
fake_mp3 = b"ID3\x03fake_mp3_data"
svc._cache_set(key, fake_mp3, text_length=5, voice_type="v")
got = svc._cache_get(key)
self.assertEqual(got, fake_mp3)
def test_cache_table_created_on_init(self):
svc = self._make_service()
conn = sqlite3.connect(self.db_path)
cur = conn.execute(
"SELECT name FROM sqlite_master WHERE type='table' AND name='tts_cache'"
)
self.assertIsNotNone(cur.fetchone())
conn.close()
if __name__ == "__main__":
unittest.main()Step 2: 运行测试 → 失败
python -m pytest tests/test_tts_service.py -vExpected: ImportError / AttributeError（TTSService 还不存在）。
Step 3: 实现缓存层最小代码
创建 adata/aohan/quant/utils/tts_service.py：
import os
import json
import uuid
import base64
import hashlib
import sqlite3
import urllib.request
import urllib.error
from typing import Optional
class TTSError(Exception):
pass
VOICE_PRESETS = [
{"id": "BV700_V2_streaming", "name": "灿灿 2.0", "category": "通用/多情感",
"desc": "自然女声，22种情感，适合通用朗读"},
{"id": "BV701_V2_streaming", "name": "擎苍 2.0", "category": "有声阅读/旁白",
"desc": "沉浸旁白男声，10种情感，最适合读书"},
{"id": "BV406_V2_streaming", "name": "梓梓 2.0", "category": "超自然女声",
"desc": "自然女声，7种情感"},
{"id": "BV407_V2_streaming", "name": "燃燃 2.0", "category": "超自然男声",
"desc": "活力男声"},
{"id": "BV001_V2_streaming", "name": "通用女声 2.0", "category": "通用",
"desc": "标准女声，免费"},
{"id": "BV123_streaming", "name": "阳光青年", "category": "有声阅读",
"desc": "阳光男声，多情感"},
{"id": "BV102_streaming", "name": "儒雅青年", "category": "有声阅读",
"desc": "儒雅男声旁白"},
{"id": "BV104_streaming", "name": "温柔淑女", "category": "有声阅读",
"desc": "温柔女声旁白"},
]
class TTSService:
TTS_URL = "https://openspeech.bytedance.com/api/v1/tts"
MAX_TEXT_BYTES = 900 # 留余量，API限制1024字节
def __init__(self, config: dict, db_path: Optional[str] = None):
self.appid = config.get("appid", "")
self.token = config.get("access_token", "")
self.cluster = config.get("cluster", "volcano_tts")
self.default_voice = config.get("default_voice", "BV700_V2_streaming")
self.default_speed = float(config.get("default_speed", 1.0))
self.enabled = bool(self.appid and self.token)
if db_path is None:
db_path = os.path.join(
os.path.dirname(os.path.abspath(__file__)),
"..", "db", "tts_cache.db"
)
self.db_path = os.path.abspath(db_path)
os.makedirs(os.path.dirname(self.db_path), exist_ok=True)
self._init_db()
def _init_db(self):
conn = sqlite3.connect(self.db_path)
try:
conn.execute("""
CREATE TABLE IF NOT EXISTS tts_cache (
cache_key TEXT PRIMARY KEY,
audio_blob BLOB NOT NULL,
text_length INTEGER,
voice_type TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
conn.execute("CREATE INDEX IF NOT EXISTS idx_tts_cache_voice ON tts_cache(voice_type)")
conn.execute("CREATE INDEX IF NOT EXISTS idx_tts_cache_created ON tts_cache(created_at)")
conn.commit()
finally:
conn.close()
@staticmethod
def _make_cache_key(text: str, voice_type: str, speed: float) -> str:
raw = f"{text}|{voice_type}|{round(speed, 2)}"
return hashlib.sha256(raw.encode("utf-8")).hexdigest()
def _cache_get(self, key: str) -> Optional[bytes]:
conn = sqlite3.connect(self.db_path)
try:
cur = conn.execute("SELECT audio_blob FROM tts_cache WHERE cache_key=?", (key,))
row = cur.fetchone()
return row[0] if row else None
finally:
conn.close()
def _cache_set(self, key: str, audio: bytes, text_length: int, voice_type: str):
conn = sqlite3.connect(self.db_path)
try:
conn.execute(
"INSERT OR REPLACE INTO tts_cache (cache_key,audio_blob,text_length,voice_type) VALUES (?,?,?,?)",
(key, audio, text_length, voice_type)
)
conn.commit()
finally:
conn.close()
def get_voices(self):
return VOICE_PRESETS
def get_stats(self) -> dict:
conn = sqlite3.connect(self.db_path)
try:
cur = conn.execute("SELECT COUNT(*), COALESCE(SUM(text_length),0) FROM tts_cache")
cnt, total_chars = cur.fetchone()
return {"cached_items": cnt, "total_chars_cached": total_chars}
finally:
conn.close()
def synthesize(self, text: str, voice_type: Optional[str] = None,
speed: Optional[float] = None) -> bytes:
raise NotImplementedError("Will be implemented in next task")Step 4: 运行测试 → 通过
python -m pytest tests/test_tts_service.py -vExpected: 3 passed.
Step 5: Commit
git add adata/aohan/quant/utils/tts_service.py tests/test_tts_service.py api_balance_config.example.json
git commit -m "feat(tts): TTSService cache layer with SQLite"---
Files:
adata/aohan/quant/utils/tts_service.pytests/test_tts_service.pyStep 1: 写测试
class TestTextChunking(unittest.TestCase):
def setUp(self):
from adata.aohan.quant.utils.tts_service import TTSService
self.svc = TTSService(
config={"appid":"x","access_token":"x"},
db_path=os.path.join(tempfile.mkdtemp(),"t.db")
)
def test_short_text_single_chunk(self):
chunks = self.svc._split_text("你好世界")
self.assertEqual(len(chunks), 1)
self.assertEqual(chunks[0], "你好世界")
def test_long_text_split_by_punctuation(self):
text = "第一句。" * 200 # 约 800 字节
chunks = self.svc._split_text(text)
self.assertGreater(len(chunks), 1)
for c in chunks:
self.assertLessEqual(len(c.encode("utf-8")), self.svc.MAX_TEXT_BYTES)
self.assertEqual("".join(chunks), text)
def test_empty_text_returns_empty(self):
self.assertEqual(self.svc._split_text(""), [])
self.assertEqual(self.svc._split_text(" \n "), [])
def test_split_respects_paragraph(self):
text = "第一段内容。" * 50 + "\n\n" + "第二段内容。" * 50
chunks = self.svc._split_text(text)
joined = "".join(chunks)
# 不丢内容
self.assertEqual(joined.replace("\n",""), text.replace("\n",""))
Step 2: 运行测试 → 失败（_split_text 不存在）
Step 3: 实现 _split_text
在 TTSService 类中添加：
_SENTENCE_ENDINGS = "。！？!?；;…\n"
def _split_text(self, text: str) -> list:
text = text.strip()
if not text:
return []
chunks, buf, buf_bytes = [], [], 0
for ch in text:
buf.append(ch)
buf_bytes += len(ch.encode("utf-8"))
if ch in self._SENTENCE_ENDINGS and buf_bytes > 100:
if buf_bytes >= self.MAX_TEXT_BYTES:
chunks.append("".join(buf))
buf, buf_bytes = [], 0
elif buf_bytes >= self.MAX_TEXT_BYTES:
chunks.append("".join(buf))
buf, buf_bytes = [], 0
if buf:
if chunks and buf_bytes < 200:
chunks[-1] += "".join(buf)
else:
chunks.append("".join(buf))
return [c.strip() for c in chunks if c.strip()]Step 4: 运行测试 → 通过
python -m pytest tests/test_tts_service.py -vExpected: 7 passed.
Step 5: Commit
git add adata/aohan/quant/utils/tts_service.py tests/test_tts_service.py
git commit -m "feat(tts): add text chunking by sentence boundary"---
Files:
adata/aohan/quant/utils/tts_service.pytests/test_tts_service.pyStep 1: 写测试（mock 网络层）
from unittest.mock import patch, MagicMock
import base64, json as _json
class TestTTSSynthesize(unittest.TestCase):
def setUp(self):
from adata.aohan.quant.utils.tts_service import TTSService
self.tmpdir = tempfile.mkdtemp()
self.svc = TTSService(
config={"appid":"myapp","access_token":"mytoken","cluster":"volcano_tts",
"default_voice":"BV700_V2_streaming"},
db_path=os.path.join(self.tmpdir,"t.db")
)
@patch("adata.aohan.quant.utils.tts_service.urllib.request.urlopen")
def test_synthesize_calls_api_and_returns_mp3(self, mock_urlopen):
fake_mp3 = b"ID3fake_mp3_binary_data"
fake_resp = MagicMock()
fake_resp.read.return_value = _json.dumps({
"code": 3000,
"message": "success",
"data": base64.b64encode(fake_mp3).decode("ascii"),
"reqid": "abc123"
}).encode("utf-8")
fake_resp.__enter__ = lambda s: s
fake_resp.__exit__ = MagicMock(return_value=False)
mock_urlopen.return_value = fake_resp
result = self.svc.synthesize("你好世界")
self.assertEqual(result, fake_mp3)
call_args = mock_urlopen.call_args
req = call_args[0][0]
self.assertEqual(req.full_url, "https://openspeech.bytedance.com/api/v1/tts")
self.assertIn("Authorization", req.headers)
self.assertTrue(req.headers["Authorization"].startswith("Bearer;"))
body = _json.loads(req.data.decode("utf-8"))
self.assertEqual(body["app"]["appid"], "myapp")
self.assertEqual(body["app"]["cluster"], "volcano_tts")
self.assertEqual(body["audio"]["voice_type"], "BV700_V2_streaming")
self.assertEqual(body["audio"]["encoding"], "mp3")
self.assertEqual(body["request"]["operation"], "query")
self.assertEqual(body["request"]["text"], "你好世界")
@patch("adata.aohan.quant.utils.tts_service.urllib.request.urlopen")
def test_synthesize_caches_result(self, mock_urlopen):
fake_mp3 = b"ID3cached"
fake_resp = MagicMock()
fake_resp.read.return_value = _json.dumps({
"code": 3000, "data": base64.b64encode(fake_mp3).decode()
}).encode("utf-8")
fake_resp.__enter__ = lambda s: s
fake_resp.__exit__ = MagicMock(return_value=False)
mock_urlopen.return_value = fake_resp
self.svc.synthesize("测试缓存")
self.svc.synthesize("测试缓存")
self.assertEqual(mock_urlopen.call_count, 1)
@patch("adata.aohan.quant.utils.tts_service.urllib.request.urlopen")
def test_synthesize_api_error_raises(self, mock_urlopen):
from adata.aohan.quant.utils.tts_service import TTSError
fake_resp = MagicMock()
fake_resp.read.return_value = _json.dumps({
"code": 4001, "message": "invalid token"
}).encode("utf-8")
fake_resp.__enter__ = lambda s: s
fake_resp.__exit__ = MagicMock(return_value=False)
mock_urlopen.return_value = fake_resp
with self.assertRaises(TTSError):
self.svc.synthesize("err test")
def test_synthesize_disabled_raises(self):
from adata.aohan.quant.utils.tts_service import TTSError
svc = TTSService(config={}, db_path=os.path.join(self.tmpdir,"t2.db"))
with self.assertRaises(TTSError):
svc.synthesize("x")Step 2: 运行测试 → 失败
Step 3: 实现 synthesize 和 _call_api
把 synthesize 方法改为：
def synthesize(self, text: str, voice_type: Optional[str] = None,
speed: Optional[float] = None) -> bytes:
if not self.enabled:
raise TTSError("TTS 未启用：请在 api_balance_config.json 配置 volcengine.tts.appid 和 access_token")
voice_type = voice_type or self.default_voice
speed = float(speed) if speed is not None else self.default_speed
text = text.strip()
if not text:
raise TTSError("文本为空")
if len(text.encode("utf-8")) > self.MAX_TEXT_BYTES:
raise TTSError(f"文本过长（{len(text.encode('utf-8'))}字节），请分段调用")
key = self._make_cache_key(text, voice_type, speed)
cached = self._cache_get(key)
if cached is not None:
return cached
audio = self._call_api(text, voice_type, speed)
self._cache_set(key, audio, text_length=len(text), voice_type=voice_type)
return audio
def _call_api(self, text: str, voice_type: str, speed: float) -> bytes:
payload = {
"app": {
"appid": self.appid,
"token": "tts_token",
"cluster": self.cluster,
},
"user": {"uid": "aohan_quant"},
"audio": {
"voice_type": voice_type,
"encoding": "mp3",
"speed_ratio": round(speed, 1),
"rate": 24000,
},
"request": {
"reqid": str(uuid.uuid4()),
"text": text,
"operation": "query",
},
}
body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
req = urllib.request.Request(
self.TTS_URL,
data=body,
headers={
"Content-Type": "application/json",
"Authorization": f"Bearer;{self.token}",
},
method="POST",
)
try:
with urllib.request.urlopen(req, timeout=30) as resp:
resp_data = json.loads(resp.read().decode("utf-8"))
except urllib.error.HTTPError as e:
err_body = e.read().decode("utf-8", errors="ignore")
raise TTSError(f"TTS HTTP {e.code}: {err_body[:200]}") from e
except Exception as e:
raise TTSError(f"TTS 网络错误: {e}") from e
code = resp_data.get("code", -1)
if code != 3000:
raise TTSError(f"TTS API 错误 code={code}: {resp_data.get('message','')}")
audio_b64 = resp_data.get("data", "")
if not audio_b64:
raise TTSError("TTS 返回音频为空")
return base64.b64decode(audio_b64)Step 4: 运行测试 → 通过
python -m pytest tests/test_tts_service.py -vExpected: 10 passed。
Step 5: Commit
git add adata/aohan/quant/utils/tts_service.py tests/test_tts_service.py
git commit -m "feat(tts): implement Volcengine TTS API call with caching"---
Files:
unified_server.py
Step 1: 找到 DOC_PAGE 路由处理函数
在 unified_server.py 中搜索 def do_GET 和处理 /~learn/ 的代码段（约 L4297-L4326）。在同文件中添加路由逻辑。
Step 2: 在 import 区（文件顶部）添加
from adata.aohan.quant.utils.tts_service import TTSService, TTSError
（若 tts_service 导入失败则 TTSService = None，保持容错）
Step 3: 添加配置加载辅助函数
靠近文件顶部或工具函数区，添加：
def _get_tts_service():
global _tts_svc
if '_tts_svc' in globals() and _tts_svc is not None:
return _tts_svc
try:
from adata.aohan.quant.utils.api_balance_monitor import load_config
cfg = load_config().get("volcengine", {}).get("tts", {})
except Exception:
cfg = {}
if not cfg.get("enabled", False) or not cfg.get("appid"):
return None
_tts_svc = TTSService(cfg)
return _tts_svc
Step 4: 在 do_GET 中添加 3 个路由
在合适位置（文档相关路由附近）添加：
# /api/tts/voices
elif path == "/api/tts/voices":
svc = _get_tts_service()
self._send_json(200, {
"enabled": svc is not None,
"voices": svc.get_voices() if svc else [],
"default_voice": svc.default_voice if svc else None,
"default_speed": svc.default_speed if svc else 1.0,
})
# /api/tts?text=xxx&voice=xxx&speed=1.0
elif path.startswith("/api/tts?") or path == "/api/tts":
svc = _get_tts_service()
if svc is None:
self._send_json(503, {"error": "TTS 未配置，请联系管理员在 api_balance_config.json 中配置 volcengine.tts"})
return
from urllib.parse import urlparse, parse_qs
qs = parse_qs(urlparse(self.path).query)
text = qs.get("text", [""])[0]
voice = qs.get("voice", [None])[0]
speed = float(qs.get("speed", [1.0])[0])
try:
audio = svc.synthesize(text, voice_type=voice, speed=speed)
except TTSError as e:
self._send_json(500, {"error": str(e)})
return
self.send_response(200)
self.send_header("Content-Type", "audio/mpeg")
self.send_header("Content-Length", str(len(audio)))
self.send_header("Cache-Control", "public, max-age=31536000, immutable")
self.send_header("Access-Control-Allow-Origin", "*")
self.end_headers()
self.wfile.write(audio)
# /api/tts/stats
elif path == "/api/tts/stats":
svc = _get_tts_service()
stats = svc.get_stats() if svc else {"cached_items": 0, "total_chars_cached": 0}
self._send_json(200, {"enabled": svc is not None, **stats})
注意：_send_json 如果项目中不存在，需要确认现有的 JSON 响应写法。在现有代码中搜索 "application/json" 看看怎么发送的，复用同样模式。
Step 5: 手动测试路由
启动服务：
cd /Users/aohan/PycharmProjects/adata
.venv/bin/python unified_server.py然后在另一个终端：
curl -s http://localhost:8765/api/tts/voices | head -c 500
Expected: 返回 JSON，包含 enabled: false（因为还没配置 appid）和空 voices 或默认 voices。
Step 6: Commit
git add unified_server.py
git commit -m "feat(tts): add /api/tts, /api/tts/voices, /api/tts/stats routes"---
Files:
unified_server.py（DOC_PAGE 模板的 <style> 区块）
Step 1: 在 DOC_PAGE 的 </style> 之前（响应式 @media 之前）插入播放器 CSS
将以下 CSS 添加到 DOC_PAGE 模板字符串中，在 /* ── 响应式优化 ── */ 注释之前：
/* ── TTS 听书播放器 ── */
body{{padding-bottom:72px;}}
.tts-player{{
position:fixed;left:0;right:0;bottom:0;z-index:999;
background:rgba(13,18,37,.92);backdrop-filter:blur(12px);-webkit-backdrop-filter:blur(12px);
border-top:1px solid var(--border);
padding:10px 16px;display:flex;align-items:center;gap:12px;
font-size:13px;color:var(--text);
transform:translateY(100%);transition:transform .3s ease;
}}
.tts-player.show{{transform:translateY(0);}}
.tts-play-btn{{
width:40px;height:40px;border-radius:50%;flex-shrink:0;
background:linear-gradient(135deg,#5b9cf5,#a78bfa);border:none;color:#fff;
font-size:18px;cursor:pointer;display:flex;align-items:center;justify-content:center;
transition:transform .15s;
}}
.tts-play-btn:hover{{transform:scale(1.08);}}
.tts-play-btn:disabled{{opacity:.5;cursor:not-allowed;}}
.tts-info{{flex:1;min-width:0;}}
.tts-time{{font-family:monospace;font-size:11px;color:var(--muted);margin-bottom:4px;white-space:nowrap;}}
.tts-bar{{
height:4px;background:var(--border);border-radius:2px;cursor:pointer;position:relative;
}}
.tts-bar-fill{{height:100%;background:linear-gradient(90deg,#5b9cf5,#a78bfa);border-radius:2px;width:0%;transition:width .2s;}}
.tts-chunk-title{{font-size:12px;color:var(--muted);white-space:nowrap;overflow:hidden;text-overflow:ellipsis;max-width:300px;}}
.tts-settings-btn{{
background:none;border:1px solid var(--border);color:var(--muted);border-radius:8px;
padding:6px 10px;cursor:pointer;font-size:16px;transition:all .2s;
}}
.tts-settings-btn:hover{{border-color:var(--accent);color:var(--accent);}}
.tts-panel{{
position:fixed;right:16px;bottom:72px;background:var(--bg2);border:1px solid var(--border);
border-radius:12px;padding:14px 16px;width:280px;display:none;z-index:1000;
box-shadow:0 8px 32px rgba(0,0,0,.5);
}}
.tts-panel.show{{display:block;}}
.tts-panel h4{{margin:0 0 10px;font-size:13px;color:var(--muted);font-weight:600;}}
.tts-panel label{{display:block;font-size:12px;color:var(--muted);margin:8px 0 4px;}}
.tts-panel select,.tts-panel input[type=range]{{
width:100%;background:var(--bg);border:1px solid var(--border);color:var(--text);
border-radius:6px;padding:6px 8px;font-size:12px;box-sizing:border-box;
}}
.tts-panel .tts-speed-val{{font-size:11px;color:var(--accent);float:right;}}
.tts-error{{color:#f87171;font-size:11px;margin-top:6px;}}
.tts-loading{{color:var(--accent);font-size:11px;}}
.tts-toggle-btn{{
display:inline-flex;align-items:center;gap:6px;padding:6px 12px;border-radius:8px;
background:rgba(91,156,245,.15);border:1px solid rgba(91,156,245,.3);color:var(--accent);
font-size:12px;font-weight:600;cursor:pointer;transition:all .2s;
}}
.tts-toggle-btn:hover{{background:rgba(91,156,245,.25);}}
.tts-highlight{{
background:rgba(91,156,245,.18);border-radius:4px;padding:2px 0;
transition:background-color .3s;
}}
@media(max-width:640px){{
body{{padding-bottom:64px;}}
.tts-player{{padding:8px 10px;gap:8px;font-size:12px;}}
.tts-play-btn{{width:36px;height:36px;font-size:16px;}}
.tts-chunk-title{{display:none;}}
.tts-panel{{right:8px;left:8px;width:auto;bottom:64px;}}
}}Step 2: 在 DOC_PAGE 的 toolbar 里加启动按钮
把 .doc-toolbar 里的 share 按钮旁边加一个听书按钮。找到：
<button class="share-btn" id="shareBtn" onclick="shareAsImage()">📷 分享为图片</button>在它前面加：
<button class="tts-toggle-btn" id="ttsToggleBtn" onclick="ttsToggle()">🔊 听书</button>
Step 3: 在 DOC_PAGE 的 </div><!-- end .page --> 之前（watermark 之后）插入播放器 HTML
找到 .doc-watermark 的 </div> 后面（约 L1651 附近），加入：
<div class="tts-panel" id="ttsPanel">
<h4>🔊 播放设置</h4>
<label>🎙️ 音色</label>
<select id="ttsVoiceSelect"></select>
<label>🏃 语速 <span class="tts-speed-val" id="ttsSpeedVal">1.0x</span></label>
<input type="range" id="ttsSpeedRange" min="0.6" max="1.8" step="0.1" value="1.0">
<div class="tts-error" id="ttsError" style="display:none;"></div>
<div class="tts-loading" id="ttsLoading" style="display:none;">⏳ 合成中...</div>
</div>
<div class="tts-player" id="ttsPlayer">
<button class="tts-play-btn" id="ttsPlayBtn" onclick="ttsPlayPause()">▶</button>
<div class="tts-info">
<div class="tts-time">
<span id="ttsCurTime">00:00</span> / <span id="ttsTotalTime">00:00</span>
<span class="tts-chunk-title" id="ttsChunkTitle"></span>
</div>
<div class="tts-bar" id="ttsBar" onclick="ttsSeek(event)"><div class="tts-bar-fill" id="ttsBarFill"></div></div>
</div>
<button class="tts-settings-btn" id="ttsSettingsBtn" onclick="ttsTogglePanel()">⚙️</button>
</div>Step 4: 验证无语法错误
.venv/bin/python -c "import py_compile; py_compile.compile('unified_server.py', doraise=True); print('OK')"
Expected: OK
Step 5: Commit
git add unified_server.py
git commit -m "feat(tts): add player CSS and HTML skeleton to DOC_PAGE"---
Files:
unified_server.py（DOC_PAGE 模板的 <script> 区块）
Step 1: 在现有 </script> 之前插入 TTS JS 代码
找到 DOC_PAGE 模板末尾（约 L1749 附近 })(); 后面），加入以下完整 JS 代码。注意模板里所有 { 和 } 需要双写为 {{ 和 }}（除了 {title}、{back_url} 等原本就用 format 替换的占位符）。
⚠️ 重点注意：写在 Python 模板字符串里的 JS，需要把 JS 里的所有 { → {{，} → }}，否则 Python str.format() 会报错。代码块以 /* TTS-START */ 标记开始、/* TTS-END */ 标记结束，便于后续维护。
/* TTS-START */
var tts = {{
audio: null,
chunks: [], // [{text, el, duration, url}]
chunkIndex: 0,
voice: null,
speed: 1.0,
enabled: false,
playing: false,
progressKey: 'tts_progress:' + location.pathname,
}};
function ttsInit() {{
fetch('/api/tts/voices').then(function(r){{return r.json();}}).then(function(cfg){{
tts.enabled = cfg.enabled;
var btn = document.getElementById('ttsToggleBtn');
if (!btn) return;
if (!cfg.enabled) {{
btn.textContent = '🔊 听书(未配置)';
btn.disabled = true;
btn.title = 'TTS 未配置';
return;
}}
tts.voice = cfg.default_voice;
tts.speed = cfg.default_speed;
var sel = document.getElementById('ttsVoiceSelect');
cfg.voices.forEach(function(v){{
var opt = document.createElement('option');
opt.value = v.id; opt.textContent = v.name + '（' + v.category + '）';
if (v.id === cfg.default_voice) opt.selected = true;
sel.appendChild(opt);
}});
sel.onchange = function(){{ tts.voice = this.value; ttsClearCache(); ttsSaveProgress(); }};
var sr = document.getElementById('ttsSpeedRange');
var sv = document.getElementById('ttsSpeedVal');
sr.value = cfg.default_speed;
sr.oninput = function(){{
tts.speed = parseFloat(this.value);
sv.textContent = tts.speed.toFixed(1) + 'x';
if (tts.audio) tts.audio.playbackRate = tts.speed;
ttsSaveProgress();
}};
// load progress
try {{
var p = JSON.parse(localStorage.getItem(tts.progressKey) || '{{}}');
if (p.voice) {{ tts.voice = p.voice; sel.value = p.voice; }}
if (p.speed) {{ tts.speed = p.speed; sr.value = p.speed; sv.textContent = p.speed.toFixed(1)+'x'; }}
if (p.chunk_index || p.position_sec) {{
btn.textContent = '🔊 续听 ' + (p.title || '');
}}
}} catch(e){{}}
// extract chunks after mermaid renders
setTimeout(ttsExtractChunks, 500);
}})['catch'](function(){{
var btn = document.getElementById('ttsToggleBtn');
if (btn){{ btn.textContent = '🔊 听书(不可用)'; btn.disabled = true; }}
}});
}}
function ttsExtractChunks(){{
var body = document.getElementById('docBody');
if (!body) return;
tts.chunks = [];
var els = body.querySelectorAll('h2,h3,h4,h5,p,blockquote,li');
els.forEach(function(el){{
// skip code blocks
if (el.closest('pre') || el.closest('code') || el.closest('.mermaid')) return;
// clone and clean
var clone = el.cloneNode(true);
clone.querySelectorAll('pre,code,.mermaid,img,table,script,style').forEach(function(n){{n.remove();}});
clone.querySelectorAll('a').forEach(function(a){{
a.replaceWith(document.createTextNode(a.textContent));
}});
var text = clone.textContent.trim();
text = text.replace(/https?:\/\/[^\s]+/g, '');
text = text.replace(/\s+/g, ' ').trim();
if (!text || text.length < 2) return;
tts.chunks.push({{text: text, el: el, duration: 0, url: null, audioLoaded: false}});
}});
}}
function ttsClearHighlight(){{
document.querySelectorAll('.tts-highlight').forEach(function(el){{
el.classList.remove('tts-highlight');
}});
}}
function ttsHighlight(idx){{
ttsClearHighlight();
var c = tts.chunks[idx];
if (c && c.el){{
c.el.classList.add('tts-highlight');
c.el.scrollIntoView({{behavior:'smooth', block:'center'}});
var t = document.getElementById('ttsChunkTitle');
if (t) t.textContent = c.el.tagName.match(/H[2-5]/) ? c.text.slice(0,20) : '';
}}
}}
function ttsToggle(){{
var p = document.getElementById('ttsPlayer');
if (!p) return;
if (p.classList.contains('show')){{
ttsStop(); p.classList.remove('show');
}} else {{
p.classList.add('show');
if (tts.chunks.length === 0) ttsExtractChunks();
// resume
try {{
var prog = JSON.parse(localStorage.getItem(tts.progressKey) || '{{}}');
if (typeof prog.chunk_index === 'number' && prog.chunk_index < tts.chunks.length){{
tts.chunkIndex = prog.chunk_index;
}}
}}catch(e){{}}
}}
}}
function ttsTogglePanel(){{
document.getElementById('ttsPanel').classList.toggle('show');
}}
function ttsShowError(msg){{
var e = document.getElementById('ttsError');
e.textContent = msg; e.style.display = 'block';
setTimeout(function(){{ e.style.display='none'; }}, 5000);
}}
function ttsSetLoading(on){{
document.getElementById('ttsLoading').style.display = on ? 'block' : 'none';
}}
function ttsPlayPause(){{
if (tts.playing) {{ ttsPause(); }} else {{ ttsPlay(); }}
}}
function ttsPause(){{
if (tts.audio) tts.audio.pause();
tts.playing = false;
document.getElementById('ttsPlayBtn').textContent = '▶';
}}
function ttsStop(){{
if (tts.audio){{ tts.audio.pause(); tts.audio.src=''; tts.audio=null; }}
tts.playing = false;
document.getElementById('ttsPlayBtn').textContent = '▶';
document.getElementById('ttsBarFill').style.width = '0%';
ttsClearHighlight();
}}
function ttsPlay(){{
if (!tts.chunks.length) return;
tts.playing = true;
document.getElementById('ttsPlayBtn').textContent = '⏸';
ttsPlayChunk(tts.chunkIndex);
}}
function ttsPlayChunk(idx){{
if (idx >= tts.chunks.length) {{ ttsStop(); return; }}
tts.chunkIndex = idx;
ttsHighlight(idx);
var chunk = tts.chunks[idx];
if (chunk.audioLoaded) {{
ttsPlayAudio(chunk.url);
return;
}}
ttsSetLoading(true);
var url = '/api/tts?text=' + encodeURIComponent(chunk.text) + '&voice=' + encodeURIComponent(tts.voice) + '&speed=' + tts.speed;
fetch(url).then(function(r){{
if (!r.ok) {{ return r.json().then(function(j){{ throw new Error(j.error || '合成失败'); }}); }}
return r.blob();
}}).then(function(blob){{
chunk.url = URL.createObjectURL(blob);
chunk.audioLoaded = true;
ttsSetLoading(false);
ttsPlayAudio(chunk.url);
// prefetch next
if (idx+1 < tts.chunks.length && !tts.chunks[idx+1].audioLoaded){{
var next = tts.chunks[idx+1];
var nu = '/api/tts?text=' + encodeURIComponent(next.text) + '&voice=' + encodeURIComponent(tts.voice) + '&speed=' + tts.speed;
fetch(nu).then(function(r){{ if(r.ok) return r.blob(); else throw 0; }}).then(function(b){{
next.url = URL.createObjectURL(b); next.audioLoaded = true;
}})['catch'](function(){{}});
}}
}})['catch'](function(e){{
ttsSetLoading(false);
ttsShowError(e.message || '合成失败');
tts.chunkIndex++;
if (tts.chunkIndex < tts.chunks.length) ttsPlayChunk(tts.chunkIndex);
}});
}}
function ttsPlayAudio(url){{
if (tts.audio) {{ try{{ tts.audio.pause(); }}catch(e){{}} }}
tts.audio = new Audio(url);
tts.audio.playbackRate = tts.speed;
try {{
var prog = JSON.parse(localStorage.getItem(tts.progressKey) || '{{}}');
if (prog.chunk_index === tts.chunkIndex && prog.position_sec > 0){{
tts.audio.currentTime = prog.position_sec;
}}
}}catch(e){{}}
tts.audio.ontimeupdate = function(){{
var pct = (tts.audio.currentTime / (tts.audio.duration || 1)) * 100;
document.getElementById('ttsBarFill').style.width = Math.min(100,pct) + '%';
document.getElementById('ttsCurTime').textContent = ttsFmtTime(tts.totalOffset() + tts.audio.currentTime);
ttsSaveProgress();
}};
tts.audio.onloadedmetadata = function(){{
document.getElementById('ttsTotalTime').textContent = ttsFmtTime(ttsTotalDuration());
}};
tts.audio.onended = function(){{
tts.chunkIndex++;
if (tts.chunkIndex < tts.chunks.length) {{
ttsPlayChunk(tts.chunkIndex);
}} else {{
ttsStop();
localStorage.removeItem(tts.progressKey);
}}
}};
tts.audio.onerror = function(){{
ttsShowError('播放失败');
tts.chunkIndex++;
if (tts.chunkIndex < tts.chunks.length) ttsPlayChunk(tts.chunkIndex);
}};
tts.audio.play()['catch'](function(e){{ ttsShowError('播放失败：'+e.message); }});
}}
function ttsTotalOffset(){{
var t = 0;
for (var i=0;i<tts.chunkIndex;i++){{ if(tts.chunks[i].audio && tts.chunks[i].audio.duration) t += tts.chunks[i].audio.duration; }}
return t;
}}
function ttsTotalDuration(){{
var t = 0;
tts.chunks.forEach(function(c){{ if(c.audioLoaded && tts.audio) {{ if(c===tts.chunks[tts.chunkIndex]) t += tts.audio.duration||0; else if(c.audio && c.audio.duration) t += c.audio.duration; }} }});
if (tts.audio && tts.audio.duration) return tts.audio.duration;
return tts.chunks.length * 8; // rough estimate ~4.5 chinese chars/sec
}}
function ttsFmtTime(s){{
s = Math.floor(s||0);
var m = Math.floor(s/60); s = s%60;
return (m<10?'0':'')+m+':'+(s<10?'0':'')+s;
}}
function ttsSeek(e){{
var bar = document.getElementById('ttsBar');
var pct = (e.clientX - bar.getBoundingClientRect().left) / bar.offsetWidth;
// 简化：跳到按比例估算的 chunk
var idx = Math.floor(pct * tts.chunks.length);
idx = Math.max(0, Math.min(tts.chunks.length-1, idx));
tts.chunkIndex = idx;
ttsPlayChunk(idx);
}}
function ttsSaveProgress(){{
if (!tts.audio) return;
var data = {{
chunk_index: tts.chunkIndex,
position_sec: tts.audio.currentTime || 0,
voice: tts.voice,
speed: tts.speed,
title: document.getElementById('srcTitle')?.textContent?.slice(0,10) || '',
saved_at: Date.now()
}};
localStorage.setItem(tts.progressKey, JSON.stringify(data));
}}
function ttsClearCache(){{
// 清除已加载但同voice的缓存（简化：重载后新voice才生效）
tts.chunks.forEach(function(c){{ c.audioLoaded=false; c.url=null; }});
}}
setTimeout(ttsInit, 300);
/* TTS-END */
⚠️ 关键提醒：上面代码在放入 DOC_PAGE 模板字符串时，必须把所有 { 和 } 双写（已在上面示例里做了）。否则 Python 的 .format() 会把它们当作模板占位符，导致 KeyError。
Step 2: 编译验证
.venv/bin/python -c "import py_compile; py_compile.compile('unified_server.py', doraise=True); print('SYNTAX OK')"
Expected: SYNTAX OK。如果报 KeyError，说明有 { 没有双写。
Step 3: 启动服务验证
./restart.sh
# 或手动启动
.venv/bin/python unified_server.py打开一篇文档，应该看到右上角有"🔊 听书"按钮。如果 TTS 未配置，按钮会显示"🔊 听书(未配置)"且禁用。
Step 4: Commit
git add unified_server.py
git commit -m "feat(tts): add frontend audiobook player JS with chunks/progress/highlight"---
前置条件：需要用户先手动去火山引擎控制台操作：
Step 1: 编辑 api_balance_config.json
在 volcengine 字段下添加：
"tts": {
"enabled": true,
"appid": "你从控制台获取的APPID",
"access_token": "你从控制台获取的Access_Token",
"default_voice": "BV700_V2_streaming",
"default_speed": 1.0,
"cluster": "volcano_tts"
}Step 2: 重启并端到端测试
./restart.sh
浏览器访问任意一篇文档，例如 https://quant.aohan.net/~learn/... 或本地 http://localhost:8765/~learn/...：
Step 3: 发布到公网（根据项目铁律）
./run_web_daemon.sh restart
# 确认 cloudflared tunnel 在线访问 https://quant.aohan.net 验证线上功能。
Step 4: 最终 Commit
git add api_balance_config.json # 注意：如果有真实密钥，确保在.gitignore里
git commit -m "feat(tts): enable TTS with volcengine credentials"---
如果 TTS 未配置或 API 调用失败，自动降级到浏览器原生语音合成，保证基本可用：
实现思路：
ttsPlayChunk 里，如果 fetch 失败且错误是"未配置"或"503"，调用 ttsPlayChunkNative(idx)ttsPlayChunkNative(idx)：
``javascript
var u = new SpeechSynthesisUtterance(tts.chunks[idx].text);
u.lang = 'zh-CN';
u.rate = tts.speed;
u.onend = function(){ tts.chunkIndex++; ttsPlayChunk(tts.chunkIndex); };
window.speechSynthesis.speak(u);
`
这个功能可以放到 Task 8 完成后作为增量，如果不需要可跳过。
---
是敏感凭证，确保 api_balance_config.json 已在 .gitignore 中（目前已如此，example 文件用占位符）。 返回字符统计，后续可以加日调用上限（比如每天 5 万字符自动熔断）。（擎苍2.0，旁白-沉浸模式），这是最适合朗读文档的音色；可以在测试后对比切换。 冲突：前端 JS 里的大括号必须全部双写 {{ }}，这是本项目模板最大的坑，改动后必须 py_compile 验证。---
在未配置时返回 enabled:false，配置后返回音色列表 合成后返回 MP3 音频（用 curl -o test.mp3` 下载能播放）