#!/usr/bin/env python3 """ Final ffplayout schedule generator. Usage: python generate_schedule.py --date YYYY-MM-DD --output /path/to/playlist.json """ import json import random import argparse import os import datetime import subprocess from pathlib import Path from collections import deque # ==================== CONFIGURATION (edit these if needed) ==================== MEDIA_ROOT = "/var/lib/ffplayout/tv-media" DURATIONS_FILE = os.path.join(MEDIA_ROOT, "durations.txt") HISTORY_FILE = "/var/lib/ffplayout/tv-media/used_content_history.txt" WEATHER_DURATION = 30 # seconds SPLIT_DURATION = 1680 # 28 minutes RESUME_OFFSET = -30 # start next segment 30s earlier MAX_COMMERCIALS_PER_BREAK = 300 # 5 minutes TARGET_TOTAL_DURATION = 43200 # 12 hours # Directories to scan for commercials (relative to MEDIA_ROOT) COMMERCIAL_DIRS = [ "Shuffle-for-tv/Commercials", "Shuffle-for-tv/Commercials/bumpers" ] # Content pools (substrings that match paths under MEDIA_ROOT) WEEKDAY_POOLS = { "film": ["Movies/Horror", "Movies/Sci-Fi"], "concert": ["Music/Concerts"], "netv": ["NETV Originals/NEN", "NETV Originals/Coconut Trees", "NETV Originals/The Hastening", "NETV Originals/The Open Stage"], "syndicated": ["syndicated/captain-isotope", "syndicated/anti-memetics-division"], "cartoons": ["carntoons/Disney", "carntoons/Gumby", "carntoons/Popeye", "carntoons/Warner", "carntoons/Colonel Bleep", "carntoons/Misc Shorts", "carntoons/Superman"], "other": ["NETV Originals/Bumpers", "silent/Shorts", "Other"] } FRIDAY_POOLS = { "space_marathon": ["Classic TV/space-patrol", "Classic TV/Other Space/Tom Corbett"], "silent_marathon": ["silent/sky-pirates", "silent/features"], "sasquatch": ["NETV Originals/Sasquatch"], "closing": ["Movies/Horror", "Movies/Sci-Fi", "NETV Originals/Dark Mistresses' Midnight Movie"] } SATURDAY_POOLS = { "cartoons": ["carntoons/Disney", "carntoons/Gumby", "carntoons/Popeye", "carntoons/Warner", "carntoons/Colonel Bleep", "carntoons/Misc Shorts", "carntoons/Superman"], "live_action": ["Classic TV/space-patrol", "Classic TV/Other Space", "Classic TV/Vintage sitcoms"], "this_is_ellijay": ["NETV Originals/This Is Ellijay/ThisIsEllijayEp.2.mp4", "NETV Originals/This Is Ellijay/This Is Ellijay： Episode 3 - Expedition Bigfoot.mp4", "NETV Originals/This Is Ellijay/This Is Ellijay： Episode 2 - The Back Porch Bistro.mp4"] } # ========================================================= # Global state DURATIONS = {} HISTORY = deque(maxlen=7) COMMERCIALS = [] WEATHER_PATH = os.path.join(MEDIA_ROOT, "weather/weather_report.mp4") FILLER_PATH = os.path.join(MEDIA_ROOT, "filler/filler.mp4") def ensure_filler(): """Create a filler file if it does not exist.""" if not os.path.exists(FILLER_PATH): os.makedirs(os.path.dirname(FILLER_PATH), exist_ok=True) # Create 5 minutes of black screen with silent audio cmd = [ "ffmpeg", "-f", "lavfi", "-i", "color=c=black:s=1280x720:d=300", "-f", "lavfi", "-i", "anullsrc", "-c:v", "mpeg2video", "-c:a", "aac", "-t", "300", "-y", FILLER_PATH ] subprocess.run(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) if not os.path.exists(FILLER_PATH): print("Error: Could not create filler file. Please create it manually.") sys.exit(1) def load_durations(): durations = {} with open(DURATIONS_FILE, 'r') as f: for line in f: line = line.strip() if not line or line.startswith('==='): continue if '|' not in line: continue dur, rel = line.split('|') durations[rel] = float(dur) return durations def load_history(): if os.path.exists(HISTORY_FILE): with open(HISTORY_FILE, 'r') as f: for line in f: line = line.strip() if line: HISTORY.append(line) def save_history(): with open(HISTORY_FILE, 'w') as f: for item in HISTORY: f.write(item + '\n') def get_all_files(): """Return dict of rel_path -> absolute path for all .mp4 files in MEDIA_ROOT that are in durations.txt.""" files = {} for root, dirs, names in os.walk(MEDIA_ROOT): for name in names: if not name.endswith('.mp4'): continue full = os.path.join(root, name) rel = os.path.relpath(full, MEDIA_ROOT) if rel in DURATIONS: files[rel] = full return files def filter_by_pool(files, patterns): """Return list of (abs_path, rel_path, duration) for files matching any pattern.""" result = [] for rel, full in files.items(): for pat in patterns: if pat in rel: result.append((full, rel, DURATIONS[rel])) break return result def select_unique(pool, count=1, avoid_previous=True): if not pool: return [] available = pool[:] if avoid_previous: used_set = set(HISTORY) available = [item for item in available if item[1] not in used_set] if not available: available = pool[:] selected = random.sample(available, min(count, len(available))) for s in selected: HISTORY.append(s[1]) return selected def split_program(full_path, total_dur): segments = [] part_start = 0.0 while part_start < total_dur: part_end = min(part_start + SPLIT_DURATION, total_dur) segments.append({ "in": part_start, "out": part_end, "duration": total_dur, "source": full_path }) if part_end >= total_dur: break part_start = max(0, part_end + RESUME_OFFSET) return segments def load_commercials(): """Scan COMMERCIAL_DIRS for existing .mp4 files (absolute paths).""" coms = [] for rel_dir in COMMERCIAL_DIRS: dir_path = os.path.join(MEDIA_ROOT, rel_dir) if not os.path.isdir(dir_path): print(f"Warning: commercial directory not found: {dir_path}") continue for root, _, files in os.walk(dir_path): for f in files: if f.endswith('.mp4') and 'weather' not in f.lower(): full = os.path.join(root, f) if os.path.exists(full): coms.append(full) else: print(f"Warning: {full} listed but missing?") return coms def insert_breaks(segments): """Insert weather and commercials after every ~SPLIT_DURATION seconds of content.""" result = [] content_time = 0.0 for seg in segments: result.append(seg) content_time += seg["out"] - seg["in"] if content_time >= SPLIT_DURATION - 10: # Weather (absolute path) if os.path.exists(WEATHER_PATH): result.append({ "in": 0.0, "out": WEATHER_DURATION, "duration": WEATHER_DURATION, "source": WEATHER_PATH }) else: print("Warning: weather file not found, skipping.") # Commercials (random selection up to MAX_COMMERCIALS_PER_BREAK) com_remaining = MAX_COMMERCIALS_PER_BREAK if COMMERCIALS: shuffled_coms = random.sample(COMMERCIALS, len(COMMERCIALS)) for com in shuffled_coms: rel = os.path.relpath(com, MEDIA_ROOT) dur = DURATIONS.get(rel, 30.0) if dur <= com_remaining: result.append({ "in": 0.0, "out": dur, "duration": dur, "source": com }) com_remaining -= dur else: break content_time = 0.0 return result def ensure_total_duration(segments): """If total duration < TARGET_TOTAL_DURATION, add filler at the end.""" total = sum(seg["out"] - seg["in"] for seg in segments) if total < TARGET_TOTAL_DURATION - 1: needed = TARGET_TOTAL_DURATION - total filler_dur = DURATIONS.get(os.path.relpath(FILLER_PATH, MEDIA_ROOT), 300.0) segments.append({ "in": 0.0, "out": min(needed, filler_dur), "duration": filler_dur, "source": FILLER_PATH }) return segments def build_weekday(): all_files = get_all_files() selected = [] # Film film_pool = filter_by_pool(all_files, WEEKDAY_POOLS["film"]) if not film_pool: raise RuntimeError("No film found in pools") film = select_unique(film_pool, 1)[0] selected.extend(split_program(film[0], film[2])) # Concert concert_pool = filter_by_pool(all_files, WEEKDAY_POOLS["concert"]) if not concert_pool: raise RuntimeError("No concert found") concert = select_unique(concert_pool, 1)[0] selected.extend(split_program(concert[0], concert[2])) # NETV originals (4 shows) netv_pool = filter_by_pool(all_files, WEEKDAY_POOLS["netv"]) if netv_pool: netv_shows = select_unique(netv_pool, 4) for show in netv_shows: if show[2] > 1800: selected.extend(split_program(show[0], show[2])) else: selected.append({"in": 0, "out": show[2], "duration": show[2], "source": show[0]}) # Syndicated (5 episodes) synd_pool = filter_by_pool(all_files, WEEKDAY_POOLS["syndicated"]) if synd_pool: synd_eps = select_unique(synd_pool, 5) for ep in synd_eps: if ep[2] > 1800: selected.extend(split_program(ep[0], ep[2])) else: selected.append({"in": 0, "out": ep[2], "duration": ep[2], "source": ep[0]}) # Cartoons (at least 1 hour) cartoon_pool = filter_by_pool(all_files, WEEKDAY_POOLS["cartoons"]) if cartoon_pool: cartoon_time = 0 while cartoon_time < 3600 and cartoon_pool: cart = select_unique(cartoon_pool, 1)[0] if cartoon_time + cart[2] > 3600 + 300: break selected.append({"in": 0, "out": cart[2], "duration": cart[2], "source": cart[0]}) cartoon_time += cart[2] # Other shorts (optional) other_pool = filter_by_pool(all_files, WEEKDAY_POOLS["other"]) if other_pool: filler = select_unique(other_pool, 3) for f in filler: selected.append({"in": 0, "out": f[2], "duration": f[2], "source": f[0]}) return ensure_total_duration(insert_breaks(selected)) def build_friday(): all_files = get_all_files() selected = [] # Space Patrol marathon (2h) space_pool = filter_by_pool(all_files, FRIDAY_POOLS["space_marathon"]) if space_pool: space_time = 0 while space_time < 7200 and space_pool: ep = select_unique(space_pool, 1)[0] if space_time + ep[2] > 7200 + 300: break selected.extend(split_program(ep[0], ep[2])) space_time += ep[2] # Silent marathon (2h) silent_pool = filter_by_pool(all_files, FRIDAY_POOLS["silent_marathon"]) if silent_pool: silent_time = 0 while silent_time < 7200 and silent_pool: film = select_unique(silent_pool, 1)[0] if silent_time + film[2] > 7200 + 300: break selected.extend(split_program(film[0], film[2])) silent_time += film[2] # Sasquatch sas_pool = filter_by_pool(all_files, FRIDAY_POOLS["sasquatch"]) if sas_pool: sas = select_unique(sas_pool, 1)[0] selected.extend(split_program(sas[0], sas[2])) # Closing horror/SF close_pool = filter_by_pool(all_files, FRIDAY_POOLS["closing"]) if close_pool: close = select_unique(close_pool, 1)[0] selected.extend(split_program(close[0], close[2])) return ensure_total_duration(insert_breaks(selected)) def build_saturday(): all_files = get_all_files() selected = [] # Cartoon block (6h) cartoon_pool = filter_by_pool(all_files, SATURDAY_POOLS["cartoons"]) if cartoon_pool: cartoon_time = 0 last_ellijay = -7200 while cartoon_time < 21600 and cartoon_pool: if cartoon_time - last_ellijay >= 7200: ellijay_pool = filter_by_pool(all_files, SATURDAY_POOLS["this_is_ellijay"]) if ellijay_pool: ell = select_unique(ellijay_pool, 1)[0] selected.append({"in": 0, "out": ell[2], "duration": ell[2], "source": ell[0]}) last_ellijay = cartoon_time cart = select_unique(cartoon_pool, 1)[0] if cartoon_time + cart[2] > 21600 + 300: break selected.append({"in": 0, "out": cart[2], "duration": cart[2], "source": cart[0]}) cartoon_time += cart[2] # Live action block (6h) live_pool = filter_by_pool(all_files, SATURDAY_POOLS["live_action"]) if live_pool: live_time = 0 last_ellijay = -7200 while live_time < 21600 and live_pool: if live_time - last_ellijay >= 7200: ellijay_pool = filter_by_pool(all_files, SATURDAY_POOLS["this_is_ellijay"]) if ellijay_pool: ell = select_unique(ellijay_pool, 1)[0] selected.append({"in": 0, "out": ell[2], "duration": ell[2], "source": ell[0]}) last_ellijay = live_time ep = select_unique(live_pool, 1)[0] if live_time + ep[2] > 21600 + 300: break if ep[2] > 1800: selected.extend(split_program(ep[0], ep[2])) else: selected.append({"in": 0, "out": ep[2], "duration": ep[2], "source": ep[0]}) live_time += ep[2] return ensure_total_duration(insert_breaks(selected)) def main(): parser = argparse.ArgumentParser() parser.add_argument("--date", required=True, help="YYYY-MM-DD") parser.add_argument("--output", required=True, help="Output JSON file path") args = parser.parse_args() # Ensure filler file exists ensure_filler() # Check weather file if not os.path.exists(WEATHER_PATH): print(f"Warning: weather file not found at {WEATHER_PATH}") print("Will skip weather reports in playlist.") # (But ffplayout will still fail if it expects weather. Better to create a dummy weather file) # Let's create a dummy weather file if missing dummy_weather_dir = os.path.dirname(WEATHER_PATH) if not os.path.exists(dummy_weather_dir): os.makedirs(dummy_weather_dir, exist_ok=True) # Create a 30-second dummy weather clip if needed if not os.path.exists(WEATHER_PATH): subprocess.run([ "ffmpeg", "-f", "lavfi", "-i", "color=c=black:s=1280x720:d=30", "-f", "lavfi", "-i", "anullsrc", "-c:v", "mpeg2video", "-c:a", "aac", "-t", "30", "-y", WEATHER_PATH ], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) print(f"Created dummy weather file at {WEATHER_PATH}") # Load durations and history global DURATIONS, COMMERCIALS DURATIONS = load_durations() print(f"Loaded {len(DURATIONS)} entries from {DURATIONS_FILE}") load_history() COMMERCIALS = load_commercials() print(f"Loaded {len(COMMERCIALS)} commercial files") date_obj = datetime.datetime.strptime(args.date, "%Y-%m-%d") weekday = date_obj.weekday() if weekday == 4: print("Building Friday schedule") programs = build_friday() elif weekday == 5: print("Building Saturday schedule") programs = build_saturday() else: print("Building weekday schedule") programs = build_weekday() # Final safety: ensure every source file exists missing = [] for prog in programs: if not os.path.exists(prog["source"]): missing.append(prog["source"]) if missing: print("ERROR: The following files are missing and must be fixed:") for m in missing: print(f" {m}") print("\nSchedule not saved. Please fix missing files and re-run.") return 1 output_path = Path(args.output) output_path.parent.mkdir(parents=True, exist_ok=True) with open(output_path, 'w') as f: json.dump({ "channel": "NETV SRT OUT", "date": args.date, "program": programs }, f, indent=2) save_history() print(f"Schedule written to {output_path}") return 0 if __name__ == "__main__": exit(main())