feat: 使用 Qt 工具封装成可视执行工具.

gui/controllers/__init__.py

@@ -0,0 +1,7 @@
+# -*- coding: utf-8 -*-
+# Name:         __init__.py
+# Author:       小菜
+# Date:         2024/1/12 10:27
+# Description:
+
+from gui.controllers.controller_main import ControllerMain

gui/models/__init__.py

@@ -0,0 +1,8 @@
+# -*- coding: utf-8 -*-
+# Name:         __init__.py
+# Author:       小菜
+# Date:         2024/1/12 10:22
+# Description:
+
+from gui.models.console import ColorfulConsole
+from gui.models.model_main import ModelMain

gui/models/console.py

@@ -0,0 +1,89 @@
+# -*- coding: utf-8 -*-
+# Name:         console.py
+# Author:       小菜
+# Date:         2024/1/10 17:08
+# Description:
+import inspect
+import re
+from collections import defaultdict
+
+from rich.console import Console
+from rich.style import Style
+
+from gui.views import TextInputDialog
+from src.custom import (
+    PROMPT,
+    WARNING,
+    MASTER,
+    DISCLAIMER_TEXT
+)
+
+
+def text_to_colored_html(text, style=PROMPT):
+    # 使用正则匹配 文本是否带颜色
+    color_pattern = re.compile(r'\[.*?]')
+    matches = color_pattern.findall(text)
+    if matches:
+        text = text.replace(matches[0], '').replace(matches[1], '')
+        style = matches[0][1:-1]
+    #
+    style = Style.parse(style)  # 解析 rich 格式的颜色
+    text = text.replace("\n", "<br>")
+    colors = str()
+    if style.color:
+        color = style.color.get_truecolor()  # 获取真实颜色
+        colors = f"color: rgb({color.red}, {color.green}, {color.blue})"
+    return f'<span style="{colors}">{text}</span>'
+
+
+def live_special_extract_info(data: list):
+    show_text = str()
+    _map = defaultdict(list)
+    for item in data:
+        show_text += '直播标题: ' + item['title'] + '\n'
+        show_text += '主播昵称: ' + item['nickname'] + '\n'
+        show_text += '在线观众: ' + item['user_count_str'] + '\n'
+        show_text += '观看次数: ' + item['total_user_str'] + '\n\n'
+        show_text += "FLV 推流地址: " + '\n'
+        for k, v in item['flv_pull_url'].items():
+            show_text += '&nbsp;&nbsp;&nbsp;&nbsp;' + ': ' + k + '\n'
+            _map[k].append(v)
+        show_text += '\n' + "M3U8 推流地址: " + '\n'
+        for k, v in item["hls_pull_url_map"].items():
+            show_text += '&nbsp;&nbsp;&nbsp;&nbsp;' + ': ' + k + '\n'
+            _map[k].append(v)
+        show_text += "\n请选择下载清晰度(选择对应序号，不选代表不下载): "
+    return _map, show_text
+
+
+class ColorfulConsole(Console):
+    def __init__(self, signal, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.signal = signal
+
+    def print(self, *args, style=PROMPT, highlight=False, **kwargs):
+        with self.capture() as capture:
+            super().print(*args, style=style, highlight=highlight, **kwargs, end="")
+        res = text_to_colored_html(capture.get(), style)
+        self.signal.emit(res)
+
+    def input(self, prompt_="", *args, **kwargs):
+        style = WARNING
+        # 这个特别一些，因为文本太长
+        if inspect.currentframe().f_back.f_code.co_name == 'disclaimer':
+            prompt_ = "\n".join(DISCLAIMER_TEXT) + '\n\n' + prompt_
+            style = MASTER
+        # 直播输入的, 也比较特别
+        if inspect.currentframe().f_back.f_code.co_name == 'special_live_mode':
+            _map, _prompt = live_special_extract_info(prompt_)
+            dialog = TextInputDialog(
+                text_to_colored_html(_prompt, style=style), add_radio_btn=True, radio_btn_list=_map.keys()
+            )
+            if dialog.exec():
+                selected_radio = dialog.get_selected_radio_button()
+                return _map.get(selected_radio, list())
+            return list()
+        # 一般的弹框
+        dialog = TextInputDialog(text_to_colored_html(prompt_, style=style))
+        res = dialog.exec()
+        return 'yes' if res else 'no'

gui/models/model_main.py

@@ -0,0 +1,97 @@
+# -*- coding: utf-8 -*-
+# @Author : Frica01
+# @Time   : 2023-12-28 23:24
+# @Name   : model_main.py
+
+from collections import defaultdict
+
+from PySide6.QtCore import (QRunnable, QObject, QThreadPool, Signal, Slot)
+
+
+class WorkerRunnable(QRunnable):
+
+    def __init__(self, func, *args, **kwargs):
+        super().__init__()
+        self.func = func
+        self.kwargs = kwargs
+        self.args = args
+        self.signal = kwargs.get('signal')
+        self.thread_name = kwargs.get('thread_name')
+        self.signal_thread = kwargs.get('signal_thread')
+
+    def run(self):
+        for func in self.func:
+            func(*self.args)
+        self.signal.emit(True)
+        self.signal_thread.emit(self.thread_name)
+
+    def run_other(self):
+        self.func(*self.args)
+        self.signal_thread.emit(self.thread_name)
+
+    def run_live(self):
+        live_data = [self.func(*self.args, **param).run() for param in self.kwargs['params']]
+        self.signal.emit(live_data)
+        self.signal_thread.emit(self.thread_name)
+
+
+class ModelMain(QObject):
+    signal_initial = Signal(bool)
+    signal_live = Signal(list)
+    signal_change_status = Signal(str)
+
+    def __init__(self):
+        super().__init__()
+        self.thread_pool = QThreadPool()
+        self.thread_pool.setMaxThreadCount(10)
+        self.task_status_map = defaultdict(bool)
+        self.signal_change_status.connect(self.change_task_status)
+
+    def initial(self, func_list, *args, **kwargs):
+        thread_name = 'init'
+        if self.task_status_map.get(thread_name):
+            return
+        self.task_status_map[thread_name] = True
+        task = WorkerRunnable(
+            func_list,
+            thread_name=thread_name,
+            signal=self.signal_initial,
+            signal_thread=self.signal_change_status,
+            *args,
+            **kwargs
+        )
+        self.thread_pool.start(task)
+
+    def run_task(self, func, *args, **kwargs):
+        thread_name = kwargs.get('thread_name')
+        if self.task_status_map.get(thread_name):
+            return
+        self.task_status_map[thread_name] = True
+        # print('====> 我启动了', self.task_status_map)
+        task = WorkerRunnable(
+            func,
+            *args,
+            # **kwargs,
+            thread_name=thread_name,
+            signal_thread=self.signal_change_status
+        )
+        self.thread_pool.start(task.run_other)
+
+    def run_live(self, func, *args, **kwargs):
+        thread_name = 'mode_4'
+        if self.task_status_map.get(thread_name):
+            return
+        self.task_status_map[thread_name] = True
+        task = WorkerRunnable(
+            func,
+            *args,
+            **kwargs,
+            signal=self.signal_live,
+            thread_name=thread_name,
+            signal_thread=self.signal_change_status
+        )
+        self.thread_pool.start(task.run_live)
+
+    @Slot(str)
+    def change_task_status(self, thread_name: str):
+        del self.task_status_map[thread_name]

gui/views/__init__.py

@@ -0,0 +1,8 @@
+# -*- coding: utf-8 -*-
+# @Author : Frica01
+# @Time   : 2023-12-30 18:20
+# @Name   : __init__.py
+
+from gui.views.ui.ui_main import Ui_MainWindow
+from gui.views.widgets.custom_grips import CustomGrip
+from gui.views.widgets.view_main import (ViewMain, TextInputDialog)