System Architecture & Technical Documentation

import os
import shutil
import subprocess
from pathlib import Path
from typing import Dict, Any, Optional

def get_ffprobe_binary() -> Optional[str]:
    """Resolve ffprobe executable from system PATH or virtualenv binaries."""
    env_bin = os.getenv("FFPROBE_BINARY")
    if env_bin and (shutil.which(env_bin) or Path(env_bin).is_file()):
        return env_bin
    system_bin = shutil.which("ffprobe")
    if system_bin:
        return system_bin
    return None

def probe_media_file(file_path: str) -> Dict[str, Any]:
    """Inspect duration, streams, and codecs using UTF-8 error-tolerant decoding."""
    ffprobe = get_ffprobe_binary()
    if not ffprobe or not Path(file_path).is_file():
        return {"duration": 0.0, "has_audio": True, "has_video": False}

    cmd = [
        ffprobe, "-v", "error",
        "-show_entries", "format=duration:stream=codec_type,width,height",
        "-of", "default=noprint_wrappers=1:nokey=1",
        str(file_path)
    ]
    # Enforce UTF-8 with replace mode to eliminate ascii decode errors on remote servers
    res = subprocess.run(
        cmd,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        encoding="utf-8",
        errors="replace"
    )
    lines = [ln.strip() for ln in res.stdout.splitlines() if ln.strip()]
    duration = 0.0
    for line in lines:
        try:
            val = float(line)
            if val > duration:
                duration = val
        except ValueError:
            pass

    return {
        "duration": duration,
        "raw_probe": lines
    }

import re
from typing import List, Dict, Any

PAUSE_SPLIT_THRESHOLD_SECONDS = 0.45
PUNCTUATION_BREAK_PATTERN = re.compile(r'[\.\?\!\;\:\,\—\-]')

def segment_transcript_words_into_lines(words: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    """Segment a continuous list of word timestamps into structured musical lyric lines.
    Splits are triggered upon:
      1. Vocal pauses where gap between word_n.end and word_{n+1}.start exceeds 0.45s
      2. Natural phrase punctuation (. ! ? ; , -)
      3. Maximum word constraints per visual line (typically 8-10 words)
    """
    if not words:
        return []

    lines = []
    current_words = []
    
    for idx, w in enumerate(words):
        current_words.append(w)
        is_last_word = (idx == len(words) - 1)
        
        # Check pause gap to next word
        has_pause = False
        if not is_last_word:
            next_w = words[idx + 1]
            gap = float(next_w.get("start", 0)) - float(w.get("end", 0))
            if gap >= PAUSE_SPLIT_THRESHOLD_SECONDS:
                has_pause = True
                
        # Check terminal punctuation in word text
        has_punct = bool(PUNCTUATION_BREAK_PATTERN.search(w.get("word", "")))
        is_lengthy = len(current_words) >= 9

        if is_last_word or has_pause or has_punct or is_lengthy:
            line_text = " ".join([item.get("word", "").strip() for item in current_words]).strip()
            line_start = float(current_words[0].get("start", 0))
            line_end = float(current_words[-1].get("end", line_start + 1.0))
            
            lines.append({
                "line_id": f"line_{len(lines) + 1}",
                "start": round(line_start, 3),
                "end": round(line_end, 3),
                "text": line_text,
                "words": current_words
            })
            current_words = []
            
    return lines

import os
from openai import OpenAI
from config import Config

class OpenAITranscriptionService:
    def __init__(self, api_key: str = None, model: str = None):
        # API key is securely retrieved from environment configuration
        self.api_key = api_key or Config.OPENAI_API_KEY or os.getenv("OPENAI_API_KEY", "")
        self.model = model or Config.OPENAI_TRANSCRIPTION_MODEL or "whisper-1"
        self.client = OpenAI(api_key=self.api_key) if self.api_key else None

    def transcribe_audio_words(self, audio_file_path: str) -> dict:
        """Call Whisper API with timestamp_granularities=['word'] enabled."""
        if not self.client:
            raise ValueError("OpenAI client not initialized. Set OPENAI_API_KEY in environment.")

        with open(audio_file_path, "rb") as audio_file:
            transcript = self.client.audio.transcriptions.create(
                file=audio_file,
                model=self.model,
                response_format="verbose_json",
                timestamp_granularities=["word"]
            )
            
        return transcript.to_dict()

def format_ass_time(seconds: float) -> str:
    """Format floating point seconds into ASS timestamp standard (H:MM:SS.cs)."""
    hrs = int(seconds // 3600)
    mins = int((seconds % 3600) // 60)
    secs = int(seconds % 60)
    centis = int(round((seconds - int(seconds)) * 100))
    if centis >= 100:
        centis = 99
    return f"{hrs}:{mins:02d}:{secs:02d}.{centis:02d}"

def generate_ass_script(lines: list, style: dict, width: int = 1920, height: int = 1080) -> str:
    """Construct complete ASS subtitle specification with karaoke timing wipes."""
    font_name = style.get("font", "Outfit")
    font_size = style.get("font_size", 42)
    primary_color = "&H00FFFFFF"  # White in ASS &HAABBGGRR format
    karaoke_color = "&H00DF77B7"  # Mint/Burgundy accent

    ass_content = [
        "[Script Info]",
        "ScriptType: v4.00+",
        f"PlayResX: {width}",
        f"PlayResY: {height}",
        "ScaledBorderAndShadow: yes",
        "",
        "[V4+ Styles]",
        "Format: Name, Fontname, Fontsize, PrimaryColour, SecondaryColour, OutlineColour, BackColour, Bold, Italic, Underline, StrikeOut, ScaleX, ScaleY, Spacing, Angle, BorderStyle, Outline, Shadow, Alignment, MarginL, MarginR, MarginV, Encoding",
        f"Style: Default,{font_name},{font_size},{primary_color},{karaoke_color},&H00000000,&H80000000,-1,0,0,0,100,100,0,0,1,2,3,2,60,60,80,1",
        "",
        "[Events]",
        "Format: Layer, Start, End, Style, Name, MarginL, MarginR, MarginV, Effect, Text"
    ]

    for line in lines:
        start_ts = format_ass_time(line.get("start", 0))
        end_ts = format_ass_time(line.get("end", 0))
        words = line.get("words", [])
        
        # Build \k tags for karaoke syllable transitions
        dialogue_text = ""
        for w in words:
            duration_cs = max(1, int(round((w.get("end", 0) - w.get("start", 0)) * 100)))
            dialogue_text += f"{{\\k{duration_cs}}}{w.get('word', '')} "
            
        ass_content.append(f"Dialogue: 0,{start_ts},{end_ts},Default,,0,0,0,,{dialogue_text.strip()}")

    return "\n".join(ass_content)

import subprocess
from app.services.media_probe import get_ffmpeg_binary

def execute_render_job(video_in: str, audio_in: str, ass_path: str, output_mp4: str, resolution: str = "720"):
    """Compile final synchronized lyric MP4.
    Resolution '720' uses CRF 23 with veryfast preset for instant preview.
    Resolution '1080' uses CRF 20 with medium preset for broadcast masters.
    """
    ffmpeg_bin = get_ffmpeg_binary()
    is_draft = (resolution == "720")
    
    scale_filter = "scale=-2:720" if is_draft else "scale=-2:1080"
    preset = "veryfast" if is_draft else "medium"
    crf = "23" if is_draft else "20"
    audio_bitrate = "128k" if is_draft else "192k"

    # Escape path separators for FFmpeg libass filter syntax
    escaped_ass = str(ass_path).replace("\\", "/").replace(":", "\\:")
    vf_chain = f"{scale_filter},ass='{escaped_ass}'"

    cmd = [
        ffmpeg_bin, "-y",
        "-i", str(video_in),
        "-i", str(audio_in),
        "-vf", vf_chain,
        "-c:v", "libx264",
        "-preset", preset,
        "-crf", crf,
        "-pix_fmt", "yuv420p",
        "-c:a", "aac",
        "-b:a", audio_bitrate,
        "-shortest",
        "-movflags", "+faststart",  # Optimizes MP4 header for instant web streaming
        str(output_mp4)
    ]

    process = subprocess.run(
        cmd,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        encoding="utf-8",
        errors="replace"
    )
    if process.returncode != 0:
        raise RuntimeError(f"FFmpeg render failure: {process.stderr[-500:]}")
        
    return output_mp4

from datetime import datetime, timezone
from sqlalchemy import String, Float, Integer, Text, ForeignKey, DateTime
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.extensions import db
from app.utils.ids import generate_project_id

class Project(db.Model):
    __tablename__ = "projects"

    id: Mapped[str] = mapped_column(String(64), primary_key=True, default=generate_project_id)
    user_id: Mapped[str] = mapped_column(String(64), ForeignKey("users.id"), nullable=True)
    name: Mapped[str] = mapped_column(String(255), nullable=False)
    status: Mapped[str] = mapped_column(String(32), default="created")
    audio_duration: Mapped[float] = mapped_column(Float, default=0.0)
    width: Mapped[int] = mapped_column(Integer, default=1920)
    height: Mapped[int] = mapped_column(Integer, default=1080)
    current_revision: Mapped[int] = mapped_column(Integer, default=1)
    canonical_json: Mapped[str] = mapped_column(Text, nullable=True)
    created_at: Mapped[datetime] = mapped_column(DateTime, default=lambda: datetime.now(timezone.utc))

    user = relationship("User", back_populates="projects")
    media_assets = relationship("MediaAsset", back_populates="project", cascade="all, delete-orphan")
    lyric_lines = relationship("LyricLine", back_populates="project", cascade="all, delete-orphan")
    render_jobs = relationship("RenderJob", back_populates="project", cascade="all, delete-orphan")

from flask import Blueprint, jsonify, current_app
from flask_login import current_user

ADMIN_EMAIL = "silatanuikipngetich@gmail.com"

@views_bp.route("/health")
def health_check():
    """Restricted health diagnostics.
    Only silatanuikipngetich@gmail.com can inspect technical internals.
    Normal users receive a clean operational status.
    """
    is_admin = bool(
        current_user.is_authenticated and
        current_user.email and
        current_user.email.strip().lower() == ADMIN_EMAIL
    )

    if not is_admin:
        # Hides all server paths, table schemas, and binary paths
        return {
            "status": "healthy",
            "database": "connected",
            "service": "LyricSync Studio",
            "message": "LyricSync Studio is operational."
        }, 200

    # Full technical diagnostics reserved for administrator
    return {
        "status": "healthy",
        "database": "connected",
        "database_tables": inspector.get_table_names(),
        "storage_path": str(current_app.config.get("MEDIA_ROOT")),
        "ffmpeg_binary": get_ffmpeg_binary(),
        "ai_model": current_app.config.get("OPENAI_TRANSCRIPTION_MODEL", "whisper-1"),
        "rendering_engine": "FFmpeg (libass + h264 + aac)",
        "admin_access": True
    }, 200

// Sub-second playback synchronization loop
function startSyncLoop() {
    function loop() {
        if (!audioEl.paused) {
            const currentTime = audioEl.currentTime;
            scrubber.value = currentTime;
            currentTimeDisplay.textContent = formatTime(currentTime);

            // Synchronize active lyric chips
            const activeLine = timeline.findLineAt(currentTime);
            if (activeLine) {
                timeline.highlightLine(activeLine.line_id);
                overlayEl.textContent = activeLine.text;
            }
            
            // Align background video
            if (videoEl && Math.abs(videoEl.currentTime - currentTime) > 0.3) {
                videoEl.currentTime = currentTime;
            }
        }
        requestAnimationFrame(loop);
    }
    requestAnimationFrame(loop);
}