Ubuntu
357 lines
11 KiB
Python
Executable File
357 lines
11 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
Audiobook Player - Web Application
|
|
Minimalist immersive reading + streaming TTS
|
|
Adaptive directory structure support
|
|
"""
|
|
|
|
import os
|
|
import json
|
|
import asyncio
|
|
import edge_tts
|
|
from flask import Flask, render_template, request, Response, send_file
|
|
from pathlib import Path
|
|
from typing import List, Dict, Optional
|
|
|
|
app = Flask(__name__, static_folder='static')
|
|
|
|
# Configuration
|
|
MEDIA_DIR = Path("/home/ubuntu/media")
|
|
DEFAULT_VOICE = "en-US-AriaNeural"
|
|
|
|
|
|
def load_source_config(source_path: Path) -> Optional[Dict]:
|
|
"""加载 source.json 配置（如果存在）"""
|
|
config_file = source_path / "source.json"
|
|
if config_file.exists():
|
|
try:
|
|
with open(config_file, 'r', encoding='utf-8') as f:
|
|
return json.load(f)
|
|
except:
|
|
pass
|
|
return None
|
|
|
|
|
|
def detect_structure_type(source_path: Path) -> str:
|
|
"""
|
|
自动检测目录结构类型
|
|
- flat: 直接存放 txt 文件
|
|
- content_folder: 子目录包含 content.txt
|
|
- multi_file_folder: 子目录包含多个 txt 文件
|
|
"""
|
|
# 检查是否有子目录
|
|
subdirs = [d for d in source_path.iterdir() if d.is_dir()]
|
|
|
|
if not subdirs:
|
|
# 没有子目录，检查是否有 txt 文件
|
|
txt_files = list(source_path.glob("*.txt"))
|
|
if txt_files:
|
|
return "flat"
|
|
return "unknown"
|
|
|
|
# 有子目录，检查第一个子目录的结构
|
|
sample_dir = subdirs[0]
|
|
files = list(sample_dir.iterdir())
|
|
txt_files = [f for f in files if f.suffix == '.txt']
|
|
|
|
if len(txt_files) == 1 and txt_files[0].name == 'content.txt':
|
|
return "content_folder"
|
|
elif len(txt_files) >= 1:
|
|
return "multi_file_folder"
|
|
|
|
return "unknown"
|
|
|
|
|
|
def parse_txt_file(txt_file: Path) -> Dict:
|
|
"""解析 txt 文件，提取元数据"""
|
|
info = {
|
|
"title": txt_file.stem.replace("-", " ").title(),
|
|
"author": "Unknown",
|
|
"year": None,
|
|
"path": str(txt_file),
|
|
"has_content": True
|
|
}
|
|
|
|
try:
|
|
with open(txt_file, 'r', encoding='utf-8') as f:
|
|
lines = f.readlines()
|
|
for line in lines[:10]:
|
|
if line.startswith("# "):
|
|
info["title"] = line[2:].strip()
|
|
elif line.startswith("Author:"):
|
|
info["author"] = line[7:].strip()
|
|
elif line.startswith("Year:"):
|
|
info["year"] = line[5:].strip()
|
|
except:
|
|
pass
|
|
|
|
return info
|
|
|
|
|
|
def get_stories_from_flat(source_path: Path) -> List[Dict]:
|
|
"""处理扁平结构: /source/poem.txt"""
|
|
stories = []
|
|
for txt_file in source_path.glob("*.txt"):
|
|
info = parse_txt_file(txt_file)
|
|
info["id"] = txt_file.stem
|
|
info["type"] = "flat"
|
|
stories.append(info)
|
|
return stories
|
|
|
|
|
|
def get_stories_from_content_folder(source_path: Path) -> List[Dict]:
|
|
"""处理 content_folder 结构: /source/story-dir/content.txt"""
|
|
stories = []
|
|
for story_dir in source_path.iterdir():
|
|
if story_dir.is_dir() and story_dir.name != '__pycache__':
|
|
content_file = story_dir / "content.txt"
|
|
if content_file.exists():
|
|
info = parse_txt_file(content_file)
|
|
info["id"] = story_dir.name
|
|
info["type"] = "content_folder"
|
|
stories.append(info)
|
|
return stories
|
|
|
|
|
|
def get_stories_from_multi_file_folder(source_path: Path) -> List[Dict]:
|
|
"""处理 multi_file_folder 结构: /source/poet-dir/poem.txt"""
|
|
stories = []
|
|
for poet_dir in source_path.iterdir():
|
|
if poet_dir.is_dir() and poet_dir.name != '__pycache__':
|
|
poet_name = poet_dir.name.replace("-", " ").title()
|
|
for txt_file in poet_dir.glob("*.txt"):
|
|
info = parse_txt_file(txt_file)
|
|
info["id"] = f"{poet_dir.name}/{txt_file.stem}"
|
|
info["title"] = info["title"]
|
|
info["author"] = poet_name if info["author"] == "Unknown" else info["author"]
|
|
info["type"] = "multi_file_folder"
|
|
stories.append(info)
|
|
return stories
|
|
|
|
|
|
def get_media_sources():
|
|
"""获取所有媒体来源（自适应结构）"""
|
|
sources = []
|
|
if MEDIA_DIR.exists():
|
|
for item in MEDIA_DIR.iterdir():
|
|
if item.is_dir():
|
|
# 加载配置或自动检测
|
|
config = load_source_config(item)
|
|
if config:
|
|
structure = config.get("structure", "auto")
|
|
display_name = config.get("name", format_source_name(item.name))
|
|
else:
|
|
structure = detect_structure_type(item)
|
|
display_name = format_source_name(item.name)
|
|
|
|
# 统计故事数量
|
|
stories = get_stories_from_source(item.name)
|
|
|
|
sources.append({
|
|
"id": item.name,
|
|
"name": display_name,
|
|
"path": str(item),
|
|
"structure": structure,
|
|
"count": len(stories)
|
|
})
|
|
return sorted(sources, key=lambda x: x["name"])
|
|
|
|
|
|
def format_source_name(name):
|
|
"""转换目录名称为显示名称"""
|
|
name_map = {
|
|
"libraryofshortstories": "Library of Short Stories",
|
|
"public-domain-poetry": "Public Domain Poetry",
|
|
}
|
|
return name_map.get(name, name.replace("-", " ").title())
|
|
|
|
|
|
def get_stories_from_source(source_id: str) -> List[Dict]:
|
|
"""从来源获取所有故事（自适应结构）"""
|
|
source_path = MEDIA_DIR / source_id
|
|
|
|
if not source_path.exists():
|
|
return []
|
|
|
|
# 获取结构类型
|
|
config = load_source_config(source_path)
|
|
if config and config.get("structure") != "auto":
|
|
structure = config["structure"]
|
|
else:
|
|
structure = detect_structure_type(source_path)
|
|
|
|
# 根据结构类型处理
|
|
handlers = {
|
|
"flat": get_stories_from_flat,
|
|
"content_folder": get_stories_from_content_folder,
|
|
"multi_file_folder": get_stories_from_multi_file_folder,
|
|
}
|
|
|
|
handler = handlers.get(structure, get_stories_from_content_folder)
|
|
stories = handler(source_path)
|
|
|
|
return sorted(stories, key=lambda x: x["title"])
|
|
|
|
|
|
def read_story_content(source_id: str, story_id: str) -> Optional[str]:
|
|
"""读取故事内容（自适应路径）"""
|
|
source_path = MEDIA_DIR / source_id
|
|
|
|
if not source_path.exists():
|
|
return None
|
|
|
|
# 尝试不同的路径格式
|
|
possible_paths = [
|
|
# content_folder: story-id/content.txt
|
|
source_path / story_id / "content.txt",
|
|
# multi_file_folder: poet-slug/poem-slug
|
|
source_path / f"{story_id}.txt",
|
|
# flat: story-id.txt
|
|
source_path / f"{story_id}.txt",
|
|
]
|
|
|
|
# 如果 story_id 包含 /，直接解析
|
|
if "/" in story_id:
|
|
parts = story_id.split("/", 1)
|
|
possible_paths.insert(0, source_path / parts[0] / f"{parts[1]}.txt")
|
|
|
|
for path in possible_paths:
|
|
if path.exists():
|
|
with open(path, 'r', encoding='utf-8') as f:
|
|
return f.read()
|
|
|
|
return None
|
|
|
|
|
|
# Routes
|
|
@app.route("/")
|
|
def index():
|
|
"""Home page - show media sources"""
|
|
sources = get_media_sources()
|
|
return render_template("index.html", sources=sources)
|
|
|
|
|
|
@app.route("/list/<source_id>")
|
|
def story_list(source_id):
|
|
"""Story list page for a source"""
|
|
source = None
|
|
for s in get_media_sources():
|
|
if s["id"] == source_id:
|
|
source = s
|
|
break
|
|
|
|
if not source:
|
|
return "Source not found", 404
|
|
|
|
stories = get_stories_from_source(source_id)
|
|
return render_template("list.html", source=source, stories=stories)
|
|
|
|
|
|
@app.route("/play/<path:story_path>")
|
|
def player(story_path):
|
|
"""Player page for a story (支持多级路径)"""
|
|
# 解析路径: source_id/story_id 或 source_id/poet/poem
|
|
parts = story_path.split("/", 1)
|
|
if len(parts) < 2:
|
|
return "Invalid story path", 400
|
|
|
|
source_id = parts[0]
|
|
story_id = parts[1]
|
|
|
|
content = read_story_content(source_id, story_id)
|
|
if not content:
|
|
return "Story not found", 404
|
|
|
|
# 解析元数据
|
|
title = story_id.replace("-", " ").replace("/", " - ").title()
|
|
author = "Unknown"
|
|
year = ""
|
|
|
|
lines = content.split('\n')
|
|
body_start = 0
|
|
for i, line in enumerate(lines[:20]):
|
|
if line.startswith("# "):
|
|
title = line[2:].strip()
|
|
elif line.startswith("Author:"):
|
|
author = line[7:].strip()
|
|
elif line.startswith("Year:"):
|
|
year = line[5:].strip()
|
|
elif line.startswith("==="):
|
|
body_start = i + 1
|
|
break
|
|
|
|
body = '\n'.join(lines[body_start:]).strip()
|
|
|
|
return render_template("player.html",
|
|
source_id=source_id,
|
|
story_id=story_id,
|
|
title=title,
|
|
author=author,
|
|
year=year,
|
|
content=body)
|
|
|
|
|
|
# API Routes
|
|
@app.route("/api/sources")
|
|
def api_sources():
|
|
"""API: Get all sources"""
|
|
return json.dumps(get_media_sources())
|
|
|
|
|
|
@app.route("/api/stories/<source_id>")
|
|
def api_stories(source_id):
|
|
"""API: Get stories from a source"""
|
|
return json.dumps(get_stories_from_source(source_id))
|
|
|
|
|
|
async def generate_audio_stream(text, voice=DEFAULT_VOICE, rate="0%"):
|
|
"""Generate audio stream using edge-tts with speed control"""
|
|
communicate = edge_tts.Communicate(text, voice, rate=rate)
|
|
async for chunk in communicate.stream():
|
|
if chunk["type"] == "audio":
|
|
yield chunk["data"]
|
|
|
|
|
|
@app.route("/api/stream")
|
|
def stream_audio():
|
|
"""Stream audio from text with speed control"""
|
|
text = request.args.get("text", "")
|
|
voice = request.args.get("voice", DEFAULT_VOICE)
|
|
# edge-tts rate format: [+-]\d+% (e.g., +20%, -20%)
|
|
# "0%" is invalid, use "-10%" as default for comfortable speed
|
|
rate = request.args.get("rate", "-10%")
|
|
if rate == "0%":
|
|
rate = "-10%"
|
|
|
|
if not text:
|
|
return "No text provided", 400
|
|
|
|
def generate():
|
|
loop = asyncio.new_event_loop()
|
|
asyncio.set_event_loop(loop)
|
|
|
|
async def stream():
|
|
async for chunk in generate_audio_stream(text, voice, rate):
|
|
yield chunk
|
|
|
|
gen = stream()
|
|
while True:
|
|
try:
|
|
chunk = loop.run_until_complete(gen.__anext__())
|
|
yield chunk
|
|
except StopAsyncIteration:
|
|
break
|
|
loop.close()
|
|
|
|
return Response(generate(), mimetype="audio/mpeg")
|
|
|
|
|
|
@app.route("/screenshots/<path:filename>")
|
|
def serve_screenshot(filename):
|
|
"""提供截图文件下载"""
|
|
return send_file(f"static/{filename}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
app.run(host="0.0.0.0", port=8899, debug=False, threaded=True)
|