Spaces:
Sleeping
Sleeping
| #!/usr/bin/env python3 | |
| """ | |
| Hugging Face Docker Space for the Ms. Nova lesson. | |
| Serves the static page (index.html + lesson.json) AND a same-origin neural | |
| text-to-speech endpoint, so the browser just plays an MP3 (reliable on mobile, | |
| no browser TTS needed) with real Microsoft Edge neural voices + word timings. | |
| Endpoints: | |
| GET /ping -> "ok" | |
| GET /tts?text=..&voice=..&rate=.. -> {"audio": <base64 mp3>, "boundaries":[...]} | |
| GET / (and any file) -> static files from this folder | |
| """ | |
| import asyncio, json, base64, urllib.parse, os, re, mimetypes | |
| from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer | |
| import edge_tts # pip install edge-tts | |
| HOST = "0.0.0.0" | |
| PORT = int(os.environ.get("PORT", "7860")) # HF Docker default port | |
| DEFAULT_VOICE = "en-GB-SoniaNeural" | |
| ROOT = os.path.dirname(os.path.abspath(__file__)) | |
| _RATE_RE = re.compile(r"^[+-]\d{1,3}%$") | |
| def _clean_rate(rate: str) -> str: | |
| rate = (rate or "").strip() | |
| return rate if _RATE_RE.match(rate) else "+0%" | |
| async def _synth_one(text, voice, rate="+0%"): | |
| communicate = edge_tts.Communicate(text, voice, rate=rate, boundary="WordBoundary") | |
| audio = bytearray() | |
| boundaries = [] | |
| async for chunk in communicate.stream(): | |
| if chunk["type"] == "audio": | |
| audio += chunk["data"] | |
| elif chunk["type"] == "WordBoundary": | |
| boundaries.append({"text": chunk["text"], "offset": chunk["offset"] / 10000.0}) | |
| if not audio: | |
| raise RuntimeError("no audio for voice " + voice) | |
| return bytes(audio), boundaries | |
| def _candidates(voice): | |
| chain = [voice] | |
| if "Libby" in voice: chain.append("en-GB-LibbyNeural") | |
| if "Sonia" in voice: chain.append("en-GB-SoniaNeural") | |
| # keep the whole fallback chain British so a failed voice never drops to a US accent | |
| chain += ["en-GB-SoniaNeural", "en-GB-LibbyNeural", "en-GB-RyanNeural"] | |
| seen, out = set(), [] | |
| for v in chain: | |
| if v not in seen: | |
| seen.add(v); out.append(v) | |
| return out | |
| async def synth(text, voice, rate="+0%"): | |
| last = None | |
| for v in _candidates(voice): | |
| try: | |
| return await _synth_one(text, v, rate) | |
| except Exception as e: | |
| last = e | |
| raise last if last else RuntimeError("synthesis failed") | |
| class Handler(BaseHTTPRequestHandler): | |
| def _cors(self): | |
| self.send_header("Access-Control-Allow-Origin", "*") | |
| def log_message(self, *args): | |
| pass | |
| def do_OPTIONS(self): | |
| self.send_response(204); self._cors(); self.end_headers() | |
| def _serve_static(self, rel): | |
| path = os.path.normpath(os.path.join(ROOT, rel)) | |
| if not path.startswith(ROOT) or not os.path.isfile(path): | |
| self.send_response(404); self._cors(); self.end_headers(); return | |
| ctype = mimetypes.guess_type(path)[0] or "application/octet-stream" | |
| with open(path, "rb") as f: | |
| data = f.read() | |
| self.send_response(200); self._cors() | |
| self.send_header("Content-Type", ctype) | |
| self.send_header("Content-Length", str(len(data))) | |
| self.end_headers(); self.wfile.write(data) | |
| def do_GET(self): | |
| parsed = urllib.parse.urlparse(self.path) | |
| params = urllib.parse.parse_qs(parsed.query) | |
| path = parsed.path | |
| if path == "/ping": | |
| self.send_response(200); self._cors() | |
| self.send_header("Content-Type", "text/plain"); self.end_headers() | |
| self.wfile.write(b"ok"); return | |
| if path == "/tts": | |
| text = (params.get("text", [""])[0]).strip() | |
| voice = params.get("voice", [DEFAULT_VOICE])[0] or DEFAULT_VOICE | |
| rate = _clean_rate(params.get("rate", ["+0%"])[0]) | |
| if not text: | |
| self.send_response(400); self._cors(); self.end_headers(); return | |
| try: | |
| audio, boundaries = asyncio.run(synth(text, voice, rate)) | |
| body = json.dumps({ | |
| "audio": base64.b64encode(audio).decode("ascii"), | |
| "boundaries": boundaries, | |
| }).encode("utf-8") | |
| self.send_response(200); self._cors() | |
| self.send_header("Content-Type", "application/json") | |
| self.send_header("Content-Length", str(len(body))) | |
| self.end_headers(); self.wfile.write(body) | |
| except Exception as e: | |
| self.send_response(500); self._cors(); self.end_headers() | |
| self.wfile.write(str(e).encode("utf-8")) | |
| return | |
| # static files | |
| rel = "index.html" if path in ("/", "") else path.lstrip("/") | |
| self._serve_static(rel) | |
| if __name__ == "__main__": | |
| print(f"[lesson] serving on http://{HOST}:{PORT}") | |
| ThreadingHTTPServer((HOST, PORT), Handler).serve_forever() | |