selfreleaseandDevin
490f032d60
- 连点页脚提示 6 次唤出管理面板：本机偏好/播放状态 + 服务器统计 - server.py 新增 GET /api/stats：运行时长、访客 IP（X-Forwarded-For 末位取真实 IP 防伪造）、TTS 请求与缓存命中率、缓存体积、经典数 - 计数为进程内存统计，重启清零 Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
203 lines
8.2 KiB
Python
203 lines
8.2 KiB
Python
#!/usr/bin/env python3
|
||
"""心经诵读站点后端：静态文件托管 + 千问 TTS 代理。"""
|
||
|
||
import hashlib
|
||
import json
|
||
import os
|
||
import time
|
||
import urllib.error
|
||
import urllib.request
|
||
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
|
||
from pathlib import Path
|
||
|
||
BASE_DIR = Path(__file__).resolve().parent
|
||
CACHE_DIR = BASE_DIR / "cache"
|
||
SUTRA_DIR = BASE_DIR / "sutras"
|
||
ENV_PATH = BASE_DIR / ".env"
|
||
|
||
|
||
def load_env(path: Path) -> None:
|
||
"""读取 .env 文件并注入环境变量（不覆盖已有变量）。"""
|
||
if not path.exists():
|
||
return
|
||
for raw in path.read_text(encoding="utf-8").splitlines():
|
||
line = raw.strip()
|
||
if not line or line.startswith("#") or "=" not in line:
|
||
continue
|
||
key, _, value = line.partition("=")
|
||
os.environ.setdefault(key.strip(), value.strip())
|
||
|
||
|
||
load_env(ENV_PATH)
|
||
|
||
API_KEY = os.environ.get("DASHSCOPE_API_KEY", "")
|
||
DASHSCOPE_BASE = os.environ.get(
|
||
"DASHSCOPE_BASE",
|
||
"https://llm-znfsxsdp6uwik7ad.cn-beijing.maas.aliyuncs.com/api/v1",
|
||
).rstrip("/")
|
||
TTS_MODEL = os.environ.get("TTS_MODEL", "qwen3-tts-flash")
|
||
PORT = int(os.environ.get("PORT", "8000"))
|
||
|
||
GENERATION_URL = f"{DASHSCOPE_BASE}/services/aigc/multimodal-generation/generation"
|
||
|
||
# 运行期统计：进程内存内计数，重启清零
|
||
STATS = {"start": time.time(), "tts_requests": 0, "cache_hits": 0}
|
||
VISITORS = set() # 见过的客户端 IP（本次进程生命周期内）
|
||
|
||
|
||
def http_post_json(url: str, payload: dict, headers: dict, timeout: int = 60) -> dict:
|
||
"""发送 JSON POST 请求并返回解析后的响应。"""
|
||
req = urllib.request.Request(
|
||
url,
|
||
data=json.dumps(payload).encode("utf-8"),
|
||
headers={"Content-Type": "application/json", **headers},
|
||
method="POST",
|
||
)
|
||
with urllib.request.urlopen(req, timeout=timeout) as resp:
|
||
return json.loads(resp.read().decode("utf-8"))
|
||
|
||
|
||
def synthesize(text: str, voice: str) -> bytes:
|
||
"""调用千问 TTS 合成语音，返回 wav 音频字节流。"""
|
||
payload = {
|
||
"model": TTS_MODEL,
|
||
"input": {"text": text, "voice": voice},
|
||
}
|
||
data = http_post_json(GENERATION_URL, payload, {"Authorization": f"Bearer {API_KEY}"})
|
||
audio_url = (data.get("output") or {}).get("audio", {}).get("url")
|
||
if not audio_url:
|
||
raise RuntimeError(f"TTS 响应缺少音频地址：{json.dumps(data, ensure_ascii=False)[:300]}")
|
||
with urllib.request.urlopen(audio_url, timeout=60) as resp:
|
||
return resp.read()
|
||
|
||
|
||
class Handler(SimpleHTTPRequestHandler):
|
||
"""静态资源 + /api/tts 合成接口。"""
|
||
|
||
def __init__(self, *args, **kwargs):
|
||
super().__init__(*args, directory=str(BASE_DIR), **kwargs)
|
||
|
||
def _track_visitor(self) -> None:
|
||
"""记录客户端 IP 到访客集合；nginx 反代后真实来源在 X-Forwarded-For / X-Real-IP。"""
|
||
# nginx 把真实 IP 追加在 X-Forwarded-For 末尾，取最后一个元素防伪造
|
||
xff = (self.headers.get("X-Forwarded-For") or "").split(",")[-1].strip()
|
||
ip = xff or (self.headers.get("X-Real-IP") or "").strip() or self.client_address[0]
|
||
VISITORS.add(ip)
|
||
|
||
def do_GET(self) -> None:
|
||
"""处理 GET 请求：/api/* 接口，其余按静态文件处理。"""
|
||
self._track_visitor()
|
||
if self.path == "/api/stats":
|
||
cache_files = list(CACHE_DIR.glob("*.wav")) if CACHE_DIR.is_dir() else []
|
||
self._send_json({
|
||
"uptime_sec": int(time.time() - STATS["start"]),
|
||
"tts_requests": STATS["tts_requests"],
|
||
"cache_hits": STATS["cache_hits"],
|
||
"cache_files": len(cache_files),
|
||
"cache_bytes": sum(f.stat().st_size for f in cache_files),
|
||
"sutra_count": len(list(SUTRA_DIR.glob("*.json"))) if SUTRA_DIR.is_dir() else 0,
|
||
"visitors": len(VISITORS),
|
||
"model": TTS_MODEL,
|
||
})
|
||
return
|
||
if self.path == "/api/sutras":
|
||
try:
|
||
items = []
|
||
if SUTRA_DIR.is_dir():
|
||
for f in sorted(SUTRA_DIR.glob("*.json")):
|
||
d = json.loads(f.read_text(encoding="utf-8"))
|
||
items.append({
|
||
"id": d.get("id") or f.stem,
|
||
"name": d.get("name") or f.stem,
|
||
"cat": d.get("cat") or "其他",
|
||
"order": d.get("order", 99),
|
||
})
|
||
items.sort(key=lambda x: (x["order"], x["id"]))
|
||
self._send_json({"sutras": items})
|
||
except Exception as exc: # noqa: BLE001
|
||
self._send_json({"error": f"读取经典目录失败：{exc}"}, status=500)
|
||
return
|
||
super().do_GET()
|
||
|
||
def do_POST(self) -> None:
|
||
"""处理 POST 请求，目前仅支持 /api/tts。"""
|
||
self._track_visitor()
|
||
if self.path != "/api/tts":
|
||
self._send_json({"error": "not found"}, status=404)
|
||
return
|
||
try:
|
||
length = int(self.headers.get("Content-Length", "0"))
|
||
body = json.loads(self.rfile.read(length).decode("utf-8"))
|
||
except (ValueError, json.JSONDecodeError):
|
||
self._send_json({"error": "请求体不是合法 JSON"}, status=400)
|
||
return
|
||
|
||
text = (body.get("text") or "").strip()
|
||
voice = (body.get("voice") or "").strip()
|
||
if not text or not voice:
|
||
self._send_json({"error": "缺少 text 或 voice 参数"}, status=400)
|
||
return
|
||
if len(text) > 800:
|
||
self._send_json({"error": "文本超过 800 字上限"}, status=400)
|
||
return
|
||
if not API_KEY:
|
||
self._send_json({"error": "服务端未配置 DASHSCOPE_API_KEY"}, status=500)
|
||
return
|
||
|
||
cache_key = hashlib.sha256(f"{TTS_MODEL}|{voice}|{text}".encode("utf-8")).hexdigest()[:24]
|
||
cache_file = CACHE_DIR / f"{cache_key}.wav"
|
||
STATS["tts_requests"] += 1
|
||
try:
|
||
if cache_file.exists():
|
||
STATS["cache_hits"] += 1
|
||
audio = cache_file.read_bytes()
|
||
else:
|
||
audio = synthesize(text, voice)
|
||
CACHE_DIR.mkdir(exist_ok=True)
|
||
# 先写临时文件再原子重命名，避免中途崩溃留下半截 WAV 毒化缓存
|
||
tmp_file = cache_file.with_suffix(".tmp")
|
||
tmp_file.write_bytes(audio)
|
||
tmp_file.replace(cache_file)
|
||
except urllib.error.HTTPError as exc:
|
||
detail = exc.read().decode("utf-8", errors="replace")[:300]
|
||
self._send_json({"error": f"TTS 服务返回 {exc.code}：{detail}"}, status=502)
|
||
return
|
||
except Exception as exc: # noqa: BLE001 - 统一兜底，向前端返回可读错误
|
||
self._send_json({"error": f"语音合成失败：{exc}"}, status=502)
|
||
return
|
||
|
||
self.send_response(200)
|
||
self.send_header("Content-Type", "audio/wav")
|
||
self.send_header("Content-Length", str(len(audio)))
|
||
self.end_headers()
|
||
self.wfile.write(audio)
|
||
|
||
def _send_json(self, obj: dict, status: int = 200) -> None:
|
||
"""以 JSON 形式返回响应。"""
|
||
data = json.dumps(obj, ensure_ascii=False).encode("utf-8")
|
||
self.send_response(status)
|
||
self.send_header("Content-Type", "application/json; charset=utf-8")
|
||
self.send_header("Content-Length", str(len(data)))
|
||
self.end_headers()
|
||
self.wfile.write(data)
|
||
|
||
def end_headers(self) -> None:
|
||
"""本地开发环境禁用静态资源缓存，避免页面拿到旧版 JS/CSS。"""
|
||
if not self.path.startswith("/api/"):
|
||
self.send_header("Cache-Control", "no-cache")
|
||
super().end_headers()
|
||
|
||
def log_message(self, fmt, *args) -> None:
|
||
"""精简请求日志输出。"""
|
||
if "/api/" in self.path:
|
||
super().log_message(fmt, *args)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
server = ThreadingHTTPServer(("127.0.0.1", PORT), Handler)
|
||
print(f"心经诵读站点已启动：http://127.0.0.1:{PORT}")
|
||
print(f"TTS 模型：{TTS_MODEL} | 上游：{DASHSCOPE_BASE}")
|
||
if not API_KEY:
|
||
print("警告：未配置 DASHSCOPE_API_KEY，请复制 .env.example 为 .env 并填入密钥")
|
||
server.serve_forever()
|