Windows
# VODimageWindows.py
# - One-shot HTTP calls with context managers (no pooled sessions)
# - Retry/backoff
# - Open output once
# - Cross-platform handle/FD logging (Windows + POSIX)
#
# Requires: requests, ipytv
# Install on Windows (PowerShell): py -m pip install requests ipytv
import os
import re
import time
import logging
import argparse
import configparser
import platform
from logging.handlers import RotatingFileHandler
import requests
from requests.exceptions import RequestException
from ipytv import playlist
# ---------- Cross-platform diagnostics (FDs/handles) ----------
def _windows_handle_count() -> int | None:
try:
import ctypes
import ctypes.wintypes as wt
GetProcessHandleCount = ctypes.windll.kernel32.GetProcessHandleCount
GetCurrentProcess = ctypes.windll.kernel32.GetCurrentProcess
GetProcessHandleCount.argtypes = [wt.HANDLE, ctypes.POINTER(wt.DWORD)]
GetProcessHandleCount.restype = wt.BOOL
hproc = GetCurrentProcess()
out = wt.DWORD()
if GetProcessHandleCount(hproc, ctypes.byref(out)):
return int(out.value)
except Exception:
pass
return None
def log_handle_usage(note: str = ""):
"""Logs open file descriptors on POSIX or handle count on Windows."""
msg = None
try:
if os.name == "nt":
count = _windows_handle_count()
if count is not None:
msg = f"Open handles (Windows): {count} {note}"
else:
n = len(os.listdir("/proc/self/fd"))
msg = f"Open FDs (POSIX): {n} {note}"
except Exception:
pass
if msg:
print(msg)
logging.info(msg)
# --------------------------
# Logging setup
# --------------------------
def configure_logging(conf_path: str | None = None):
config = configparser.ConfigParser()
if conf_path:
config.read(conf_path)
else:
config.read('VODimage.conf')
log_to_file = False
log_level = logging.ERROR
if config.has_section('Logging'):
log_to_file = config.getboolean('Logging', 'log_to_file', fallback=False)
log_level_str = config.get('Logging', 'log_level', fallback='ERROR')
log_level = getattr(logging, log_level_str.upper(), logging.ERROR)
if log_to_file:
log_file_path = config.get('Logging', 'log_file', fallback='VODimage.log')
handler = RotatingFileHandler(log_file_path, maxBytes=5 * 1024 * 1024, backupCount=3, encoding="utf-8")
formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
handler.setFormatter(formatter)
root = logging.getLogger()
root.setLevel(log_level)
root.handlers = [handler]
else:
logging.basicConfig(level=log_level, format='%(asctime)s - %(levelname)s - %(message)s')
def log_info(msg: str):
print(msg)
logging.info(msg)
# --------------------------
# Title cleanup helpers
# --------------------------
JUNK_TAGS = [
r'\[.*?\]', r'\b1080p\b', r'\b720p\b', r'\b2160p\b', r'\b4K\b',
r'\bBluRay\b', r'\bWEBRip\b', r'\bWEB\-DL\b', r'\bWEB\b', r'\bDVD\b',
r'\bAVI\b', r'\bHEVC\b', r'\bx265\b', r'\bx264\b', r'\bH\.?264\b', r'\bH\.?265\b'
]
def strip_junk(title: str) -> str:
t = title
for tag in JUNK_TAGS:
t = re.sub(tag, '', t, flags=re.IGNORECASE)
t = re.sub(r'\bS\d{1,2}E\d{1,3}\b', '', t, flags=re.IGNORECASE)
t = re.sub(r'\b\d{4}\b', '', t) # year
t = re.sub(r'\s{2,}', ' ', t).strip(' -_,')
return t.strip()
# --------------------------
# TMDB search via raw requests (one-shot; auto-close)
# --------------------------
def tmdb_search(api_key: str, query: str, is_movie: bool, retries: int = 3, backoff: float = 1.5):
base = "https://api.themoviedb.org/3/search/movie" if is_movie else "https://api.themoviedb.org/3/search/tv"
params = {
"api_key": api_key,
"query": query,
"page": 1,
"language": "en",
}
for attempt in range(1, retries + 1):
try:
with requests.get(base, params=params, timeout=(5, 15)) as r:
r.raise_for_status()
data = r.json()
return data.get("results", []) or []
except RequestException as e:
if attempt == retries:
raise
sleep_for = backoff ** (attempt - 1)
log_info(f"TMDB transient error: {e}; retry in {sleep_for:.1f}s ({attempt}/{retries})")
time.sleep(sleep_for)
# --------------------------
# Processing one VOD entry
# --------------------------
def process_vod(vod, outfh, api_key: str):
try:
attributes = str(getattr(vod, 'attributes', ''))
title = getattr(vod, 'name', '') or ''
url = getattr(vod, 'url', '') or ''
log_info(f"Title: {title}")
processed = False
if "Movie VOD" in attributes:
search_term = strip_junk(re.sub(r'(\s*)HD\s*:\s*', ' ', title, flags=re.IGNORECASE)).strip()
log_info(f"Search term for TMDB (movie): {search_term}")
results = tmdb_search(api_key, search_term, is_movie=True)
if results:
first = results[0]
poster_path = first.get('poster_path')
poster = f"https://image.tmdb.org/t/p/w500{poster_path}" if poster_path else "NONE"
outfh.write(f'#EXTINF:0 tvg-logo="{poster}", group-title="Movie VOD", {search_term}\n')
outfh.write("#EXTGRP:Movie VOD\n")
outfh.write(url + "\n")
processed = True
elif "TV VOD" in attributes:
original_title = title
t = re.sub(r'(\s*)HD\s*:\s*|\s+AU\s+|\s+NZ\s+', ' ', title, flags=re.IGNORECASE)
anchors = ['Jimmy Fallon', 'Jimmy Kimmel', 'Stephen Colbert', '1923', 'Seth Meyers', 'The Daily Show']
pattern = r'(\b(?:' + '|'.join(map(re.escape, anchors)) + r')\b).*'
t = re.sub(pattern, r'\1', t, flags=re.IGNORECASE)
search_term = strip_junk(t)
log_info(f"Search term for TMDB (tv): {search_term}")
results = tmdb_search(api_key, search_term, is_movie=False)
if results:
first = results[0]
poster_path = first.get('poster_path')
poster = f"https://image.tmdb.org/t/p/w500{poster_path}" if poster_path else "NONE"
outfh.write(f'#EXTINF:0 tvg-logo="{poster}", group-title="TV VOD", {original_title}\n')
outfh.write("#EXTGRP:TV VOD\n")
outfh.write(url + "\n")
processed = True
if processed:
log_info("VOD processed successfully.")
else:
log_info("No processing done for this VOD.")
log_info("") # spacing
except KeyError as e:
log_info(f"Key error occurred: {e}")
except IndexError as e:
log_info(f"Index error occurred: {e}")
except Exception as e:
log_info(f"An unexpected error occurred: {e}")
# --------------------------
# Example config writer
# --------------------------
def print_example_config(output_file):
example_config = """\
[UserVariables]
api_key = your_api_key_here
m3u_url = your_m3u_url_here
output_file = C:/path/to/example.m3u
[Logging]
log_to_file = True
log_file = VODimage.log
log_level = INFO
"""
print("VODimage.conf has been created in this directory. Please open it and modify the variables.")
print("Example Configuration File:")
print(example_config)
with open(output_file, 'w', encoding="utf-8") as config_file:
config_file.write(example_config)
# --------------------------
# Path helpers (Windows-friendly)
# --------------------------
def resolve_path(p: str, base_dir: str) -> str:
"""Expand env vars/tilde and make absolute relative to base_dir."""
if not p:
return p
p = os.path.expandvars(os.path.expanduser(p))
if not os.path.isabs(p):
p = os.path.join(base_dir, p)
return os.path.normpath(p)
# --------------------------
# Main
# --------------------------
def main():
parser = argparse.ArgumentParser(description='Process VODs and generate M3U file')
parser.add_argument('--config', default='VODimage.conf', help='Config file (default: VODimage.conf)')
parser.add_argument('--api_key', help='TMDB API key')
parser.add_argument('--m3u_url', help='M3U playlist URL')
parser.add_argument('--output_file', help='Output M3U file')
parser.add_argument('--setup', action='store_true', help='Print example configuration and save as VODimage.conf')
args = parser.parse_args()
# Base dir = folder of the script (Windows users often run from another CWD)
script_dir = os.path.dirname(os.path.abspath(__file__))
conf_path = resolve_path(args.config, script_dir)
if args.setup:
example_config_file = os.path.join(script_dir, 'VODimage.conf')
print_example_config(example_config_file)
return 0
# Load config
config = configparser.ConfigParser()
config.read(conf_path)
# Configure logging early
configure_logging(conf_path)
# Inputs (allow env/cmd overrides; resolve Windows paths)
api_key = args.api_key or config.get('UserVariables', 'api_key', fallback=None)
m3u_url = args.m3u_url or config.get('UserVariables', 'm3u_url', fallback=None)
output_file_raw = args.output_file or config.get('UserVariables', 'output_file', fallback=None)
output_file = resolve_path(output_file_raw, script_dir) if output_file_raw else None
if not api_key or not m3u_url or not output_file:
logging.error("Missing required arguments. Provide either through command-line or in the config file.")
return 1
# Ensure output directory exists on Windows too
out_dir = os.path.dirname(output_file)
if out_dir and not os.path.isdir(out_dir):
os.makedirs(out_dir, exist_ok=True)
# Recreate output
try:
if os.path.exists(output_file):
os.remove(output_file)
except Exception as e:
logging.warning(f"Could not remove existing output file: {e}")
with open(output_file, "w", encoding="utf-8", errors="replace", newline="\n") as outfh:
outfh.write("#EXTM3U\n")
# Load playlist (URL)
pl = playlist.loadu(m3u_url)
# Try counting; if it consumes, reload
try:
total_vods = sum(1 for _ in pl)
except Exception:
total_vods = None
try:
pl = playlist.loadu(m3u_url)
except Exception:
pass
if total_vods is None:
total_vods = 0
for _ in pl:
total_vods += 1
pl = playlist.loadu(m3u_url)
log_info(f"System: {platform.system()} {platform.release()} | Python: {platform.python_version()}")
log_info(f"Total VODs to process: {total_vods}")
for i, vod in enumerate(pl):
log_info(f"Processing VOD {i + 1} of {total_vods}")
process_vod(vod, outfh, api_key)
# Log handle usage for visibility every 50 items
if (i + 1) % 50 == 0:
log_handle_usage(f"after {i+1} VODs")
log_info("Processing completed.")
return 0
if __name__ == '__main__':
raise SystemExit(main())
Linux
#!/usr/bin/env python3
# VODimageLinux.py — script-only fix for "Too many open files"
# Strategy:
# - Avoid tmdbv3api for search calls (which may retain descriptors via pooled sessions)
# - Use one-shot HTTP requests with context managers (requests.get in a `with` block)
# so sockets are deterministically closed after each request
# - Open the output file once
# - Add retry/backoff and frequent FD logging for visibility
#
# Requires:
# requests, ipytv
#
# NOTE: Replace your existing file with this one if you prefer code-only mitigation.
import os
import re
import time
import logging
import argparse
import configparser
from logging.handlers import RotatingFileHandler
import requests
from requests.exceptions import RequestException
from ipytv import playlist
# --------------------------
# Logging setup
# --------------------------
def configure_logging():
config = configparser.ConfigParser()
config.read('VODimage.conf')
log_to_file = False
log_level = logging.ERROR
if config.has_section('Logging'):
log_to_file = config.getboolean('Logging', 'log_to_file', fallback=False)
log_level_str = config.get('Logging', 'log_level', fallback='ERROR')
log_level = getattr(logging, log_level_str.upper(), logging.ERROR)
if log_to_file:
log_file_path = config.get('Logging', 'log_file', fallback='VODimage.log')
handler = RotatingFileHandler(log_file_path, maxBytes=5 * 1024 * 1024, backupCount=3)
formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
handler.setFormatter(formatter)
root = logging.getLogger()
root.setLevel(log_level)
root.handlers = [handler]
else:
logging.basicConfig(level=log_level, format='%(asctime)s - %(levelname)s - %(message)s')
def log_info(msg: str):
print(msg)
logging.info(msg)
def log_fd_count(note=""):
"""Linux-only visibility; safe no-op elsewhere."""
try:
n = len(os.listdir("/proc/self/fd"))
log_info(f"Open FDs: {n} {note}")
except Exception:
pass
# --------------------------
# Title cleanup helpers
# --------------------------
JUNK_TAGS = [
r'\[.*?\]', r'\b1080p\b', r'\b720p\b', r'\b2160p\b', r'\b4K\b',
r'\bBluRay\b', r'\bWEBRip\b', r'\bWEB\-DL\b', r'\bWEB\b', r'\bDVD\b',
r'\bAVI\b', r'\bHEVC\b', r'\bx265\b', r'\bx264\b', r'\bH\.?264\b', r'\bH\.?265\b'
]
def strip_junk(title: str) -> str:
t = title
for tag in JUNK_TAGS:
t = re.sub(tag, '', t, flags=re.IGNORECASE)
# Remove SxxEyy and years and extra spaces/commas
t = re.sub(r'\bS\d{1,2}E\d{1,3}\b', '', t, flags=re.IGNORECASE)
t = re.sub(r'\b\d{4}\b', '', t) # year
t = re.sub(r'\s{2,}', ' ', t).strip(' -_,')
return t.strip()
# --------------------------
# TMDB search via raw requests (one-shot; auto-close)
# --------------------------
def tmdb_search(api_key: str, query: str, is_movie: bool, retries: int = 3, backoff: float = 1.5):
"""
Query TMDB using a one-shot request so the socket is deterministically closed.
Returns a list of result dicts (possibly empty).
"""
base = "https://api.themoviedb.org/3/search/movie" if is_movie else "https://api.themoviedb.org/3/search/tv"
params = {
"api_key": api_key,
"query": query,
"page": 1,
"language": "en",
}
for attempt in range(1, retries + 1):
try:
# context manager ensures the socket/FD is closed when leaving the block
with requests.get(base, params=params, timeout=(5, 15)) as r:
r.raise_for_status()
data = r.json()
return data.get("results", []) or []
except RequestException as e:
if attempt == retries:
raise
sleep_for = backoff ** (attempt - 1)
log_info(f"TMDB transient error: {e}; retry in {sleep_for:.1f}s ({attempt}/{retries})")
time.sleep(sleep_for)
# --------------------------
# Processing one VOD entry
# --------------------------
def process_vod(vod, outfh, api_key: str):
try:
attributes = str(getattr(vod, 'attributes', ''))
title = getattr(vod, 'name', '') or ''
url = getattr(vod, 'url', '') or ''
log_info(f"Title: {title}")
processed = False
if "Movie VOD" in attributes:
search_term = strip_junk(re.sub(r'(\s*)HD\s*:\s*', ' ', title, flags=re.IGNORECASE)).strip()
log_info(f"Search term for TMDB (movie): {search_term}")
results = tmdb_search(api_key, search_term, is_movie=True)
if results:
first = results[0]
poster_path = first.get('poster_path')
poster = f"https://image.tmdb.org/t/p/w500{poster_path}" if poster_path else "NONE"
outfh.write(f'#EXTINF:0 tvg-logo="{poster}", group-title="Movie VOD", {search_term}\n')
outfh.write("#EXTGRP:Movie VOD\n")
outfh.write(url + "\n")
processed = True
elif "TV VOD" in attributes:
original_title = title
t = re.sub(r'(\s*)HD\s*:\s*|\s+AU\s+|\s+NZ\s+', ' ', title, flags=re.IGNORECASE)
anchors = ['Jimmy Fallon', 'Jimmy Kimmel', 'Stephen Colbert', '1923', 'Seth Meyers', 'The Daily Show']
pattern = r'(\b(?:' + '|'.join(map(re.escape, anchors)) + r')\b).*'
t = re.sub(pattern, r'\1', t, flags=re.IGNORECASE)
search_term = strip_junk(t)
log_info(f"Search term for TMDB (tv): {search_term}")
results = tmdb_search(api_key, search_term, is_movie=False)
if results:
first = results[0]
poster_path = first.get('poster_path')
poster = f"https://image.tmdb.org/t/p/w500{poster_path}" if poster_path else "NONE"
outfh.write(f'#EXTINF:0 tvg-logo="{poster}", group-title="TV VOD", {original_title}\n')
outfh.write("#EXTGRP:TV VOD\n")
outfh.write(url + "\n")
processed = True
if processed:
log_info("VOD processed successfully.")
else:
log_info("No processing done for this VOD.")
log_info("") # spacing
except KeyError as e:
log_info(f"Key error occurred: {e}")
except IndexError as e:
log_info(f"Index error occurred: {e}")
except Exception as e:
log_info(f"An unexpected error occurred: {e}")
# --------------------------
# Example config writer
# --------------------------
def print_example_config(output_file):
example_config = """\
[UserVariables]
api_key = your_api_key_here
m3u_url = your_m3u_url_here
output_file = /path/to/example.m3u
[Logging]
log_to_file = True
log_file = VODimage.log
log_level = INFO
"""
print("VODimage.conf has been created in this directory. Please open it and modify the variables.")
print("Example Configuration File:")
print(example_config)
with open(output_file, 'w') as config_file:
config_file.write(example_config)
# --------------------------
# Main
# --------------------------
def main():
parser = argparse.ArgumentParser(description='Process VODs and generate M3U file')
parser.add_argument('--config', default='VODimage.conf', help='Config file (default: VODimage.conf)')
parser.add_argument('--api_key', help='TMDB API key')
parser.add_argument('--m3u_url', help='M3U playlist URL')
parser.add_argument('--output_file', help='Output M3U file')
parser.add_argument('--setup', action='store_true', help='Print example configuration and save as VODimage.conf')
args = parser.parse_args()
if args.setup:
example_config_file = 'VODimage.conf'
print_example_config(example_config_file)
return 0
# Load config
config = configparser.ConfigParser()
config.read(args.config)
# Configure logging early
configure_logging()
# Inputs
api_key = args.api_key or config.get('UserVariables', 'api_key', fallback=None)
m3u_url = args.m3u_url or config.get('UserVariables', 'm3u_url', fallback=None)
output_file = args.output_file or config.get('UserVariables', 'output_file', fallback=None)
if not api_key or not m3u_url or not output_file:
logging.error("Missing required arguments. Provide either through command-line or in the config file.")
return 1
# Recreate output
try:
if os.path.exists(output_file):
os.remove(output_file)
except Exception as e:
logging.warning(f"Could not remove existing output file: {e}")
with open(output_file, "w", encoding="utf-8", errors="replace") as outfh:
outfh.write("#EXTM3U\n")
# Load playlist
pl = playlist.loadu(m3u_url)
# Try counting; if it consumes, reload
try:
total_vods = sum(1 for _ in pl)
except Exception:
total_vods = None
try:
pl = playlist.loadu(m3u_url)
except Exception:
pass
if total_vods is None:
total_vods = 0
for _ in pl:
total_vods += 1
pl = playlist.loadu(m3u_url)
log_info(f"Total VODs to process: {total_vods}")
for i, vod in enumerate(pl):
log_info(f"Processing VOD {i + 1} of {total_vods}")
process_vod(vod, outfh, api_key)
# Log FD count frequently for visibility
if (i + 1) % 50 == 0:
log_fd_count(f"after {i+1} VODs")
log_info("Processing completed.")
return 0
if __name__ == '__main__':
raise SystemExit(main())
Total 0 Votes
0
0