import re import os import cv2 import time import glob import logging import requests from pathlib import Path from ffmpeg import FFmpeg from datetime import datetime from watchdog.observers import Observer from moviepy.editor import VideoFileClip from watchdog.events import FileSystemEventHandler # Get environment variables # Streamed Channel Name channel_name = os.environ.get("CHANNEL_NAME") segment_duration = os.environ.get("SEGMENT_DURATION") max_ts_file_number = os.environ.get("MAX_TS_FILE_NUMBER") output_path = "/app/Streams" folder_to_watch = f"{output_path}/{channel_name}/hls" playlist_file_path = f"{output_path}/{channel_name}/hls/playlist.m3u8" # Telegram bot token bot_token="6856325904:AAHSaLJKN8A8VxDGvUq120tJt6QTZWoz_s4" # Telegram group chat ID chat_id="-4111803118" # Create a logger logger = logging.getLogger(__name__) logger.setLevel(logging.INFO) # Create a file handler file_handler = logging.FileHandler(os.path.join(f"{output_path}/{channel_name}/logs", "2M_stream_listner.log")) file_handler.setLevel(logging.INFO) # Create a formatter and set it for the handler formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s') file_handler.setFormatter(formatter) # Add the file handler to the logger logger.addHandler(file_handler) # Function to send a message to the Telegram group def send_telegram_message(bot_token, chat_id, message): """ Sends a message to a Telegram group. Args: bot_token (str): The bot token obtained from BotFather. chat_id (str): The ID of the Telegram group or user. message (str): The message to be sent. Returns: bool: True if the message is sent successfully, False otherwise. """ try: url = f"https://api.telegram.org/bot{bot_token}/sendMessage" payload = { "chat_id": chat_id, "text": message } response = requests.post(url, json=payload) if response.status_code == 200: print("Message sent successfully.") return True else: print(f"Failed to send message. Status code: {response.status_code}") return False except Exception as e: print(f"An error occurred: {e}") return False # Function to create the output directory if it does not exist def create_output_directory(output_dir, channel_name): # Construct the output path output_path = os.path.join(output_dir, channel_name, 'hls') # Create the directory if it doesn't exist if not os.path.isdir(output_path): os.makedirs(output_path, exist_ok=True) logger.info(f"Created directory: {output_path}") else: logger.info(f"Directory already exists: {output_path}") # Function to wait until index.m3u8 file created def wait_for_index_m3u8(output_dir, channel_name): index_m3u8_path = Path(output_dir).resolve() / channel_name / 'hls' / 'index.m3u8' while not index_m3u8_path.exists(): time.sleep(0.2) # Function to ensure playlist.m3u8 file exists def ensure_playlist_m3u8_exists(playlist_file_path): playlist_file_path = Path(playlist_file_path) if not playlist_file_path.is_file(): playlist_content = [ '#EXTM3U', '#EXT-X-VERSION:3', '#EXT-X-TARGETDURATION:6', '#EXT-X-MEDIA-SEQUENCE:0\n' ] with open(playlist_file_path, 'w') as f: f.write('\n'.join(playlist_content)) logging.info(f"Created playlist file: {playlist_file_path}") else: logging.info(f"Playlist file already exists: {playlist_file_path}") # Function to ensure master.m3u8 file exists with required text def ensure_master_m3u8_exists(master_dir): master_path = Path(master_dir) / "master.m3u8" if not master_path.is_file(): master_content = [ '#EXTM3U', '#EXT-X-VERSION:3', '#EXT-X-INDEPENDENT-SEGMENTS', '#EXT-X-MEDIA:TYPE=CLOSED-CAPTIONS,GROUP-ID="CC",LANGUAGE="eng",NAME="english",DEFAULT=YES,AUTOSELECT=YES,INSTREAM-ID="CC1"', '#EXT-X-STREAM-INF:BANDWIDTH=3405600,RESOLUTION=1280x720,CODECS="avc1.4d401f,mp4a.40.2",FRAME-RATE=25,CLOSED-CAPTIONS="CC"', 'http://173.212.199.208/Sanoamedia/Streams/2M_Monde/hls/playlist.m3u8' ] with open(master_path, 'w') as f: f.write('\n'.join(master_content)) logging.info(f"Created master file: {master_path}") else: logging.info(f"Master file already exists: {master_path}") class JingleDetector: def __init__(self, jingles_folder, iframes_folder): self.jingles_folder = jingles_folder self.iframes_folder = iframes_folder self.jingle_paths = self.load_jingle() def create_folder(self, folder_path): """Creates a folder if it doesn't exist.""" os.makedirs(folder_path, exist_ok=True) print(f"Folder created: {folder_path}") def extract_iframes(self, video_path): try: self.create_folder(self.iframes_folder) ffmpeg = ( FFmpeg() .option("y") .input(video_path) .output( os.path.join(self.iframes_folder, f"{os.path.splitext(os.path.basename(video_path))[0]}_%d.png"), vf="select=gt(scene\\,0.10)", vsync="vfr", frame_pts="true" ) ) # Run FFmpeg command ffmpeg.execute() # Get list of created frames return glob.glob(os.path.join(self.iframes_folder, f"{os.path.splitext(os.path.basename(video_path))[0]}_*.png")) except Exception as e: logging.error(f"Error extracting iframes from {video_path}: {e}") return [] def compare_images(self, image_1, image_2): start_time = time.time() image_1=cv2.imread(image_1) image_2=cv2.imread(image_2) first_image_hist = cv2.calcHist([image_1], [0], None, [256], [0, 256]) second_image_hist = cv2.calcHist([image_2], [0], None, [256], [0, 256]) img_hist_diff = cv2.compareHist(first_image_hist, second_image_hist, cv2.HISTCMP_BHATTACHARYYA) img_template_probability_match = cv2.matchTemplate(first_image_hist, second_image_hist, cv2.TM_CCOEFF_NORMED)[0][0] img_template_diff = 1 - img_template_probability_match # taking only 10% of histogram diff, since it's less accurate than template method commutative_image_diff = (img_hist_diff * 0.50) + img_template_diff end_time = time.time() # print("Time taken to compare images in Seconds: {}".format(end_time - start_time)) return commutative_image_diff def load_jingle(self): jingle_paths = [] for jingle_name in os.listdir(self.jingles_folder): jingle_path = os.path.join(self.jingles_folder, jingle_name) if os.path.isdir(jingle_path): for image_name in os.listdir(jingle_path): image_path = os.path.join(jingle_path, image_name) jingle_paths.append((os.path.splitext(image_name)[0], image_path)) return jingle_paths def detect_jingle(self, ts_file): """Detects jingles in the given .ts file.""" iframes = self.extract_iframes(ts_file) # if not iframes: logger.error(f"Error: No IFrames extracted from the .ts file. [{ts_file}]") return None # for iframe_path in iframes: for jingle_name, jingle_image_path in self.jingle_paths: similarity = self.compare_images(jingle_image_path, iframe_path) if similarity < 0.1: # Adjust threshold as needed return jingle_name, iframe_path, similarity return None class TsFilesHandler(FileSystemEventHandler): def __init__(self, playlist_file, segment_limit, use_jingles=False): super(TsFilesHandler, self).__init__() self.use_jingles = use_jingles if self.use_jingles : self.jingle_detector = JingleDetector(f"{output_path}/jingles", f"{output_path}/{channel_name}/iframes") self.playlist_file = playlist_file self.segment_limit = int(segment_limit) self.hls_media_sequence = 0 def on_created(self, event): if not event.is_directory and event.src_path.lower().endswith(".ts"): # Before Adding .ts file to m3u8 file we need to check if file .ts content a jingle frame # And Split the file into Parts # Before adding .ts file to m3u8 file, check if it contains a jingle frame # Detect jingle in the .ts file if self.use_jingles : detection_result = self.jingle_detector.detect_jingle(event.src_path) if detection_result: jingle_name, iframe_path, similarity = detection_result iframe_url = iframe_path.replace("/app/Streams/2M_Monde_HLS/iframes/", "http://173.212.199.208/demo-sect35/Streams/2M_Monde_HLS/iframes/") iframe_path = iframe_path.replace("/app/Streams/2M_Monde_HLS/iframes/", "/var/www/html/demo-sect35/Streams/iframes/") send_telegram_message(bot_token, chat_id, f'Jingle Detected:\nName: {jingle_name}\nSimilarity: {similarity}\nIFrame Path: {iframe_path}\nIFrame URL: {iframe_url}\nDetection Time: {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}') # else: # send_telegram_message(bot_token, chat_id, "No jingle detected.") # Add the .ts file to the playlist self.add_ts_file_to_playlist(os.path.basename(event.src_path), self.get_media_duration(event.src_path)) # Update the media sequence based on the first line in the .ts file self.update_media_sequence(self.extract_first_ts_line()) # Delete the first .ts file if the segment count exceeds the limit self.delete_ts_file_from_playlist() if self.count_segments() >= self.segment_limit + 1 else None def get_media_duration(self, file_path): """Get duration of the media file.""" try: clip = VideoFileClip(file_path) duration = clip.duration clip.close() return duration except Exception as e: logger.error(f"Error while getting duration for {file_path}: {e}") return None def count_segments(self): segment_count = 0 with open(self.playlist_file, 'r') as f: previous_line = None for line in f: if line.strip().endswith(".ts") and previous_line and previous_line.startswith('#EXTINF'): segment_count += 1 previous_line = line.strip() return segment_count def extract_first_ts_line(self): with open(self.playlist_file, 'r') as f: previous_line = None for line in f: # Check if the current line contains a .ts filename and the previous line contains #EXTINF if re.search(r'^[^#].*\.ts$', line.strip()) and previous_line and previous_line.startswith('#EXTINF'): return re.findall(r'\d+', line.strip())[0] previous_line = line.strip() return 0 def update_media_sequence(self, hls_media_sequence): """Update #EXT-X-MEDIA-SEQUENCE tag based on added files.""" with open(self.playlist_file, 'r+') as f: lines = f.readlines() for i, line in enumerate(lines): if line.startswith("#EXT-X-MEDIA-SEQUENCE:"): lines[i] = f"#EXT-X-MEDIA-SEQUENCE:{hls_media_sequence}\n" break f.seek(0) f.writelines(lines) f.truncate() # logger.info(f"Updated #EXT-X-MEDIA-SEQUENCE to {self.hls_media_sequence}") def add_ts_file_to_playlist(self, filename, duration): with open(self.playlist_file, 'a') as f: f.write(f"#EXTINF:{duration}0000,\n") f.write(f"{filename}\n") print(f"Added {filename} to playlist") def delete_ts_file_from_playlist(self): """ Deletes the first .ts file entry from the playlist file. """ # Open the playlist file for reading and writing ('r+') with open(self.playlist_file, 'r+') as f: # Read the content of the playlist file content = f.read() # Move the file pointer to the beginning of the file f.seek(0) # Read the first line after moving the file pointer first_line = f.readline() # Find the position of the first occurrence of '#EXT-X-CUE-OUT' match_cue_out = re.search(r'#EXT-X-CUE-OUT', first_line) # Find the position of the first occurrence of '#EXT-X-CUE-OUT' match_cue_out_cont = re.search(r'#EXT-X-CUE-OUT-CONT', first_line) # Check if match_cue_out: # Move the file pointer to the beginning of the file f.seek(0) # Use regular expression to find and replace the first occurrence of lines # containing '#EXT-X-CUE-OUT', '#EXTINF' and the following line starting with a filename ending with '.ts', # and lines containing '#EXT-X-CUE-IN', effectively removing the entry from the playlist # f.write(re.sub(r'#EXT-X-CUE-OUT.*?\n#EXTINF:[^\n]+\n([^#]+\n).*?#EXT-X-CUE-IN', '', content, count=1)) f.write(re.sub(r'#EXT-X-CUE-OUT.*?\n#EXTINF:[^\n]+\n([^#]+\n)', '', content, count=1)) # Check elif match_cue_out_cont: # Move the file pointer to the beginning of the file f.seek(0) # Use regular expression to find and replace the first occurrence of lines # containing '#EXT-X-CUE-OUT', '#EXTINF' and the following line starting with a filename ending with '.ts', # and lines containing '#EXT-X-CUE-IN', effectively removing the entry from the playlist # f.write(re.sub(r'#EXT-X-CUE-OUT.*?\n#EXTINF:[^\n]+\n([^#]+\n).*?#EXT-X-CUE-IN', '', content, count=1)) f.write(re.sub(r'#EXT-X-CUE-OUT-CONT.*?\n#EXTINF:[^\n]+\n([^#]+\n)', '', content, count=1)) else: # Move the file pointer to the beginning of the file f.seek(0) # Use regular expression to find and replace the first occurrence of a line containing '#EXTINF' and the following line starting with a filename ending with '.ts' # Replace it with an empty string, effectively removing the entry from the playlist f.write(re.sub(r'#EXTINF:[^\n]+\n([^#]+\n)', '', content, count=1)) # Truncate the file to remove any remaining content beyond the new data written f.truncate() # Update the media sequence based on the first line in the .ts file self.update_media_sequence(int(self.extract_first_ts_line())) def main(): # Call the function to create the directory create_output_directory(output_path, channel_name) # Wait until index.m3u8 is created wait_for_index_m3u8(output_path, channel_name) # Call the function to ensure the playlist file exists :: Create initial playlist file if not exists ensure_playlist_m3u8_exists(playlist_file_path) # Call the function to ensure the master file exists ensure_master_m3u8_exists(folder_to_watch) # use_jingles = False # event_handler = TsFilesHandler(playlist_file_path, max_ts_file_number, use_jingles) # observer = Observer() # observer.schedule(event_handler, folder_to_watch, recursive=True) # observer.start() # try: while True: time.sleep(1) except KeyboardInterrupt: observer.stop() # observer.join() if __name__ == "__main__": # Print a message to the console print(f"Starting Monitoring On Folder <{folder_to_watch}> ...") # Run the main function: Call the main function to execute the script. main() # logger.info(f"segment_limit : [{self.segment_limit}]") # logger.info(f"segment_limit : [{self.count_segments()}]")