Spaces:
Running
Running
| """ | |
| ساخت بسته‌های درس (متن + صدا + تصویر) و فایل فهرست (manifest.json). | |
| ورودی : curriculum.json + lessons_src/*.json (+ اختیاری: lessons_src/<id>/img/*.jpg|png|webp) | |
| خروجی : OUT_DIR/manifest.json و OUT_DIR/packs/<id>-v<version>.zip | |
| صداها: | |
| • واژه‌ها (آیتم‌های بخش واژگان و هر متن تک‌کلمه‌ای): | |
| edge-tts با صدای en-US-GuyNeural (مرد آمریکایی) و سرعت -15٪ | |
| (همان تنظیمی که در «آلفا مترجم» انتخاب کرده‌اید) | |
| • مکالمه‌ی درس‌ها: جمله‌های گوینده‌ی A با صدای مرد (Kokoro، Adam) و همان سرعت صدای زن؛ کلید صدایشان "m|<متن>" است | |
| • جمله‌ها: Space صدای Kokoro (صدای پیش‌فرض Heart ، سرعت 0.80) | |
| اگر Kokoro فایل WAV بدهد، با ffmpeg به MP3 کم‌حجم (تک‌کاناله، 24kHz، ~48kbps) تبدیل و همان ذخیره می‌شود. | |
| اگر آن Space جواب ندهد، با edge-tts (Jenny ، سرعت -25٪) جایگزین می‌شود و در گزارش نوشته می‌شود. | |
| اجرای مستقل: python builder.py (FAKE_TTS=1 برای تست بدون اینترنت) | |
| """ | |
| import asyncio | |
| import hashlib | |
| import io | |
| import json | |
| import os | |
| import re | |
| import shutil | |
| import time | |
| import wave | |
| import zipfile | |
| from pathlib import Path | |
| import audio_mp3 | |
| import grammar | |
| import kokoro_client | |
| ROOT = Path(__file__).parent | |
| SRC = ROOT / "lessons_src" | |
| OUT = Path(os.getenv("OUT_DIR", "/tmp/content_out")) | |
| STORE_DIR = Path(os.getenv("STORE_DIR", "/tmp/store")) # با store.py در Dataset ذخیره می‌شود | |
| CACHE = Path(os.getenv("CACHE_DIR", str(STORE_DIR / "audio"))) # صدای هر متن (یک بار ساخته و برای همیشه نگه داشته می‌شود) | |
| TAKES_FILE = STORE_DIR / "takes.json" | |
| TAKES = {} # متن ← شماره‌ی «بازسازی»؛ بالا رفتنش یعنی صدای تازه برای آن متن | |
| REVS_FILE = STORE_DIR / "revs.json" | |
| REVS = {} # شناسه‌ی درس ← عددی که خودکار به version اضافه می‌شود (مثلاً وقتی صدای گرامر عوض شد) | |
| METAS = {} # شناسه‌ی درس ← اطلاعات آخرین بسته‌ی ساخته‌شده (برای manifest و صفحه‌ی اصلی) | |
| LESSON_AUDIO = {} # برای صفحه‌ی بررسی صداها | |
| FALLBACK_TEXTS = set() # جمله‌هایی که با صدای جایگزین ساخته شده‌اند | |
| STATE = {"dirty": False} # آیا صدای تازه‌ای هست که هنوز در ذخیره‌ی پایدار نرفته؟ | |
| WORD_VOICE = os.getenv("WORD_VOICE", "en-US-GuyNeural") | |
| WORD_RATE = os.getenv("WORD_RATE", "-15%") | |
| FALLBACK_VOICE = os.getenv("FALLBACK_SENT_VOICE", "en-US-JennyNeural") | |
| FALLBACK_RATE = os.getenv("FALLBACK_SENT_RATE", "-25%") | |
| KOKORO_BASE = os.getenv("KOKORO_BASE", "https://opera10-translate-tts-aloha.hf.space") | |
| KOKORO_SPEED = float(os.getenv("KOKORO_SPEED", "0.80")) | |
| KOKORO_VOICE = os.getenv("KOKORO_VOICE", "") # خالی = صدای پیش‌فرض خود Space (Heart) | |
| KOKORO_URL = os.getenv("KOKORO_URL", "") # اگر آدرس دقیق API را می‌دانید | |
| KOKORO_BODY = os.getenv("KOKORO_BODY", "") # قالب JSON دستی (اختیاری) | |
| CONCURRENCY = int(os.getenv("TTS_CONCURRENCY", "4")) | |
| FAKE = os.getenv("FAKE_TTS") == "1" | |
| # صدای مرد (Adam) برای گوینده‌ی A در بخش «مکالمه»ی درس‌ها (نه پادکست، نه آزمون، نه جای دیگر). | |
| # سرعتش همان KOKORO_SPEED صدای زن است. CONV_A_VOICE خالی = خاموش. اگر Space نام صدا را جور دیگری می‌خواهد، | |
| # KOKORO_VOICE_MALE را بگذارید (مثلاً Adam). | |
| CONV_A_VOICE = os.getenv("CONV_A_VOICE", "male") | |
| # صدای مرد برای خط‌های گوینده‌ای که اسمش در این فهرست باشد (نه «گوینده‌ی اول»). Ali برای درس‌های قدیمی است. | |
| MALE_SPEAKERS = {x.strip().lower() for x in os.getenv("CONV_MALE_SPEAKERS", "John,Ali").split(",") if x.strip()} | |
| MALE_VOICE = os.getenv("KOKORO_VOICE_MALE", "") | |
| MALE_VOICE_TRIES = [MALE_VOICE] if MALE_VOICE else ["am_adam", "Adam", "adam"] | |
| MALE_FALLBACK_VOICE = os.getenv("FALLBACK_MALE_VOICE", "en-US-GuyNeural") | |
| MALE = "m|" # پیشوند کلید صدا در lesson.audio برای جمله‌هایی که با صدای مرد ساخته می‌شوند | |
| # عددی که به version همه‌ی درس‌ها اضافه می‌شود. وقتی صداها عوض شدند (مثلاً سرعت یا فرمت) این را یکی بالا ببرید | |
| # تا اپ کاربران قدیمی همه‌ی درس‌ها را دوباره بگیرد. پیش‌فرض 0 یعنی هیچ تغییری در version نمی‌دهد. | |
| AUDIO_REV = int(os.getenv("AUDIO_REV", "0") or 0) | |
| MP3_STATE = {"warned": False, "ok": 0, "kept_wav": 0} # آمار تبدیل WAV به MP3 در همین بار ساخت | |
| # اطلاعاتی که صفحه‌ی وضعیت نشان می‌دهد | |
| INFO = {"words": f"{WORD_VOICE} {WORD_RATE}", "sentences": "—", "fallback_used": False, "kokoro_error": "", | |
| "male": "—", "male_fallback": False} | |
| _male_good = {"v": None} | |
| def collect(node, acc): | |
| """همه‌ی متن‌هایی که باید صدا بگیرند را به ترتیب پیدا می‌کند.""" | |
| if isinstance(node, dict): | |
| en = node.get("en") | |
| if isinstance(en, str) and en.strip(): | |
| acc.setdefault((node.get("tts") or en).strip(), None) | |
| for v in node.values(): | |
| collect(v, acc) | |
| elif isinstance(node, list): | |
| for v in node: | |
| collect(v, acc) | |
| def tag_male_lines(lesson): | |
| """ | |
| در هر بخش «مکالمه»ی یک درس معمولی (نه پادکست)، خط‌های گوینده‌ای که اسمش در MALE_SPEAKERS است (John) کلید صدای مرد | |
| می‌گیرند: \"ak\": \"m|<متن>\". بقیه (Sara) با صدای زن مثل قبل. مهم نیست چه کسی مکالمه را شروع کند. | |
| کلیدهای مرد را برمی‌گرداند. | |
| """ | |
| out = set() | |
| if not CONV_A_VOICE or lesson.get("kind") == "podcast": | |
| return out | |
| for s in lesson.get("sections", []): | |
| if s.get("type") != "conversation": | |
| continue | |
| for ln in s.get("lines") or []: | |
| en = (ln.get("tts") or ln.get("en") or "").strip() | |
| if en and (ln.get("speaker") or "").strip().lower() in MALE_SPEAKERS: | |
| ln["ak"] = MALE + en | |
| out.add(MALE + en) | |
| return out | |
| def all_texts(lesson): | |
| """همه‌ی کلیدهای صدای یک درس (متن‌ها + کلیدهای مرد). lesson تغییر می‌کند (خط‌های مرد علامت می‌خورند).""" | |
| texts = {} | |
| collect(lesson, texts) | |
| for k in tag_male_lines(lesson): | |
| texts.setdefault(k, None) | |
| return texts | |
| def collect_imgs(node, acc): | |
| if isinstance(node, dict): | |
| v = node.get("img") | |
| if isinstance(v, str) and v.strip(): | |
| acc.setdefault(v.strip(), None) | |
| for x in node.values(): | |
| collect_imgs(x, acc) | |
| elif isinstance(node, list): | |
| for x in node: | |
| collect_imgs(x, acc) | |
| def vocab_texts(lesson): | |
| out = set() | |
| for s in lesson.get("sections", []): | |
| if s.get("type") == "vocab": | |
| for it in s.get("items", []): | |
| out.add((it.get("tts") or it.get("en") or "").strip()) | |
| return out | |
| def find_cached(base): | |
| found = [p for p in CACHE.glob(base + ".*") if p.stat().st_size > 500] # فایل ناقص را قبول نکن | |
| if not found: | |
| return None | |
| # اگر از یک متن هم mp3 و هم wav مانده بود، همیشه mp3 (کم‌حجم‌تر) برداشته می‌شود | |
| found.sort(key=lambda p: (p.suffix != ".mp3", p.name)) | |
| return found[0] | |
| def load_takes(): | |
| TAKES.clear() | |
| try: | |
| TAKES.update(json.loads(TAKES_FILE.read_text(encoding="utf-8"))) | |
| except Exception: # noqa: BLE001 | |
| pass | |
| def save_takes(): | |
| TAKES_FILE.parent.mkdir(parents=True, exist_ok=True) | |
| TAKES_FILE.write_text(json.dumps(TAKES, ensure_ascii=False, indent=1), encoding="utf-8") | |
| STATE["dirty"] = True | |
| def load_revs(): | |
| REVS.clear() | |
| try: | |
| REVS.update({k: int(v) for k, v in json.loads(REVS_FILE.read_text(encoding="utf-8")).items()}) | |
| except Exception: # noqa: BLE001 | |
| pass | |
| def bump_rev(lesson_id): | |
| """version همین درس یکی بالا می‌رود تا اپ کاربران نسخه‌ی تازه را بگیرد (بدون دست زدن به فایل درس).""" | |
| load_revs() | |
| REVS[lesson_id] = REVS.get(lesson_id, 0) + 1 | |
| REVS_FILE.parent.mkdir(parents=True, exist_ok=True) | |
| REVS_FILE.write_text(json.dumps(REVS, ensure_ascii=False, indent=1), encoding="utf-8") | |
| STATE["dirty"] = True | |
| def take_suffix(text): | |
| n = TAKES.get(text, 0) | |
| return f"|t{n}" if n else "" | |
| def regen(lesson_id, texts=None): | |
| """صدای کل یک درس (یا فقط متن‌های داده‌شده) را برای ساخت بعدی «تازه» علامت می‌زند. تعداد متن‌ها را برمی‌گرداند.""" | |
| if not re.fullmatch(r"[A-Za-z0-9_-]+", lesson_id or ""): | |
| raise ValueError("شناسه‌ی درس نامعتبر است") | |
| src = SRC / f"{lesson_id}.json" | |
| if not src.exists(): | |
| raise FileNotFoundError(f"فایل {src.name} پیدا نشد") | |
| found = all_texts(json.loads(src.read_text(encoding="utf-8"))) | |
| wanted = [t for t in texts if t in found] if texts else list(found) | |
| if not wanted: | |
| raise ValueError("این متن‌ها در این درس پیدا نشدند") | |
| load_takes() | |
| for t in wanted: | |
| TAKES[t] = TAKES.get(t, 0) + 1 | |
| save_takes() | |
| return len(wanted) | |
| _kokoro = None | |
| def _get_kokoro(): | |
| global _kokoro | |
| if _kokoro is None: | |
| _kokoro = kokoro_client.KokoroSpace(KOKORO_BASE, KOKORO_SPEED, KOKORO_VOICE, KOKORO_URL, KOKORO_BODY) | |
| return _kokoro | |
| async def edge_save(text, voice, rate, path): | |
| import edge_tts # بارگذاری دیرهنگام تا تست بدون این کتابخانه هم کار کند | |
| last = None | |
| for attempt in range(4): | |
| try: | |
| await edge_tts.Communicate(text, voice, rate=rate).save(str(path)) | |
| if path.exists() and path.stat().st_size > 500: | |
| return | |
| except Exception as e: # noqa: BLE001 | |
| last = e | |
| path.unlink(missing_ok=True) | |
| await asyncio.sleep(1.5 * (attempt + 1)) | |
| raise RuntimeError(f"ساخت صدا ناموفق بود برای {text!r}: {last}") | |
| def fake_wav(): | |
| """برای تست بدون اینترنت: یک WAV واقعی و کوتاه (یک ثانیه سکوت) می‌سازد تا مسیر تبدیل به MP3 هم آزمایش شود.""" | |
| buf = io.BytesIO() | |
| with wave.open(buf, "wb") as w: | |
| w.setnchannels(1) | |
| w.setsampwidth(2) | |
| w.setframerate(24000) | |
| w.writeframes(b"\x00\x00" * 24000) | |
| return buf.getvalue() | |
| def shrink(data, ext, log): | |
| """اگر صدا WAV (یا FLAC) بود به MP3 کم‌حجم تبدیل می‌کند. اگر نشد، همان صدای اصلی را برمی‌گرداند (ساخت متوقف نمی‌شود).""" | |
| if ext not in audio_mp3.LOSSLESS: | |
| return data, ext | |
| try: | |
| out = audio_mp3.bytes_to_mp3(data, ext) | |
| MP3_STATE["ok"] += 1 | |
| return out, "mp3" | |
| except Exception as e: # noqa: BLE001 | |
| MP3_STATE["kept_wav"] += 1 | |
| if not MP3_STATE["warned"]: | |
| MP3_STATE["warned"] = True | |
| log(f"⚠️ تبدیل صدا به MP3 نشد ({str(e)[:160]}). همان {ext.upper()} نگه داشته می‌شود.") | |
| return data, ext | |
| async def make_male_audio(key, sfx, log): | |
| """صدای مرد (Adam) از همان Space Kokoro و با همان سرعت صدای زن. اگر نشد، با صدای مرد edge-tts جایگزین می‌شود.""" | |
| text = key[len(MALE):] | |
| kbase = hashlib.sha1(f"m|kokoro|{MALE_VOICE or 'adam'}|{KOKORO_SPEED}|{text}{sfx}".encode()).hexdigest()[:12] | |
| hit = find_cached(kbase) | |
| if hit: | |
| FALLBACK_TEXTS.discard(key) | |
| return hit.name | |
| fbase = hashlib.sha1(f"m|edge|{MALE_FALLBACK_VOICE}|{FALLBACK_RATE}|{text}{sfx}".encode()).hexdigest()[:12] | |
| fhit = find_cached(fbase) | |
| if fhit: | |
| FALLBACK_TEXTS.add(key) | |
| return fhit.name | |
| if FAKE: | |
| data, ext = await asyncio.to_thread(shrink, fake_wav(), "wav", log) | |
| path = CACHE / f"{kbase}.{ext}" | |
| path.write_bytes(data) | |
| INFO["male"] = "Kokoro (آزمایشی)" | |
| STATE["dirty"] = True | |
| return path.name | |
| if not INFO["male_fallback"]: | |
| tries = ([_male_good["v"]] if _male_good["v"] else []) + [v for v in MALE_VOICE_TRIES if v != _male_good["v"]] | |
| last = None | |
| for v in tries: | |
| try: | |
| data, ext = await asyncio.to_thread(_get_kokoro().synth, text, v) | |
| data, ext = await asyncio.to_thread(shrink, data, ext, log) | |
| path = CACHE / f"{kbase}.{ext}" | |
| path.write_bytes(data) | |
| _male_good["v"] = v | |
| STATE["dirty"] = True | |
| INFO["male"] = f"Kokoro ({v}) سرعت {KOKORO_SPEED}" | |
| FALLBACK_TEXTS.discard(key) | |
| return path.name | |
| except Exception as e: # noqa: BLE001 | |
| last = e | |
| INFO["male_fallback"] = True | |
| log(f"⚠️ صدای مرد (Adam) از Kokoro گرفته نشد ({str(last)[:160]}). با صدای جایگزین ادامه می‌دهیم؛ " | |
| "اگر نام صدا فرق دارد، متغیر KOKORO_VOICE_MALE را بگذارید.") | |
| path = CACHE / (fbase + ".mp3") | |
| await edge_save(text, MALE_FALLBACK_VOICE, FALLBACK_RATE, path) | |
| STATE["dirty"] = True | |
| FALLBACK_TEXTS.add(key) | |
| INFO["male"] = f"جایگزین: {MALE_FALLBACK_VOICE} {FALLBACK_RATE}" | |
| return path.name | |
| async def make_audio(text, kind, log): | |
| """فایل صوتی یک متن را می‌سازد؛ اگر قبلاً ساخته شده باشد (در ذخیره‌ی پایدار) همان را برمی‌دارد و دیگر نمی‌سازد.""" | |
| sfx = take_suffix(text) | |
| if kind == "w": | |
| base = hashlib.sha1(f"w|{WORD_VOICE}|{WORD_RATE}|{text}{sfx}".encode()).hexdigest()[:12] | |
| hit = find_cached(base) | |
| if hit: | |
| return hit.name | |
| path = CACHE / (base + ".mp3") | |
| if FAKE: | |
| path.write_bytes(b"ID3" + b"\x00" * 700 + text.encode()) | |
| else: | |
| await edge_save(text, WORD_VOICE, WORD_RATE, path) | |
| STATE["dirty"] = True | |
| return path.name | |
| if kind == "m": | |
| return await make_male_audio(text, sfx, log) | |
| # جمله: اگر صدای Kokoro یا صدای جایگزین قبلاً ساخته شده، همان می‌ماند (تثبیت) | |
| kbase = hashlib.sha1(f"s|kokoro|{KOKORO_VOICE}|{KOKORO_SPEED}|{text}{sfx}".encode()).hexdigest()[:12] | |
| hit = find_cached(kbase) | |
| if hit: | |
| FALLBACK_TEXTS.discard(text) | |
| return hit.name | |
| fbase = hashlib.sha1(f"s|edge|{FALLBACK_VOICE}|{FALLBACK_RATE}|{text}{sfx}".encode()).hexdigest()[:12] | |
| fhit = find_cached(fbase) | |
| if fhit: | |
| FALLBACK_TEXTS.add(text) | |
| return fhit.name | |
| if FAKE: | |
| data, ext = await asyncio.to_thread(shrink, fake_wav(), "wav", log) | |
| path = CACHE / f"{kbase}.{ext}" | |
| path.write_bytes(data) | |
| INFO["sentences"] = "Kokoro (آزمایشی)" | |
| STATE["dirty"] = True | |
| return path.name | |
| if not INFO["fallback_used"]: | |
| try: | |
| data, ext = await asyncio.to_thread(_get_kokoro().synth, text) | |
| # WAV را قبل از نوشتن در CACHE به MP3 کم‌حجم تبدیل می‌کنیم (اگر نشد، همان WAV می‌ماند) | |
| data, ext = await asyncio.to_thread(shrink, data, ext, log) | |
| path = CACHE / f"{kbase}.{ext}" | |
| path.write_bytes(data) | |
| STATE["dirty"] = True | |
| INFO["sentences"] = f"Kokoro (Heart) سرعت {KOKORO_SPEED}" | |
| FALLBACK_TEXTS.discard(text) | |
| return path.name | |
| except Exception as e: # noqa: BLE001 | |
| if not INFO["fallback_used"]: | |
| INFO["fallback_used"] = True | |
| INFO["kokoro_error"] = str(e)[:300] | |
| log(f"⚠️ صدای جمله از Kokoro گرفته نشد ({str(e)[:160]}). با صدای جایگزین ادامه می‌دهیم.") | |
| path = CACHE / (fbase + ".mp3") | |
| await edge_save(text, FALLBACK_VOICE, FALLBACK_RATE, path) | |
| STATE["dirty"] = True | |
| FALLBACK_TEXTS.add(text) | |
| INFO["sentences"] = f"جایگزین: {FALLBACK_VOICE} {FALLBACK_RATE}" | |
| return path.name | |
| def attach_lecture(lesson, log): | |
| """ | |
| اگر برای این درس صدای گرامر ذخیره شده باشد، آن را به بخش گرامر درس وصل می‌کند: | |
| section["lecture"] = {"f": نام فایل در بسته, "dur": ثانیه} | |
| اگر درس بخش گرامر نداشت، یکی (قبل از آزمون) ساخته می‌شود. این صدا TTS نمی‌گیرد. | |
| """ | |
| lec = grammar.lecture(lesson["id"]) | |
| if not lec: | |
| return None | |
| lec["name"] = f"g-{lec['sha']}.mp3" | |
| sec = next((s for s in lesson.get("sections", []) if s.get("type") == "grammar"), None) | |
| if sec is None: | |
| sec = {"type": "grammar", "title": "Grammar", "script": []} | |
| secs = lesson.setdefault("sections", []) | |
| pos = next((i for i, s in enumerate(secs) if s.get("type") == "exercises"), len(secs)) | |
| secs.insert(pos, sec) | |
| sec["lecture"] = {"f": lec["name"], "dur": round(lec["dur"], 2)} | |
| log(f" {lesson['id']}: صدای گرامر ({lec['size'] // 1024} KB)") | |
| return lec | |
| async def build_lesson(src_file, packs_dir, sem, log): | |
| lesson = json.loads(src_file.read_text(encoding="utf-8")) | |
| lesson.pop("audio", None) | |
| # AUDIO_REV بالا برود، همه‌ی درس‌ها دوباره دانلود می‌شوند؛ REVS مخصوص همین درس است (مثلاً بعد از عوض شدن صدای گرامر) | |
| lesson["version"] = int(lesson.get("version", 1)) + AUDIO_REV + REVS.get(lesson["id"], 0) | |
| texts = all_texts(lesson) | |
| vocab = vocab_texts(lesson) | |
| lec = attach_lecture(lesson, log) # صدای گرامر (دستی) ؛ TTS نمی‌گیرد | |
| kinds = {t: ("m" if t.startswith(MALE) else "w" if (t in vocab or len(t.split()) == 1) else "s") for t in texts} | |
| async def one(t): | |
| async with sem: | |
| return t, await make_audio(t, kinds[t], log) | |
| log(f" {lesson['id']}: {len(texts)} متن برای صدا…") | |
| # واژه‌ها اول ساخته می‌شوند تا اگر Kokoro نیاز به بیدار شدن داشت، همزمان گرم شود | |
| results = await asyncio.gather(*[one(t) for t in texts]) | |
| audio = {t: {"f": f, "k": kinds[t]} for t, f in results} | |
| LESSON_AUDIO[lesson["id"]] = { | |
| "title": lesson.get("title", ""), | |
| "items": [{"t": t, "k": kinds[t], "f": f, "fb": t in FALLBACK_TEXTS} for t, f in results], | |
| } | |
| lesson["audio"] = audio | |
| imgs = {} | |
| collect_imgs(lesson, imgs) | |
| zip_name = f"{lesson['id']}-v{lesson['version']}.zip" | |
| zip_path = packs_dir / zip_name | |
| tmp_zip = packs_dir / (zip_name + ".tmp") # اول کامل ساخته می‌شود، بعد جا می‌افتد (دانلود نیمه‌کاره نمی‌دهد) | |
| with zipfile.ZipFile(tmp_zip, "w", zipfile.ZIP_DEFLATED) as z: | |
| z.writestr("lesson.json", json.dumps(lesson, ensure_ascii=False)) | |
| for a in audio.values(): | |
| z.write(CACHE / a["f"], f"audio/{a['f']}") | |
| if lec: | |
| z.write(lec["path"], f"audio/{lec['name']}") | |
| for name in imgs: | |
| f = SRC / lesson["id"] / "img" / Path(name).name | |
| if f.exists(): | |
| z.write(f, f"img/{Path(name).name}") | |
| else: | |
| log(f" هشدار: تصویر {name} برای درس {lesson['id']} پیدا نشد ({f})") | |
| os.replace(tmp_zip, zip_path) | |
| data = zip_path.read_bytes() | |
| return lesson, { | |
| "id": lesson["id"], | |
| "title": lesson["title"], | |
| "subtitle": lesson.get("subtitle", ""), | |
| "version": lesson["version"], | |
| "zip": f"packs/{zip_name}", | |
| "size": len(data), | |
| "sha256": hashlib.sha256(data).hexdigest(), | |
| } | |
| def load_curriculum(): | |
| return json.loads((ROOT / "curriculum.json").read_text(encoding="utf-8")) | |
| def lesson_ids(curriculum=None): | |
| """[(شناسه, گروه)] به ترتیب سطح ← ترم ← درس و بعد پادکست""" | |
| c = curriculum or load_curriculum() | |
| out = [(i, f"{lv.get('title', '')} · {t.get('title', '')}") for lv in c["levels"] for t in lv.get("terms", []) | |
| for i in t["lessons"]] | |
| seen = {i for i, _ in out} | |
| out += [(i, "پادکست") for i in c.get("podcast", []) if i not in seen] | |
| return out | |
| def make_manifest(curriculum, metas): | |
| levels = [] | |
| for lv in curriculum["levels"]: | |
| terms = [] | |
| for t in lv.get("terms", []): | |
| lessons = [metas[i] for i in t["lessons"] if i in metas] | |
| if lessons: | |
| terms.append({k: v for k, v in t.items() if k != "lessons"} | {"lessons": lessons}) | |
| levels.append({k: v for k, v in lv.items() if k != "terms"} | {"terms": terms}) | |
| pod_ids = list(curriculum.get("podcast", [])) | |
| return {"version": int(time.time()), "levels": levels, "podcast": [metas[i] for i in pod_ids if i in metas]} | |
| async def build_one(lesson_id, log=print, on_lesson_done=None): | |
| """ | |
| فقط یک درس (یا اپیزود) را می‌سازد: بسته‌ی زیپ و manifest همان درس تازه می‌شود و بقیه دست نمی‌خورند. | |
| صداهای آماده در ذخیره‌ی پایدار دوباره ساخته نمی‌شوند؛ فقط متن‌های تازه‌ی این درس صدا می‌گیرند. | |
| """ | |
| if not re.fullmatch(r"[A-Za-z0-9_-]+", lesson_id or ""): | |
| raise ValueError("شناسه‌ی درس نامعتبر است") | |
| src = SRC / f"{lesson_id}.json" | |
| if not src.exists(): | |
| raise FileNotFoundError(f"فایل {src.name} پیدا نشد") | |
| CACHE.mkdir(parents=True, exist_ok=True) | |
| load_takes() | |
| load_revs() | |
| INFO.update(fallback_used=False, male_fallback=False) # برای این درس دوباره Kokoro امتحان شود | |
| _male_good["v"] = None | |
| packs = OUT / "packs" | |
| packs.mkdir(parents=True, exist_ok=True) | |
| log(f"درس {lesson_id}…") | |
| _, meta = await build_lesson(src, packs, asyncio.Semaphore(CONCURRENCY), log) | |
| METAS[lesson_id] = meta | |
| curriculum = load_curriculum() | |
| tmp = OUT / "manifest.json.tmp" | |
| tmp.write_text(json.dumps(make_manifest(curriculum, METAS), ensure_ascii=False, indent=2), encoding="utf-8") | |
| os.replace(tmp, OUT / "manifest.json") | |
| keep = Path(meta["zip"]).name | |
| for old in packs.glob(f"{lesson_id}-v*.zip"): # نسخه‌های قدیمی همین درس | |
| if old.name != keep: | |
| old.unlink(missing_ok=True) | |
| if on_lesson_done: | |
| await asyncio.to_thread(on_lesson_done) | |
| log(f"درس {lesson_id} آماده شد (نسخه‌ی {meta['version']}).") | |
| return meta | |
| async def build_all(log=print, on_lesson_done=None): | |
| """همه‌ی درس‌ها را می‌سازد. خروجی: (manifest, errors)""" | |
| CACHE.mkdir(parents=True, exist_ok=True) | |
| load_takes() | |
| load_revs() | |
| FALLBACK_TEXTS.clear() | |
| new = OUT.parent / (OUT.name + ".new") | |
| shutil.rmtree(new, ignore_errors=True) | |
| (new / "packs").mkdir(parents=True) | |
| INFO.update(sentences="—", fallback_used=False, kokoro_error="", male="—", male_fallback=False) | |
| _male_good["v"] = None | |
| MP3_STATE.update(warned=False, ok=0, kept_wav=0) | |
| curriculum = load_curriculum() | |
| sem = asyncio.Semaphore(CONCURRENCY) | |
| metas, errors = {}, {} | |
| ids = [i for lv in curriculum["levels"] for t in lv.get("terms", []) for i in t["lessons"]] | |
| # اپیزودهای پادکست هم مثل درس ساخته می‌شوند (متن + صدا + بسته‌ی زیپ) و در manifest زیر کلید "podcast" می‌آیند | |
| pod_ids = list(curriculum.get("podcast", [])) | |
| ids += [i for i in pod_ids if i not in ids] | |
| for lid in ids: | |
| src = SRC / f"{lid}.json" | |
| try: | |
| if not src.exists(): | |
| raise FileNotFoundError(f"فایل {src.name} پیدا نشد") | |
| log(f"درس {lid}…") | |
| _, meta = await build_lesson(src, new / "packs", sem, log) | |
| metas[lid] = meta | |
| except Exception as e: # noqa: BLE001 | |
| errors[lid] = str(e) | |
| log(f" خطا در درس {lid}: {e}") | |
| if on_lesson_done: # بعد از هر درس، صداهای تازه در ذخیره‌ی پایدار نگه داشته می‌شود | |
| await asyncio.to_thread(on_lesson_done) | |
| manifest = make_manifest(curriculum, metas) | |
| (new / "manifest.json").write_text(json.dumps(manifest, ensure_ascii=False, indent=2), encoding="utf-8") | |
| METAS.clear() | |
| METAS.update(metas) | |
| shutil.rmtree(OUT, ignore_errors=True) | |
| new.rename(OUT) | |
| log(f"صدای واژه‌ها: {INFO['words']} | صدای جمله‌ها: {INFO['sentences']}") | |
| if MP3_STATE["ok"] or MP3_STATE["kept_wav"]: | |
| log(f"تبدیل به MP3: {MP3_STATE['ok']} جمله تبدیل شد، {MP3_STATE['kept_wav']} جمله WAV ماند.") | |
| log(f"تمام شد: {len(metas)} درس آماده، {len(errors)} خطا.") | |
| return manifest, errors | |
| if __name__ == "__main__": | |
| asyncio.run(build_all()) |