This Python script helps manage IPTV streams by checking their status in an M3U playlist file. It uses ffprobe to verify whether each stream is online or offline. As it processes the streams, the script provides real-time feedback, displaying offline streams immediately. At the end of the check, a summary of all offline streams is displayed.
Additionally, the script offers the functionality to disable offline streams by commenting them out in the M3U file for future reference. It also automatically skips streams with specific channel names (e.g., "✦●✦"). This tool is ideal for users who want to efficiently monitor and maintain their IPTV playlists, ensuring that only active streams are available.
To run the script, you'll need the following dependencies:
- Python 3.x: The script is written in Python 3.
- ffmpeg: Required for
ffprobeto check stream status.
#!/usr/bin/env python3
import os
import subprocess
import sys
import time
import re
def parse_m3u(file_path):
"""Parse an M3U file and extract stream URLs along with their channel names."""
streams = []
channel_names = []
try:
with open(file_path, "r", encoding="utf-8") as file:
lines = file.readlines()
for line in lines:
line = line.strip()
if line.startswith("#EXTINF:"):
# Extract channel name (tvg-name)
match = re.search(r'tvg-name="([^"]+)"', line)
if match:
channel_name = match.group(1)
# Ignore lines with tvg-name containing "✦●✦"
if "✦●✦" not in channel_name:
channel_names.append(channel_name)
else:
channel_names.append(None) # Add None for this stream to skip later
elif line and not line.startswith("#"):
streams.append(line)
except FileNotFoundError:
print(f"Error: File '{file_path}' not found.")
return [], []
return streams, channel_names, lines # Return streams, channel names, and original lines
def check_stream_ffprobe(url):
"""Check if a stream is online using ffprobe."""
try:
command = [
"ffprobe",
"-hide_banner",
"-loglevel", "error",
"-i", url
]
result = subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=10)
if result.returncode == 0:
return True # Stream is online
else:
return False # Stream is offline
except subprocess.TimeoutExpired:
return False # Timeout indicates offline stream
except Exception as e:
print(f"Error checking stream {url}: {e}")
return False
def update_m3u(file_path, results, original_lines):
"""Update the M3U file by commenting out non-working streams."""
updated_lines = []
for line in original_lines:
stripped_line = line.strip()
if stripped_line and not stripped_line.startswith("#") and stripped_line in results:
status = results[stripped_line]
if "❌ Offline" in status:
# Comment out the offline stream
updated_lines.append(f"#DISABLED {line.strip()}\n")
else:
# Restore previously disabled streams if they're online
updated_lines.append(line.replace("#DISABLED ", ""))
else:
updated_lines.append(line)
# Write the updated lines back to the file
with open(file_path, "w", encoding="utf-8") as file:
file.writelines(updated_lines)
def display_progress_bar(current, total):
"""Display an animated progress bar."""
bar_length = 40 # Length of the progress bar
progress = int(bar_length * current / total)
bar = f"[{'#' * progress}{'-' * (bar_length - progress)}]"
sys.stdout.write(f"\r{bar} {current}/{total} streams checked")
sys.stdout.flush()
def check_streams(file_path):
"""Check all streams in the given M3U file."""
streams, channel_names, original_lines = parse_m3u(file_path)
if not streams:
print("No streams found in the file or the file could not be parsed.")
return {}
results = {}
total_streams = len(streams)
offline_streams = [] # Store offline streams to display them at the end
try:
for index, (url, channel_name) in enumerate(zip(streams, channel_names), start=1):
if channel_name is None:
# Skip streams with tvg-name containing "✦●✦"
continue
is_online = check_stream_ffprobe(url)
status = "✅ Online" if is_online else f"❌ Offline - {channel_name} - {url}"
results[url] = status
# If offline, store the offline stream
if not is_online:
offline_streams.append(status)
# Update the progress bar
display_progress_bar(index, total_streams)
# Print offline summary at the end
if offline_streams:
print("\n\nOffline Streams Summary:")
for offline in offline_streams:
print(offline)
# Update the M3U file based on the results
update_m3u(file_path, results, original_lines)
print() # Newline after the progress bar
except KeyboardInterrupt:
print("\n\nProcess interrupted by user. Cleaning up and exiting...")
sys.exit(0)
return results
def main():
if len(sys.argv) != 2:
print("Usage: ./iptv_check.py <path_to_m3u_file>")
return
input_file = sys.argv[1]
if not os.path.exists(input_file):
print(f"Error: File '{input_file}' not found. Please check the path and try again.")
return
print(f"Starting to check streams in '{input_file}'...\n")
check_streams(input_file)
if __name__ == "__main__":
main()@ubuntu:~/scripts$ ./iptv_check.py iptv.m3u8
Starting to check streams in 'iptv.m3u8'...
[########################################] 147/147 streams checked
Offline Streams Summary:
❌ Offline - AR| GOLD HD - http://example.com/yxcycyxc
❌ Offline - AR| SUPER HD - http://example.com/173
❌ Offline - AR| CNN - http://example.com/40149
❌ Offline - AR| 3 HD - http://example.com/174
❌ Offline - AR| SCIENCE HD - http://example.com/160
❌ Offline - AR| HOME HD - http://example.com/17