b239f04eb6
Pipeline audio-only (fără video pe disc) prin ffmpeg direct din URL CloudFront, transcriere cu faster-whisper (CPU), sumarizări structurate cu diagrame SVG la scară reală (STYLE.md documentează toate cerințele de format).
114 lines
3.8 KiB
Python
114 lines
3.8 KiB
Python
"""
|
|
Extract audio-only tracks for the core-curriculum lectures of the ATM course.
|
|
|
|
No video ever touches disk: ffmpeg reads the CloudFront MP4 URL directly over
|
|
HTTP and writes only the audio track (mono, 16kHz -> what whisper wants
|
|
anyway) to disk. Manifest-free; resumable by checking file existence.
|
|
"""
|
|
|
|
import argparse
|
|
import logging
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import requests
|
|
from dotenv import load_dotenv
|
|
|
|
from download import login, get_course_id, get_menu, get_lecture_video_url, slugify
|
|
|
|
# First-phase scope: core curriculum only, not the 147 "ANALIZĂ ZILNICĂ"
|
|
# recaps / Q&A / live-room recordings (confirmed with user 2026-09-11).
|
|
CORE_MODULES = {
|
|
"MODUL 1", "MODUL 2", "MODUL 3", "MODUL 4", "MODUL 5", "MODUL 6",
|
|
"CATALIZATORI", "ANALIZĂ FUNDAMENTALĂ",
|
|
"MASTER ATM 1", "MASTER ATM 2", "MASTER ATM 3", "MASTER ATM 4",
|
|
}
|
|
|
|
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
def extract_audio(url: str, dest: Path) -> bool:
|
|
dest.parent.mkdir(parents=True, exist_ok=True)
|
|
tmp = dest.with_suffix(dest.suffix + ".tmp")
|
|
cmd = [
|
|
"ffmpeg", "-y", "-i", url,
|
|
"-vn", "-acodec", "libmp3lame", "-ar", "16000", "-ac", "1", "-b:a", "64k",
|
|
"-f", "mp3", str(tmp),
|
|
]
|
|
result = subprocess.run(cmd, capture_output=True, text=True, timeout=3600)
|
|
if result.returncode != 0 or not tmp.exists():
|
|
log.error(f" ffmpeg failed: {result.stderr[-1500:]}")
|
|
return False
|
|
tmp.rename(dest)
|
|
return True
|
|
|
|
|
|
def main():
|
|
p = argparse.ArgumentParser(description="Extract audio for ATM core-curriculum lectures")
|
|
p.add_argument("--out", default="audio", help="Output directory")
|
|
p.add_argument("--only", default=None, help="Limit to one module name (for piloting)")
|
|
args = p.parse_args()
|
|
|
|
load_dotenv()
|
|
email = os.getenv("ATM_EMAIL", "")
|
|
password = os.getenv("ATM_PASSWORD", "")
|
|
if not email or not password:
|
|
log.error("Set ATM_EMAIL and ATM_PASSWORD in .env")
|
|
sys.exit(1)
|
|
|
|
out_dir = Path(args.out)
|
|
session = requests.Session()
|
|
session.headers.update({"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"})
|
|
|
|
log.info("Logging in...")
|
|
login(session, email, password)
|
|
course_id = get_course_id(session)
|
|
modules = get_menu(session, course_id)
|
|
|
|
done = skipped = failed = 0
|
|
for mod_idx, mod in enumerate(modules, 1):
|
|
if mod["name"] not in CORE_MODULES:
|
|
continue
|
|
if args.only and mod["name"] != args.only:
|
|
continue
|
|
|
|
mod_dir = out_dir / f"{mod_idx:02d}_{slugify(mod['name'])}"
|
|
log.info(f"Module: {mod['name']} ({len(mod['lectures'])} lectures)")
|
|
|
|
for lec_idx, lec in enumerate(mod["lectures"], 1):
|
|
stem = f"{lec_idx:02d}_{slugify(lec['name'])}"
|
|
dest = mod_dir / f"{stem}.mp3"
|
|
if dest.exists() and dest.stat().st_size > 10_000:
|
|
log.info(f" Skipping (exists): {dest.name}")
|
|
skipped += 1
|
|
continue
|
|
|
|
try:
|
|
video_url = get_lecture_video_url(session, lec["id"])
|
|
except requests.exceptions.HTTPError as e:
|
|
if e.response is not None and e.response.status_code == 403:
|
|
log.info(f" Skipping (not included in access): {lec['name']}")
|
|
continue
|
|
raise
|
|
if not video_url:
|
|
log.info(f" No video (text/quiz): {lec['name']}")
|
|
continue
|
|
|
|
log.info(f" Extracting: {lec['name']}")
|
|
if extract_audio(video_url, dest):
|
|
done += 1
|
|
else:
|
|
failed += 1
|
|
|
|
log.info("=" * 60)
|
|
log.info(f"Extracted {done}, skipped {skipped}, failed {failed}.")
|
|
if failed:
|
|
sys.exit(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|