1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
|
"""Audio assembly: combining chunks, speed adjustment, cleanup."""
import logging
import traceback
from pathlib import Path
from typing import Dict, List, Optional
from . import config
logger = logging.getLogger(__name__)
def speed_export_params(speed: float) -> List[str]:
"""Return ffmpeg filter args for pitch-preserving speed adjustment.
Uses ffmpeg's atempo filter, which accepts 0.5..2.0 per filter. Values
outside that range are handled by chaining multiple atempo filters.
"""
if speed <= 0:
raise ValueError(f"Speed must be a positive number, got {speed}")
if abs(speed - 1.0) < 1e-6:
return []
remaining = float(speed)
chain = []
while remaining > 2.0:
chain.append("atempo=2.0")
remaining /= 2.0
while remaining < 0.5:
chain.append("atempo=0.5")
remaining /= 0.5
chain.append(f"atempo={remaining:g}")
return ["-filter:a", ",".join(chain)]
def combine_chunks(total_chunks: int, output_path: Path,
results: Optional[Dict[int, bool]] = None, speed: float = 1.0) -> bool:
"""Combine audio chunks into the final audiobook.
``results`` maps chunk numbers to success flags; failed chunks are
skipped. When ``speed`` differs from 1.0, an additional speed-adjusted
copy is written next to the normal-speed file.
"""
try:
from pydub import AudioSegment
except ImportError:
logger.error("pydub is required to combine audio chunks (pip install pydub)")
return False
try:
combined = AudioSegment.empty()
successful = 0
missing_chunks = []
for i in range(1, total_chunks + 1):
# Skip chunks that failed if we have results tracking
if results is not None and not results.get(i, False):
missing_chunks.append(i)
continue
chunk_file = config.CHUNKS_FOLDER / f"chunk_{i:04d}.wav"
if chunk_file.exists():
try:
combined += AudioSegment.from_wav(str(chunk_file))
successful += 1
if successful % 10 == 0:
logger.info("Combined %d chunks", successful)
except Exception as exc:
logger.warning("Failed to load chunk %d: %s", i, exc)
missing_chunks.append(i)
else:
logger.warning("Chunk file not found: %s", chunk_file)
missing_chunks.append(i)
if successful == 0:
raise RuntimeError("No valid chunks found")
if missing_chunks:
logger.warning("Missing chunks: %s", missing_chunks)
combined.export(str(output_path), format=config.AUDIO_FORMAT, bitrate=config.AUDIO_BITRATE)
logger.info("Audiobook saved: %s (%d/%d chunks)", output_path, successful, total_chunks)
print(f"[INFO] Saved audiobook: {output_path.name} ({successful}/{total_chunks} chunks)")
export_params = speed_export_params(speed)
if export_params:
speed_path = output_path.with_name(f"{output_path.stem}_{speed:g}{output_path.suffix}")
combined.export(str(speed_path), format=config.AUDIO_FORMAT,
bitrate=config.AUDIO_BITRATE, parameters=export_params)
logger.info("Saved speed-adjusted audiobook (%gx): %s", speed, speed_path)
print(f"[INFO] Saved speed-adjusted audiobook: {speed_path.name} ({speed:g}x)")
if missing_chunks:
print(f"[WARNING] Missing chunks: {missing_chunks}")
return True
except Exception as exc:
logger.error("Failed to combine chunks: %s", exc)
logger.error(traceback.format_exc())
return False
def cleanup_chunks() -> None:
"""Remove temporary chunk files from the scratch folder."""
try:
chunk_count = 0
for chunk_file in config.CHUNKS_FOLDER.glob("chunk_*.wav"):
try:
chunk_file.unlink()
chunk_count += 1
except Exception as exc:
logger.warning("Failed to delete %s: %s", chunk_file, exc)
if chunk_count > 0:
logger.info("Cleaned up %d chunk files", chunk_count)
print(f"[INFO] Cleaned up {chunk_count} chunk files")
except Exception as exc:
logger.warning("Cleanup failed: %s", exc)
|
