import asyncio
import hmac
import os
import re
import threading
import time
from pathlib import Path
from typing import List, Optional
from fastapi import FastAPI, Header, HTTPException
from fastapi.responses import FileResponse, HTMLResponse, JSONResponse
from pydantic import BaseModel
import builder
import store
app = FastAPI(title="Lesson content server")
state = {"status": "idle", "log": [], "errors": {}, "finished": None, "voices": {}, "fallback": {}}
lock = threading.Lock()
def _log(msg):
state["log"].append(msg)
del state["log"][:-200]
def run_build():
if not lock.acquire(blocking=False):
return
try:
state.update(status="building", log=[], errors={})
# صداهای قبلی را از ذخیره‌ی پایدار برمی‌گردانیم. اگر نشد، ساخت را متوقف می‌کنیم تا چیزی از نو ساخته و بازنویسی نشود.
if not store.pull(_log):
_log("ساخت متوقف شد تا صداهای قبلی بازنویسی نشوند. کمی بعد «ساخت دوباره» را بزنید.")
state["status"] = "failed"
return
def sync():
if builder.STATE["dirty"] and store.push(_log):
builder.STATE["dirty"] = False
_, errors = asyncio.run(builder.build_all(_log, sync))
sync()
fb = {k: sum(1 for i in v["items"] if i["fb"]) for k, v in builder.LESSON_AUDIO.items()}
state.update(status="ready", errors=errors, voices=dict(builder.INFO), fallback={k: n for k, n in fb.items() if n},
finished=time.strftime("%Y-%m-%d %H:%M:%S"))
if state["fallback"]:
_log("⚠️ جمله‌هایی با صدای جایگزین ساخته شده‌اند: " + ", ".join(f"{k}({n})" for k, n in state["fallback"].items())
+ " . در صفحه‌ی «بررسی صداها» می‌توانید دوباره با Kokoro بسازید.")
except Exception as e: # noqa: BLE001
_log(f"خطای کلی: {e}")
state["status"] = "failed"
finally:
lock.release()
@app.on_event("startup")
def startup():
threading.Thread(target=run_build, daemon=True).start()
def check_password(p):
expected = os.getenv("REBUILD_PASSWORD", "")
if expected and not hmac.compare_digest(p or "", expected):
raise HTTPException(401, "رمز اشتباه است")
@app.get("/manifest.json")
def manifest():
f = builder.OUT / "manifest.json"
if not f.exists():
return JSONResponse({"status": state["status"]}, status_code=503)
return FileResponse(f, media_type="application/json", headers={"Cache-Control": "no-cache"})
STRINGS_FILE = Path(__file__).parent / "strings.json"
@app.get("/strings.json")
def strings():
"""متن‌های برنامه‌ی اندروید: { "متن فعلی": "متن جدید" }. اپ هنگام باز شدن آن را می‌گیرد."""
if not STRINGS_FILE.exists():
return JSONResponse({}, headers={"Cache-Control": "no-cache"})
return FileResponse(STRINGS_FILE, media_type="application/json", headers={"Cache-Control": "no-cache"})
@app.get("/packs/{name}")
def pack(name: str):
if not re.fullmatch(r"[A-Za-z0-9_.-]+\.zip", name):
raise HTTPException(400, "bad name")
f = builder.OUT / "packs" / name
if not f.exists():
raise HTTPException(404, "not found")
return FileResponse(f, media_type="application/zip")
@app.get("/status")
def status():
return state
@app.post("/rebuild")
def rebuild(x_password: str = Header(default="")):
check_password(x_password)
if state["status"] == "building":
return {"ok": False, "msg": "ساخت در حال انجام است"}
threading.Thread(target=run_build, daemon=True).start()
return {"ok": True}
class RegenBody(BaseModel):
lesson: str
texts: Optional[List[str]] = None # خالی = همه‌ی متن‌های درس
@app.post("/api/regen")
def api_regen(body: RegenBody, x_password: str = Header(default="")):
"""صدای کل یک درس یا فقط چند جمله/واژه را دوباره می‌سازد. بقیه‌ی درس‌ها و صداها دست نمی‌خورند."""
check_password(x_password)
if state["status"] == "building":
raise HTTPException(409, "ساخت در حال انجام است؛ بعد از پایانش دوباره تلاش کنید")
try:
n = builder.regen(body.lesson, body.texts)
except Exception as e: # noqa: BLE001
raise HTTPException(400, str(e))
threading.Thread(target=run_build, daemon=True).start()
return {"ok": True, "count": n}
@app.get("/api/lessons")
def api_lessons():
return builder.LESSON_AUDIO
@app.get("/audio/{name}")
def audio(name: str):
if not re.fullmatch(r"[A-Za-z0-9_-]+\.(mp3|wav|ogg|m4a|flac)", name):
raise HTTPException(400, "bad name")
f = builder.CACHE / name
if not f.exists():
raise HTTPException(404, "not found")
return FileResponse(f)
REVIEW = r"""
بررسی صداها
هر متن را پخش کنید. اگر تلفظ بد بود «بازسازی» را بزنید. فقط همان صدا ساخته می‌شود و بقیه‌ی درس‌ها دست نمی‌خورند.
اگر بعد از بازسازی هم همان‌طور خوانده شد، در فایل درس کنار آن متن فیلد "tts" بگذارید (مثلاً املای دیگر یا علامت‌گذاری دیگر).
بعد از تغییر صدا، version همان درس را بالا ببرید تا کاربران قدیمی هم نسخه‌ی تازه را بگیرند.