server.py

import os
import sys
import json
import time
import socket
import signal
import datetime
import threading
import urllib.parse
import concurrent.futures
from pathlib import Path
from contextlib import asynccontextmanager
from typing import List, Dict, Any, Optional

import uvicorn
from fastapi import FastAPI, HTTPException
from fastapi.staticfiles import StaticFiles
from fastapi.responses import Response
from mutagen.mp4 import MP4
from feedgen.feed import FeedGenerator

# --- CONFIGURATION ---
HOST = os.environ.get("HOST", "0.0.0.0")
PORT = int(os.environ.get("PORT", "8000"))
PODCAST_CONFIG_JSON = os.environ.get("PODCAST_CONFIG_JSON", "[]")

try:
    podcasts_config = json.loads(PODCAST_CONFIG_JSON)
except Exception as e:
    print(f"Error parsing podcast config: {e}")
    podcasts_config = []

# Global State
# Dictionary mapping podcast ID to its RSS XML string
rss_feeds: Dict[str, str] = {}
# Dictionary mapping podcast ID to its current generation status
generation_statuses: Dict[str, str] = {pod['id']: "Idle" for pod in podcasts_config}
# Dictionary mapping podcast ID to processing counts
processing_stats: Dict[str, Dict[str, int]] = {
    pod['id']: {"processed": 0, "total": 0} for pod in podcasts_config
}

shutdown_event = threading.Event()

def get_local_ip() -> str:
    """Attempts to retrieve the local IP address of the machine."""
    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    try:
        s.connect(('10.255.255.255', 1))
        IP = s.getsockname()[0]
    except Exception:
        IP = '127.0.0.1'
    finally:
        s.close()
    return IP

def parse_file_metadata(file_path: Path, base_url: str) -> Optional[Dict[str, Any]]:
    """Reads metadata for a single file. Designed to run in a thread pool."""
    if shutdown_event.is_set():
        return None
        
    try:
        audio = MP4(file_path)
        title = audio.tags.get('\xa9nam', [file_path.stem])[0]
        
        date_str = audio.tags.get('\xa9day', [''])[0]
        pub_date = None
        if date_str:
            try:
                 pub_date = datetime.datetime.fromisoformat(date_str.replace('Z', '+00:00'))
                 if pub_date.tzinfo is None:
                     pub_date = pub_date.replace(tzinfo=datetime.timezone.utc)
            except Exception:
                pass
        
        if not pub_date:
            try:
                mtime = file_path.stat().st_mtime
                pub_date = datetime.datetime.fromtimestamp(mtime, tz=datetime.timezone.utc)
            except Exception:
                pub_date = datetime.datetime.now(datetime.timezone.utc)

        description = ""
        if 'desc' in audio.tags:
             description = audio.tags['desc'][0]
        elif '\xa9cmt' in audio.tags:
             description = audio.tags['\xa9cmt'][0]

        filename_quoted = urllib.parse.quote(file_path.name)
        # The base URL now includes the podcast ID mount point, so we just append the filename
        file_url = f"{base_url}/{filename_quoted}"
        
        return {
            'id': file_url,
            'title': title,
            'description': description,
            'pub_date': pub_date,
            'url': file_url,
            'length': str(file_path.stat().st_size)
        }
    except Exception as e:
        print(f"Error reading metadata for {file_path.name}: {e}", flush=True)
        return None

def generate_feed_for_podcast(podcast: dict, base_url: str):
    """Generates the RSS feed for a specific podcast."""
    pod_id = podcast['id']
    pod_title = podcast['title']
    pod_dir = Path(podcast['dir'])
    
    if not pod_dir.exists():
         generation_statuses[pod_id] = "Error: Directory not found"
         return

    try:
        generation_statuses[pod_id] = "Scanning files..."
        files = sorted(list(pod_dir.glob("*.m4a")))
        total = len(files)
        processing_stats[pod_id]["total"] = total
        
        if total == 0:
            generation_statuses[pod_id] = "Complete (No files)"
            return

        # Phase 1: Concurrently extract metadata (I/O heavy)
        metadata_results = []
        max_workers = min(16, (os.cpu_count() or 1) * 2) 
        
        print(f"[{pod_id}] Starting parallel metadata extraction for {total} files...", flush=True)
        
        # Audio mount base URL for this specific podcast
        audio_base_url = f"{base_url}/audio/{pod_id}"

        with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
            future_to_file = {executor.submit(parse_file_metadata, fp, audio_base_url): fp for fp in files}
            
            for i, future in enumerate(concurrent.futures.as_completed(future_to_file)):
                if shutdown_event.is_set():
                    return
                    
                processed = i + 1
                processing_stats[pod_id]["processed"] = processed
                
                if processed % 50 == 0 or processed == total:
                    print(f"[{pod_id}] Processed {processed}/{total} files...", flush=True)
                
                result = future.result()
                if result:
                    metadata_results.append(result)

        if shutdown_event.is_set():
            return

        # Phase 2: Build the feed sequentially
        generation_statuses[pod_id] = "Building RSS Feed..."
        
        metadata_results.sort(key=lambda x: x['pub_date'], reverse=True)
        
        fg = FeedGenerator()
        fg.load_extension('podcast')
        fg.title(pod_title)
        fg.description(f'Local archive of {pod_title}')
        fg.link(href=f"{base_url}/{pod_id}/feed.xml", rel='alternate')
        fg.language('en')
        
        for meta in metadata_results:
            fe = fg.add_entry()
            fe.id(meta['id'])
            fe.title(meta['title'])
            fe.description(meta['description'])
            fe.pubDate(meta['pub_date'])
            fe.enclosure(meta['url'], meta['length'], 'audio/mp4')

        rss_feeds[pod_id] = fg.rss_str(pretty=True).decode('utf-8')
        generation_statuses[pod_id] = "Complete"
        print(f"[{pod_id}] Feed generation complete! Ready for clients.", flush=True)
        
    except Exception as e:
        generation_statuses[pod_id] = f"Error: {e}"
        print(f"[{pod_id}] CRITICAL ERROR: {e}", flush=True)

def generate_all_feeds_background(base_url: str):
    """Background thread to handle the long-running scan for all configured podcasts."""
    # We can run each podcast sequentially to save CPU, or thread them.
    # Given they already use ThreadPools under the hood, running the podcasts sequentially is safer for I/O.
    for pod in podcasts_config:
        if shutdown_event.is_set():
            break
        generate_feed_for_podcast(pod, base_url)


@asynccontextmanager
async def lifespan(app: FastAPI):
    local_ip = get_local_ip()
    base_url = f"http://{local_ip}:{PORT}"
    
    print(f"--- SERVER STARTING ON {HOST}:{PORT} ---", flush=True)
    print(f"Found {len(podcasts_config)} podcasts configured.", flush=True)
    
    shutdown_event.clear()
    
    thread = threading.Thread(target=generate_all_feeds_background, args=(base_url,))
    thread.daemon = True
    thread.start()
    
    # Mount podcast directories dynamically
    for pod in podcasts_config:
        pod_dir = Path(pod['dir'])
        pod_id = pod['id']
        if pod_dir.exists():
            app.mount(f"/audio/{pod_id}", StaticFiles(directory=str(pod_dir)), name=f"audio_{pod_id}")
            print(f"Mounted /audio/{pod_id} -> {pod_dir}", flush=True)
        else:
            print(f"Warning: Directory not found for {pod_id} at {pod_dir}", flush=True)
            
    yield
    
    print("Initiating graceful shutdown...", flush=True)
    shutdown_event.set()
    time.sleep(0.5)

app = FastAPI(lifespan=lifespan)

@app.get("/{podcast_id}/feed.xml")
async def get_feed(podcast_id: str):
    if podcast_id not in generation_statuses:
        raise HTTPException(status_code=404, detail="Podcast not found")
        
    if podcast_id in rss_feeds and rss_feeds[podcast_id]:
        return Response(content=rss_feeds[podcast_id], media_type="application/xml")
    
    status = generation_statuses.get(podcast_id, "Unknown")
    stats = processing_stats.get(podcast_id, {"processed": 0, "total": 0})
    processed = stats["processed"]
    total = stats["total"]
    
    progress = f" ({processed}/{total})" if total > 0 else ""
    return Response(
        content=f"<?xml version='1.0' encoding='UTF-8'?><status><state>{status}</state><progress>{progress}</progress></status>",
        media_type="application/xml"
    )

@app.get("/")
async def index():
    return {
        "status": "Server Running",
        "port": PORT,
        "podcasts": [
            {
                "id": pod['id'],
                "title": pod['title'],
                "status": generation_statuses.get(pod['id'], "Unknown"),
                "progress": f"{processing_stats[pod['id']]['processed']}/{processing_stats[pod['id']]['total']}"
            } for pod in podcasts_config
        ]
    }

if __name__ == "__main__":
    signal.signal(signal.SIGINT, lambda s, f: sys.exit(0))
    signal.signal(signal.SIGTERM, lambda s, f: sys.exit(0))
    
    uvicorn.run(app, host=HOST, port=PORT, log_level="info", access_log=False)