Refactor project structure for modularity

Move model and pipeline components into dedicated modules
Extract utilities from CLI into shared helpers
Update imports throughout codebase
Refactor project structure for modularity

Move model configuration and hardware selection into `models/` package
Extract pipeline steps into `pipeline/` package
Add utility helpers in `utils/` package
Update CLI imports and references
This commit is contained in:
Damien
2026-06-17 09:55:07 +02:00
parent 1f4e7feb10
commit 383e85c05e
11 changed files with 105 additions and 34 deletions

View File

@@ -0,0 +1 @@
"""Transcribe and summarize audio/video files on Apple Silicon via MLX."""

View File

@@ -2,29 +2,23 @@
Download (optional), transcribe, and summarize audio/video files fully on
Apple Silicon via MLX. Models are selected automatically from the detected
unified memory (see ``config`` and ``device``).
unified memory (see ``models.config`` and ``models.device``).
"""
import argparse
from pathlib import Path
from .device import select_models
from .download import download_from_youtube
from .summarization import summarize_text
from .transcription import transcribe_file
DEFAULT_LANGUAGE = "en" # Use "auto" on the CLI for automatic detection.
from .models import select_models
from .pipeline import download_from_youtube, summarize_text, transcribe_file
from .utils import (
ensure_directory,
print_model_banner,
resolve_language,
write_summary,
)
def _resolve_language(arg_language: str | None) -> str | None:
"""Map the CLI ``--language`` value to a code or ``None`` (auto-detect)."""
language = arg_language if arg_language else DEFAULT_LANGUAGE
if language and language.lower() == "auto":
return None
return language
def main() -> None:
def _parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(
description="Download, transcribe, and summarize audio or video files "
"on Apple Silicon (MLX)."
@@ -49,24 +43,19 @@ def main() -> None:
help="Language code for transcription (e.g. 'en', 'fr', 'es', "
"or 'auto' for detection).",
)
args = parser.parse_args()
return parser.parse_args()
language = _resolve_language(args.language)
def main() -> None:
args = _parse_args()
language = resolve_language(args.language)
# --- Automatic model selection based on detected unified memory ---
ram_gb, tier = select_models()
print("=" * 60)
print(f"Detected unified memory : {ram_gb:.1f} GB")
print(f"Selected hardware tier : >= {tier.min_ram_gb} GB")
print(f"STT model : {tier.stt.repo} ({tier.stt.engine})")
print(f"Summarization model : {tier.summarization_repo}")
print("=" * 60)
print_model_banner(ram_gb, tier)
# --- Working directory for intermediate files ---
data_directory = Path("tmp")
if not data_directory.exists():
data_directory.mkdir(parents=True)
print(f"Created directory: {data_directory}")
data_directory = ensure_directory("tmp")
# --- Resolve input file ---
if args.from_youtube:
@@ -91,11 +80,7 @@ def main() -> None:
# --- Summarization ---
print("Generating summary...")
summary = summarize_text(transcript, tier.summarization_repo)
with open(args.output, "w") as md_file:
md_file.write("# Summary\n\n")
md_file.write(summary)
print(f"Summary written to {args.output}")
write_summary(args.output, summary)
if __name__ == "__main__":

View File

@@ -0,0 +1,13 @@
"""Model configuration and hardware-based model selection."""
from .config import TIERS, STTModel, Tier
from .device import detect_ram_gb, select_models, select_tier
__all__ = [
"TIERS",
"STTModel",
"Tier",
"detect_ram_gb",
"select_models",
"select_tier",
]

View File

@@ -0,0 +1,11 @@
"""The download -> transcribe -> summarize processing pipeline."""
from .download import download_from_youtube
from .summarization import summarize_text
from .transcription import transcribe_file
__all__ = [
"download_from_youtube",
"summarize_text",
"transcribe_file",
]

View File

@@ -7,7 +7,7 @@ Two engines are supported, selected through ``config.STTModel.engine``:
* ``whisper`` -> ``mlx-whisper`` fallback for memory-constrained Macs.
"""
from .config import STTModel
from ..models import STTModel
def _transcribe_voxtral(file_path: str, model_repo: str, language: str | None) -> str:

View File

@@ -0,0 +1,15 @@
"""Cross-cutting helpers (I/O, formatting, CLI argument resolution)."""
from .helpers import (
ensure_directory,
print_model_banner,
resolve_language,
write_summary,
)
__all__ = [
"ensure_directory",
"print_model_banner",
"resolve_language",
"write_summary",
]

View File

@@ -0,0 +1,46 @@
"""Cross-cutting helpers: CLI argument resolution, I/O, and console output.
These keep ``cli.py`` focused on orchestration and the ``pipeline`` modules
focused on the actual ML work.
"""
from pathlib import Path
from ..models import Tier
DEFAULT_LANGUAGE = "en" # Use "auto" on the CLI for automatic detection.
def resolve_language(arg_language: str | None) -> str | None:
"""Map the CLI ``--language`` value to a code or ``None`` (auto-detect)."""
language = arg_language if arg_language else DEFAULT_LANGUAGE
if language and language.lower() == "auto":
return None
return language
def ensure_directory(path: str | Path) -> Path:
"""Create ``path`` (and parents) if missing and return it as a ``Path``."""
directory = Path(path)
if not directory.exists():
directory.mkdir(parents=True)
print(f"Created directory: {directory}")
return directory
def print_model_banner(ram_gb: float, tier: Tier) -> None:
"""Print the detected hardware and the models selected for it."""
print("=" * 60)
print(f"Detected unified memory : {ram_gb:.1f} GB")
print(f"Selected hardware tier : >= {tier.min_ram_gb} GB")
print(f"STT model : {tier.stt.repo} ({tier.stt.engine})")
print(f"Summarization model : {tier.summarization_repo}")
print("=" * 60)
def write_summary(output_path: str | Path, summary: str) -> None:
"""Write ``summary`` to ``output_path`` as a titled markdown document."""
with open(output_path, "w") as md_file:
md_file.write("# Summary\n\n")
md_file.write(summary)
print(f"Summary written to {output_path}")