fj/server.py

#!/usr/bin/env python3
"""心经诵读站点后端：静态文件托管 + 千问 TTS 代理。"""

import hashlib
import json
import os
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"
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"


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 do_POST(self) -> None:
        """处理 POST 请求，目前仅支持 /api/tts。"""
        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) > 600:
            self._send_json({"error": "文本超过 600 字上限"}, 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"
        try:
            if cache_file.exists():
                audio = cache_file.read_bytes()
            else:
                audio = synthesize(text, voice)
                CACHE_DIR.mkdir(exist_ok=True)
                cache_file.write_bytes(audio)
        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()