core.py

import re
import ffmpeg
import os
import zipfile

branding = r"""
    ############################################################
    #                                                          #
    #        ___           ___                        ___      #
    #       /  /\         /  /\          ___         /  /\     #
    #      /  /::\       /  /:/         /__/\       /  /:/_    #
    #     /  /:/\:\     /  /:/          \  \:\     /  /:/ /\   #
    #    /  /:/~/::\   /  /:/  ___       \  \:\   /  /:/ /:/_  #
    #   /__/:/ /:/\:\ /__/:/  /  /\  ___  \__\:\ /__/:/ /:/ /\ #
    #   \  \:\/:/__\/ \  \:\ /  /:/ /__/\ |  |:| \  \:\/:/ /:/ #
    #    \  \::/       \  \:\  /:/  \  \:\|  |:|  \  \::/ /:/  #
    #     \  \:\        \  \:\/:/    \  \:\__|:|   \  \:\/:/   #
    #      \  \:\        \  \::/      \__\::::/     \  \::/    #
    #       \__\/         \__\/           ~~~~       \__\/     #
    #                                                          #
    #             Adobe Connect Video Extractor                #
    #                MADE BY KASRA HOSSEINI                    #
    #              Email: 13kasra83@gmail.com                  #
    #                                                          #
    ############################################################
    """
def extract(filename, extract_folder):

    local_zip_filename = filename
    try:
        if not os.path.exists(extract_folder):
            os.makedirs(extract_folder)
            
        print(f"Extracting to folder: {extract_folder}...")
        with zipfile.ZipFile(local_zip_filename, 'r') as zip_ref:
            # Get list of files to set progress bar total
            files = zip_ref.namelist()
            
            for file in files:
                zip_ref.extract(file, extract_folder)
        
        print("Process completed successfully.")
        
    except Exception as e:
        print(f"An error occurred: {e}")

def get_xml_data(filename):
    with open(filename, 'r', encoding='utf-8') as file:
        return file.read()

def get_first_pacing_tick(data):
    pattern = r'<Method><!\[CDATA\[pacingTick\]\]></Method>\s*<Number><!\[CDATA\[(\d+)\]\]></Number>'
    
    match = re.search(pattern, data)
    
    if match:
        return match.group(1)
    return None


def get_file_info(input_data):
    max_w, max_h, max_duration = 0, 0, 0
    for item in input_data:
        try:
            probe = ffmpeg.probe(item['path'])
            duration = float(probe.get('format', {}).get('duration', 0))
            end_time = (item['start_ms'] / 1000.0) + duration
            if end_time > max_duration:
                max_duration = end_time
                
            video_stream = next((s for s in probe['streams'] if s['codec_type'] == 'video'), None)
            if video_stream:
                w, h = int(video_stream.get('width', 0)), int(video_stream.get('height', 0))
                if w > max_w: max_w = w
                if h > max_h: max_h = h
        except:
            continue
    return (max_w or 1280), (max_h or 720), max_duration

def assemble_adobe_connect_with_text(input_data, output_name="final_class.mp4"):
    width, height, total_duration = get_file_info(input_data)
    
    base_layer = (
        ffmpeg.input(f'color=c=black:s={width}x{height}:r=25', f='lavfi', t=total_duration)
        .drawtext(
            text='No Screen Shared!',
            fontsize=45,
            fontcolor='white',
            x='(w-text_w)/2',
            y='(h-text_h)/2'
        )
    )
    
    video_layers = []
    audio_tracks = []

    for item in input_data:
        start_sec = item['start_ms'] / 1000.0
        inp = ffmpeg.input(item['path'])
        
        probe = ffmpeg.probe(item['path'])
        has_video = any(s['codec_type'] == 'video' for s in probe['streams'])
        has_audio = any(s['codec_type'] == 'audio' for s in probe['streams'])
        
        if has_video:
            v = (
                inp.video
                .filter('setpts', f'PTS-STARTPTS+{start_sec}/TB')
                .filter('pad', width, height, '(ow-iw)/2', '(oh-ih)/2')
            )
            video_layers.append(v)
        
        if has_audio:
            a = inp.audio.filter('adelay', f"{item['start_ms']}|{item['start_ms']}")
            audio_tracks.append(a)
    composed_video = base_layer
    for v_layer in video_layers:
        composed_video = ffmpeg.filter([composed_video, v_layer], 'overlay', x=0, y=0, eof_action='pass')

    if audio_tracks:
        combined_audio = ffmpeg.filter(audio_tracks, 'amix', inputs=len(audio_tracks), duration='longest')
    else:
        combined_audio = ffmpeg.input('anullsrc=r=44100:cl=stereo', f='lavfi', t=total_duration).audio
        
    os.system('cls' if os.name == 'nt' else 'clear')
    
    return(
        ffmpeg.output(
            composed_video, 
            combined_audio, 
            output_name,
            vcodec='libx264',
            acodec='aac',
            pix_fmt='yuv420p',
            t=total_duration
        )
        .overwrite_output()
        # .run()

    )

def get_xml_files(file_folder) -> list[str]|None: 
    try:
        xlm_files = [f for f in os.listdir(file_folder) if (f.startswith("screenshare") or f.startswith("cameraVoip")) and f.endswith('.xml')]
        print("Found XML files:", xlm_files)
        return xlm_files
    except Exception as e:
        print(f"Error accessing folder: {e}")
        return None

def import_media(xlm_files, file_folder):
    media_data = []
    try:
        for xml_file in xlm_files:
            data = get_xml_data(os.path.join(file_folder, xml_file))
            result = int(get_first_pacing_tick(data))

            if result:
                # make video delayed by 7 seconds
                if xml_file.startswith("screenshare"):
                    result += 7000
                media_data.append({'path': os.path.join(file_folder, xml_file.replace('.xml', '.flv')), 'start_ms': result})
            else:
                print(f"No pacingTick found in {xml_file}!")
    except Exception as e:
        print(f"Error processing XML files: {e}")
        return None


    media_data.sort(key=lambda x: x['start_ms'])
    return media_data