LongArticlesJob/run_video_extract_text.py

"""
@author: luojunhui
@tools: PyCharm MarsCodeAI && DeepSeek
"""
import sys
import signal
import time
import threading

from tqdm import tqdm

from coldStartTasks.multi_modal import GenerateTextFromVideo
from config import apolloConfig


class VideoProcessing:
    """
    video processing task
    """
    def __init__(self):
        self.generate_text_from_video = GenerateTextFromVideo()
        self.generate_text_from_video.connect_db()
        self.running = True
        self.lock = threading.Lock()
        self.upload_thread = None
        self.convert_thread = None

    def upload_task(self):
        """
        upload task to google cloud storage
        """
        while self.running:
            with self.lock:
                tqdm.write("start upload_video_to_google_ai task...")
                upload_videos_count = self.generate_text_from_video.upload_video_to_google_ai()
                tqdm.write("upload_video_to_google_ai task completed, total upload_videos_count: {}".format(upload_videos_count))
            time.sleep(600)

    def convert_task(self):
        """
        convert video to text
        """
        while self.running:
            with self.lock:
                tqdm.write("Starting convert_video_to_text_with_google_ai task...")
                self.generate_text_from_video.convert_video_to_text_with_google_ai()
                tqdm.write("convert_video_to_text_with_google_ai() task completed.")
            time.sleep(10)

    def stop(self):
        """停止所有线程"""
        with self.lock:
            tqdm.write("Stopping threads...")
            self.running = False  # 设置标志为 False，通知线程退出
        # 等待线程结束
        if self.upload_thread:
            self.upload_thread.join()
        if self.convert_thread:
            self.convert_thread.join()
        tqdm.write("All threads stopped.")


def signal_handler(sig, frame):
    """捕获信号，优雅退出"""
    video_processing.stop()
    sys.exit(0)


if __name__ == '__main__':
    video_processing = VideoProcessing()

    # 创建并启动第一个线程（上传任务）
    video_processing.upload_thread = threading.Thread(target=video_processing.upload_task)
    video_processing.upload_thread.daemon = False
    video_processing.upload_thread.start()

    # 创建并启动第二个线程（转换任务）
    video_processing.convert_thread = threading.Thread(target=video_processing.convert_task)
    video_processing.convert_thread.daemon = False
    video_processing.convert_thread.start()

    # 注册信号处理函数
    signal.signal(signal.SIGINT, signal_handler)
    signal.signal(signal.SIGTERM, signal_handler)

    # 主线程保持运行，防止程序退出
    while video_processing.running:
        time.sleep(1)