main.py

#!/usr/bin/python3
# -*- coding: utf-8 -*-
"""
docker build -t live_recorder:v20250703_0853 .
docker stop live_recorder_7870 && docker rm live_recorder_7870
docker run -itd \
--name live_recorder_7870 \
--restart=always \
--network host \
-e server_port=7870 \
live_recorder:v20250703_0853 /bin/bash

"""
import argparse
import asyncio
import logging
import platform
import time
import threading

import gradio as gr

import log
from project_settings import environment, project_path, log_directory, time_zone_info

log.setup_size_rotating(log_directory=log_directory, tz_info=time_zone_info)

from toolbox.os.command import Command
from toolbox.live_info_collect.manager import LiveInfoCollectManager
from toolbox.live_recorder.manager import LiveRecorderManager
from toolbox.video_downloader.manager import VideoDownloaderManager
from toolbox.youtube_video_upload.manager import YoutubeVideoUploadManager

logger = logging.getLogger("main")


def get_args():
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "--live_info_collect_tasks_file",
        default=(project_path / "data/live_info_collect_tasks.json").as_posix(),
        type=str
    )
    parser.add_argument(
        "--live_recorder_tasks_file",
        default=(project_path / "data/live_recorder_tasks.json").as_posix(),
        type=str
    )
    parser.add_argument(
        "--video_download_tasks_file",
        default=(project_path / "data/video_download_tasks.json").as_posix(),
        type=str
    )
    parser.add_argument(
        "--youtube_video_upload_tasks_file",
        default=(project_path / "data/youtube_video_upload_tasks.json").as_posix(),
        type=str
    )
    parser.add_argument(
        "--live_records_dir",
        default=(project_path / "data/live_records").as_posix(),
        type=str
    )
    parser.add_argument(
        "--server_port",
        default=environment.get("server_port", 7860),
        type=int
    )

    args = parser.parse_args()
    return args


def shell(cmd: str):
    return Command.popen(cmd)


def async_thread_wrapper(coro_task):
    logger.info("background_task start")

    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    loop.run_until_complete(coro_task)
    return


def main():
    args = get_args()

    # task info
    run_video_download_tasks = environment.get(key="run_video_download_tasks", default=False, dtype=bool)
    run_live_info_collect_tasks = environment.get(key="run_live_info_collect_tasks", default=False, dtype=bool)
    run_live_recorder_tasks = environment.get(key="run_live_recorder_tasks", default=True, dtype=bool)
    run_youtube_video_upload_tasks = environment.get(key="run_youtube_video_upload_tasks", default=False, dtype=bool)

    logger.info(f"run_video_download_tasks: {run_video_download_tasks}, "
                f"run_live_info_collect_tasks: {run_live_info_collect_tasks}, "
                f"run_live_recorder_tasks: {run_live_recorder_tasks}, "
                f"run_youtube_video_upload_tasks: {run_youtube_video_upload_tasks}"
                )

    # video download
    if run_video_download_tasks is True:
        video_download_manager = VideoDownloaderManager(args.video_download_tasks_file)
        video_download_thread = threading.Thread(target=async_thread_wrapper,
                                                 args=(video_download_manager.run(),),
                                                 daemon=True)
        video_download_thread.start()
        time.sleep(5)

    # live info collect
    if run_live_info_collect_tasks is True:
        live_info_collect_manager = LiveInfoCollectManager(args.live_info_collect_tasks_file)
        live_info_collect_thread = threading.Thread(target=async_thread_wrapper,
                                                    args=(live_info_collect_manager.run(),),
                                                    daemon=True)
        live_info_collect_thread.start()
        time.sleep(5)

    # live recording
    if run_live_recorder_tasks is True:
        live_recording_manager = LiveRecorderManager(args.live_recorder_tasks_file)
        live_recording_thread = threading.Thread(target=async_thread_wrapper,
                                                 args=(live_recording_manager.run(),),
                                                 daemon=True)
        live_recording_thread.start()
        time.sleep(5)

    # youtube video upload
    if run_youtube_video_upload_tasks is True:
        time.sleep(10)
        youtube_video_upload_manager = YoutubeVideoUploadManager(args.youtube_video_upload_tasks_file)
        youtube_video_upload_thread = threading.Thread(target=async_thread_wrapper,
                                                       args=(youtube_video_upload_manager.run(),),
                                                       daemon=True)
        youtube_video_upload_thread.start()
        time.sleep(5)

    # ui
    with gr.Blocks() as blocks:
        gr.Markdown(value="live recording.")
        with gr.Tabs():
            with gr.TabItem("fs"):
                with gr.Row():
                    with gr.Column():
                        fs_filename = gr.Textbox(label="filename")
                        fs_file = gr.File(label="file")
                        fs_file_dir = gr.Textbox(value="data", label="file_dir")
                        fs_query = gr.Button("query", variant="primary")
                    with gr.Column():
                        fs_filelist_dataset_state = gr.State(value=[])
                        fs_filelist_dataset = gr.Dataset(
                            components=[fs_filename, fs_file],
                            samples=fs_filelist_dataset_state.value,
                        )

                    def when_click_query_files(file_dir: str = "data"):
                        file_dir = project_path / file_dir
                        dataset_state = list()
                        for filename in file_dir.glob("**/*.*"):
                            if filename.stem.startswith("."):
                                continue
                            if filename.name.endswith(".py"):
                                continue
                            dataset_state.append((
                                filename.as_posix(),
                                filename.as_posix(),
                            ))

                        dataset = gr.Dataset(
                            components=[fs_filename, fs_file],
                            samples=dataset_state,
                        )
                        return dataset_state, dataset

                    fs_filelist_dataset.click(
                        fn=lambda x: (
                            x[0], x[0]
                        ),
                        inputs=[fs_filelist_dataset],
                        outputs=[fs_filename, fs_file]
                    )
                    fs_query.click(
                        fn=when_click_query_files,
                        inputs=[fs_file_dir],
                        outputs=[fs_filelist_dataset_state, fs_filelist_dataset]
                    )
            with gr.TabItem("shell"):
                shell_text = gr.Textbox(label="cmd")
                shell_button = gr.Button("run")
                shell_output = gr.Textbox(label="output")

                shell_button.click(
                    shell,
                    inputs=[shell_text,],
                    outputs=[shell_output],
                )

    # http://127.0.0.1:7870/
    # http://10.75.27.247:7870/
    blocks.queue().launch(
        # share=True,
        share=False if platform.system() == "Windows" else False,
        server_name="127.0.0.1" if platform.system() == "Windows" else "0.0.0.0",
        server_port=args.server_port
    )
    return


if __name__ == "__main__":
    main()