Кустарная аудиокнига

#!/usr/bin/env python3
from __future__ import annotations

from dataclasses import dataclass
from pathlib import Path
import json
import re
from typing import List, Optional

INPUT = "book.txt" # входной файл с текстом книги
OUT_DIR = Path("chapters")
OUT_DIR.mkdir(exist_ok=True)

# Ищем явные главы/части. "Пролог" вытащим отдельной логикой.
CHAPTER_RE = re.compile(r"(?im)^\s*(глава|часть)\s+([0-9]+|[ivxlcdm]+)\b.*$")

# Явный заголовок пролога (если есть строкой)
PROLOG_RE = re.compile(r"(?im)^\s*пролог\b.*$")

# Если глав нет — режем на части фикс. размера
FALLBACK_PART_CHARS = 120_000

@dataclass
class Section:
    index: int
    title: str      # для человека
    heading: str    # что будет озвучено в начале файла
    text: str

def safe_slug(s: str, max_len: int = 60) -> str:
    s = s.strip().lower()
    s = re.sub(r"[^\wа-яё]+", "_", s, flags=re.IGNORECASE)
    s = s.strip("_")
    return (s[:max_len] if s else "section")

def normalize_preserve_paragraphs(t: str) -> str:
    t = t.replace("\ufeff", "")
    t = t.replace("\r\n", "\n").replace("\r", "\n")
    # Схлопываем слишком много пустых строк
    t = re.sub(r"\n{3,}", "\n\n", t)
    # Пробелы/табы внутри строк — да, но НЕ переносы строк
    t = re.sub(r"[ \t]+", " ", t)
    return t.strip()

def add_paragraph_pauses(t: str) -> str:
    """
    Делает паузы на абзацах:
    Двойной перенос -> '.\n\n' (если перед ним не было знака конца предложения).
    """
    # сначала нормализуем разные варианты "пустой строки"
    t = re.sub(r"\n\s*\n", "\n\n", t)

    # добавим точку перед абзацем, если её нет
    t = re.sub(r"(?<![.!?…])\n\n", ".\n\n", t)

    return t

def split_by_markers(lines: List[str], markers: List[int]) -> List[tuple[str, str]]:
    # возвращает список (title_line, body_text)
    out = []
    for k, start in enumerate(markers):
        end = markers[k + 1] if k + 1 < len(markers) else len(lines)
        title = lines[start].strip()
        body = "\n".join(lines[start + 1:end]).strip()
        if body:
            out.append((title, body))
    return out

def split_into_sections(text: str) -> List[Section]:
    lines = text.split("\n")

    # 1) находим главы/части
    chapter_markers = [i for i, line in enumerate(lines) if CHAPTER_RE.match(line.strip())]

    # 2) если глав не нашли — fallback
    if not chapter_markers:
        sections: List[Section] = []
        raw = text
        start = 0
        idx = 1
        while start < len(raw):
            part = raw[start:start + FALLBACK_PART_CHARS].strip()
            if part:
                sections.append(Section(index=idx, title=f"Part {idx}", heading=f"Часть {idx}.", text=part))
                idx += 1
            start += FALLBACK_PART_CHARS
        return sections

    sections: List[Section] = []

    # 3) ПРОЛОГ: всё до первой главы
    first_marker = chapter_markers[0]
    before = "\n".join(lines[:first_marker]).strip()

    # Если в этом куске есть явное слово "Пролог" строкой — считаем, что это пролог.
    if before:
        sections.append(
            Section(
                index=1,
                title="Пролог",
                heading="Пролог.",
                text=before
            )
        )

    # 4) ГЛАВЫ
    chapters = split_by_markers(lines, chapter_markers)

    chapter_num = 1
    for title, body in chapters:
        sections.append(
            Section(
                index=len(sections) + 1,
                title=title,
                heading=f"Глава {chapter_num}.",
                text=body
            )
        )
        chapter_num += 1

    return sections

def main():
    raw = Path(INPUT).read_text(encoding="utf-8", errors="ignore")
    raw = normalize_preserve_paragraphs(raw)
    raw = add_paragraph_pauses(raw)

    sections = split_into_sections(raw)
    print(f"Sections: {len(sections)}")

    manifest = {
        "input": INPUT,
        "sections_dir": str(OUT_DIR),
        "sections": [],
        "total_chars": 0,
        "total_utf8_bytes": 0,
        "note": "Per-section counts include spoken heading and inserted paragraph pauses.",
    }

    for s in sections:
        fname = f"{s.index:03d}_{safe_slug(s.title)}.txt"
        path = OUT_DIR / fname

        full_text = f"{s.heading}\n\n{s.text}".strip()
        path.write_text(full_text, encoding="utf-8")

        chars = len(full_text)
        utf8_bytes = len(full_text.encode("utf-8"))

        manifest["sections"].append({
            "index": s.index,
            "title": s.title,
            "heading": s.heading,
            "file": str(path),
            "chars": chars,
            "utf8_bytes": utf8_bytes,
        })
        manifest["total_chars"] += chars
        manifest["total_utf8_bytes"] += utf8_bytes

        print(f"[{s.index:03d}] {s.heading} -> {fname} | chars={chars} | bytes={utf8_bytes}")

    (OUT_DIR / "manifest.json").write_text(
        json.dumps(manifest, ensure_ascii=False, indent=2),
        encoding="utf-8"
    )

    print("\nTotals:")
    print(f"  total chars: {manifest['total_chars']}")
    print(f"  total utf-8 bytes: {manifest['total_utf8_bytes']}")
    print(f"Manifest: {OUT_DIR / 'manifest.json'}")

if __name__ == "__main__":
    main()

#!/usr/bin/env python3
from __future__ import annotations

from pathlib import Path
import argparse
import subprocess
import re
import shutil
from google.cloud import texttospeech

# ====== Google TTS settings (defaults) ======
LANG_DEFAULT = "ru-RU"
VOICE_DEFAULT = "ru-RU-Wavenet-D"
RATE_DEFAULT = 0.95

# Google limit is 5000 BYTES per request; keep margin.
MAX_BYTES_DEFAULT = 4800

SENT_CUT_RE = re.compile(r"([.!?…]+)\s+")


def split_sentences(t: str):
    parts = SENT_CUT_RE.split(t)
    out = []
    for i in range(0, len(parts), 2):
        chunk = parts[i].strip()
        punct = parts[i + 1] if i + 1 < len(parts) else ""
        s = (chunk + punct).strip()
        if s:
            out.append(s)
    return out


def chunk_by_utf8_bytes(text: str, max_bytes: int):
    sentences = split_sentences(" ".join(text.split()))
    chunks = []
    cur = ""

    def fits(s: str) -> bool:
        return len(s.encode("utf-8")) <= max_bytes

    for s in sentences:
        if not s:
            continue

        candidate = (cur + " " + s).strip() if cur else s

        if fits(candidate):
            cur = candidate
            continue

        if cur:
            chunks.append(cur)
            cur = ""

        # single sentence too long -> split by words
        if not fits(s):
            words = s.split()
            tmp = ""
            for w in words:
                cand = (tmp + " " + w).strip() if tmp else w
                if fits(cand):
                    tmp = cand
                else:
                    if tmp:
                        chunks.append(tmp)
                    tmp = w
            if tmp:
                chunks.append(tmp)
        else:
            cur = s

    if cur:
        chunks.append(cur)

    return chunks

def ffmpeg_concat_mp3(tmp_dir: Path, output_mp3: Path):
    subprocess.run(
        ["ffmpeg", "-y", "-f", "concat", "-safe", "0", "-i", "files.txt", "-c", "copy", "out.mp3"],
        check=True,
        cwd=str(tmp_dir),
    )
    output_mp3.parent.mkdir(parents=True, exist_ok=True)
    (tmp_dir / "out.mp3").replace(output_mp3)


def main():
    ap = argparse.ArgumentParser(
        description="Generate MP3 from a text file using Google Cloud TTS (WaveNet) with safe UTF-8 byte chunking."
    )
    ap.add_argument("input", help="Path to input .txt file")
    ap.add_argument("-o", "--output", help="Output .mp3 path (default: рядом с входным файлом)")
    ap.add_argument("--voice", default=VOICE_DEFAULT, help=f"Voice name (default: {VOICE_DEFAULT})")
    ap.add_argument("--lang", default=LANG_DEFAULT, help=f"Language code (default: {LANG_DEFAULT})")
    ap.add_argument("--rate", type=float, default=RATE_DEFAULT, help=f"Speaking rate (default: {RATE_DEFAULT})")
    ap.add_argument("--max-bytes", type=int, default=MAX_BYTES_DEFAULT, help=f"Max UTF-8 bytes per request (default: {MAX_BYTES_DEFAULT})")
    ap.add_argument("--tmp-dir", default="tmp_audio_run", help="Temporary directory for mp3 chunks (default: tmp_audio_run)")
    ap.add_argument("--clean", action="store_true", help="Delete temp directory before run")

    args = ap.parse_args()

    input_path = Path(args.input)
    if not input_path.exists():
        raise SystemExit(f"Input not found: {input_path}")

    out_path = Path(args.output) if args.output else input_path.with_suffix(".mp3")

    tmp_dir = Path(args.tmp_dir)
    if args.clean and tmp_dir.exists():
        shutil.rmtree(tmp_dir)
    tmp_dir.mkdir(parents=True, exist_ok=True)

    text = input_path.read_text(encoding="utf-8", errors="ignore").strip()
    if not text:
        raise SystemExit("Input file is empty.")

    parts = chunk_by_utf8_bytes(text, args.max_bytes)
    if not parts:
        raise SystemExit("No chunks produced (unexpected).")

    client = texttospeech.TextToSpeechClient()
    voice = texttospeech.VoiceSelectionParams(language_code=args.lang, name=args.voice)
    audio_cfg = texttospeech.AudioConfig(
        audio_encoding=texttospeech.AudioEncoding.MP3,
        speaking_rate=args.rate,
    )

    filelist = []
    total_chars = 0
    total_utf8_bytes = 0

    for i, part in enumerate(parts, 1):
        b = len(part.encode("utf-8"))
        if b > args.max_bytes:
            raise SystemExit(f"Chunk {i} too big: {b} bytes (max {args.max_bytes})")
        total_utf8_bytes += b
        total_chars += len(part)

        resp = client.synthesize_speech(
            input=texttospeech.SynthesisInput(text=part),
            voice=voice,
            audio_config=audio_cfg,
        )

        fname = tmp_dir / f"part_{i:04d}.mp3"
        fname.write_bytes(resp.audio_content)
        filelist.append(fname)

        if i % 25 == 0 or i == len(parts):
            print(f"Parts: {i}/{len(parts)}")

    (tmp_dir / "files.txt").write_text(
        "\n".join(f"file '{f.name}'" for f in filelist),
        encoding="utf-8"
    )

    ffmpeg_concat_mp3(tmp_dir, out_path)

    print(f"Done: {out_path}")
    print(f"Chunks: {len(parts)}")
    print(f"Total chars sent: {total_chars}")
    print(f"Total utf-8 bytes sent: {total_utf8_bytes}")


if __name__ == "__main__":
    main()