456 lines
18 KiB
Python
456 lines
18 KiB
Python
"""Main GUI window."""
|
||
|
||
from __future__ import annotations
|
||
|
||
from PySide6.QtCore import QUrl
|
||
from PySide6.QtGui import QDesktopServices
|
||
|
||
from .. import subscription
|
||
from ..version import display_name
|
||
from .tabs.accounts import AccountsTab
|
||
from .tabs.apply import ApplyTab
|
||
from .tabs.collect import CollectTab
|
||
from .tabs.generate import GenerateTab
|
||
from .tabs.product_suite import ProductSuiteTab
|
||
from .tabs.settings import SettingsTab
|
||
from .widgets import *
|
||
from .workers import SubscriptionCheckWorker
|
||
|
||
|
||
PREFERRED_WINDOW_SIZE = (1180, 760)
|
||
MIN_WINDOW_SIZE = (960, 640)
|
||
WINDOW_SCREEN_MARGIN = 40
|
||
# 观察期真实查询会员状态，但不在客户端限制存量用户工作流。
|
||
SUBSCRIPTION_CHECK_ENABLED = True
|
||
SUBSCRIPTION_ENFORCEMENT_ENABLED = False
|
||
|
||
|
||
def _screen_available_geometry():
|
||
app = QApplication.instance()
|
||
if app is None:
|
||
return None
|
||
screen = app.primaryScreen()
|
||
if screen is None:
|
||
return None
|
||
return screen.availableGeometry()
|
||
|
||
|
||
def _bounded_dimension(available_size, preferred_size, minimum_size, margin):
|
||
if available_size <= 0:
|
||
return preferred_size
|
||
if available_size > margin * 2:
|
||
maximum = available_size - margin * 2
|
||
else:
|
||
maximum = available_size
|
||
if maximum >= minimum_size:
|
||
return max(minimum_size, min(preferred_size, maximum))
|
||
return max(1, min(preferred_size, maximum))
|
||
|
||
|
||
def _fit_and_center_window(
|
||
window,
|
||
available_geometry=None,
|
||
preferred_size=PREFERRED_WINDOW_SIZE,
|
||
minimum_size=MIN_WINDOW_SIZE,
|
||
margin=WINDOW_SCREEN_MARGIN,
|
||
):
|
||
available = available_geometry if available_geometry is not None else _screen_available_geometry()
|
||
if available is None or available.width() <= 0 or available.height() <= 0:
|
||
window.resize(*preferred_size)
|
||
return
|
||
|
||
width = _bounded_dimension(
|
||
available.width(),
|
||
preferred_size[0],
|
||
minimum_size[0],
|
||
margin,
|
||
)
|
||
height = _bounded_dimension(
|
||
available.height(),
|
||
preferred_size[1],
|
||
minimum_size[1],
|
||
margin,
|
||
)
|
||
window.resize(width, height)
|
||
|
||
x = available.x() + max(0, (available.width() - width) // 2)
|
||
y = available.y() + max(0, (available.height() - height) // 2)
|
||
max_x = available.x() + max(0, available.width() - width)
|
||
max_y = available.y() + max(0, available.height() - height)
|
||
window.move(min(max(x, available.x()), max_x), min(max(y, available.y()), max_y))
|
||
|
||
|
||
class MainWindow(QMainWindow):
|
||
"""Main application window with the fixed workflow tabs."""
|
||
|
||
def __init__(
|
||
self,
|
||
db_path=None,
|
||
config=None,
|
||
config_path=None,
|
||
ai_models_path=None,
|
||
startup_status="",
|
||
):
|
||
super().__init__()
|
||
self._initial_window_fit_applied_after_show = False
|
||
self.config = appconfig.load_config(config_path or appconfig.CONFIG_PATH) if config is None else config
|
||
self.config_path = (
|
||
config_path
|
||
or self.config.get("config_path")
|
||
or appconfig.CONFIG_PATH
|
||
)
|
||
self.db_path = _database_path(db_path, self.config)
|
||
self.ai_models_path = (
|
||
ai_models_path
|
||
or self.config.get("ai_models_path")
|
||
or appconfig.ai_models_config_path(self.config)
|
||
)
|
||
self.setWindowTitle(display_name())
|
||
window_icon = app_icon()
|
||
if window_icon is not None:
|
||
self.setWindowIcon(window_icon)
|
||
_fit_and_center_window(self)
|
||
self.setStyleSheet(BUTTON_BASE_STYLE)
|
||
self._settings_tab_index = TAB_TITLES.index("设置")
|
||
self._last_tab_index = 0
|
||
self._reverting_tab_change = False
|
||
self._subscription_status = subscription.SubscriptionStatus(
|
||
subscription.STATUS_UNAVAILABLE
|
||
)
|
||
self._subscription_worker = None
|
||
self._subscription_thread = None
|
||
self._subscription_request_token = 0
|
||
self._subscription_closed = False
|
||
self.tabs = QTabWidget()
|
||
self.tabs.setObjectName("mainTabs")
|
||
self.tabs.setStyleSheet(TAB_STYLE)
|
||
self.tabs.currentChanged.connect(self._on_tab_changed)
|
||
for title in TAB_TITLES:
|
||
self.tabs.addTab(self._build_tab(title), title)
|
||
settings_tab = self._settings_tab()
|
||
if hasattr(settings_tab, "settingsSaved"):
|
||
settings_tab.settingsSaved.connect(self._on_settings_saved)
|
||
central = QWidget()
|
||
central_layout = QVBoxLayout(central)
|
||
central_layout.setContentsMargins(0, 0, 0, 0)
|
||
central_layout.setSpacing(0)
|
||
central_layout.addWidget(self.tabs, 1)
|
||
self.setCentralWidget(central)
|
||
self.show_status("就绪", level="muted")
|
||
if startup_status:
|
||
self.show_status(startup_status, level="success")
|
||
|
||
def showEvent(self, event):
|
||
super().showEvent(event)
|
||
if not self._initial_window_fit_applied_after_show:
|
||
_fit_and_center_window(self)
|
||
self._initial_window_fit_applied_after_show = True
|
||
|
||
def show_status(self, message, level="muted"):
|
||
color = _status_level_color(level)
|
||
self.statusBar().setStyleSheet(f"QStatusBar {{ color: {color}; }}")
|
||
self.statusBar().showMessage(str(message))
|
||
|
||
def _build_tab(self, title):
|
||
if title == "① 导入采集":
|
||
return CollectTab(
|
||
db_path=self.db_path,
|
||
config=self.config,
|
||
status_callback=self.show_status,
|
||
open_accounts_callback=lambda: self.open_accounts_tab(),
|
||
refresh_workflow_callback=lambda: self.refresh_task_tabs(),
|
||
)
|
||
if title == "② AI生成":
|
||
return GenerateTab(
|
||
db_path=self.db_path,
|
||
config=self.config,
|
||
config_path=self.config_path,
|
||
status_callback=self.show_status,
|
||
title_prompt_path=appconfig.title_prompt_path(self.config),
|
||
cover_prompts_dir=appconfig.cover_prompts_dir(self.config),
|
||
open_accounts_callback=lambda: self.open_accounts_tab(),
|
||
refresh_workflow_callback=lambda: self.refresh_task_tabs(),
|
||
subscription_preflight_callback=self.ensure_subscription_for_new_submit,
|
||
)
|
||
if title == "③ 更新蝦皮":
|
||
return ApplyTab(
|
||
db_path=self.db_path,
|
||
config=self.config,
|
||
status_callback=self.show_status,
|
||
open_accounts_callback=lambda: self.open_accounts_tab(),
|
||
open_settings_callback=lambda: self.open_settings_tab(),
|
||
refresh_workflow_callback=lambda: self.refresh_task_tabs(),
|
||
)
|
||
if title == "账号管理":
|
||
return AccountsTab(
|
||
db_path=self.db_path,
|
||
config=self.config,
|
||
status_callback=self.show_status,
|
||
)
|
||
if title == "设置":
|
||
return SettingsTab(
|
||
config=self.config,
|
||
config_path=self.config_path,
|
||
ai_models_path=self.ai_models_path,
|
||
status_callback=self.show_status,
|
||
)
|
||
if title == "商品套图":
|
||
return ProductSuiteTab(
|
||
db_path=self.db_path,
|
||
config=self.config,
|
||
config_path=self.config_path,
|
||
status_callback=self.show_status,
|
||
subscription_preflight_callback=self.ensure_subscription_for_new_submit,
|
||
)
|
||
raise ValueError(f"未知主界面模块：{title}")
|
||
|
||
def refresh_task_tabs(self):
|
||
for index in range(self.tabs.count()):
|
||
widget = self.tabs.widget(index)
|
||
if hasattr(widget, "refresh_tasks"):
|
||
widget.refresh_tasks()
|
||
|
||
def _on_settings_saved(self, backend):
|
||
for index in range(self.tabs.count()):
|
||
widget = self.tabs.widget(index)
|
||
if hasattr(widget, "refresh_gateway_state"):
|
||
widget.refresh_gateway_state()
|
||
label = "自定义网关" if str(backend) == "direct" else "默认网关"
|
||
self.show_status("设置已保存，当前使用%s" % label, level="success")
|
||
self.begin_subscription_check()
|
||
|
||
@property
|
||
def subscription_status(self):
|
||
return self._subscription_status
|
||
|
||
def begin_subscription_check(self):
|
||
"""Refresh membership state without blocking the Qt GUI thread."""
|
||
|
||
self._set_membership_window_title()
|
||
if not SUBSCRIPTION_CHECK_ENABLED:
|
||
self._set_product_access(True)
|
||
self.show_status("会员订阅检测已暂停", level="muted")
|
||
return
|
||
self._subscription_request_token += 1
|
||
token = self._subscription_request_token
|
||
if self._subscription_worker is not None:
|
||
self._subscription_worker.cancel()
|
||
self._set_product_access(not SUBSCRIPTION_ENFORCEMENT_ENABLED)
|
||
self.show_status("正在验证会员状态", level="info")
|
||
worker = SubscriptionCheckWorker(
|
||
config=self.config,
|
||
cmhub_config_path=self.config.get("cmhub_config_path"),
|
||
)
|
||
worker.finished.connect(
|
||
lambda payload, token=token: self._on_subscription_checked(
|
||
token,
|
||
payload,
|
||
)
|
||
)
|
||
worker.cancelled.connect(
|
||
lambda payload, token=token: self._on_subscription_checked(
|
||
token,
|
||
payload,
|
||
)
|
||
)
|
||
thread = run_worker(worker, thread_name="SubscriptionCheckWorker", start=False)
|
||
thread.finished.connect(lambda: self._forget_subscription_thread(thread))
|
||
self._subscription_worker = worker
|
||
self._subscription_thread = thread
|
||
thread.start()
|
||
|
||
def ensure_subscription_for_new_submit(self, action_name="生成"):
|
||
"""Return whether a new product request may start from the current UI."""
|
||
|
||
if not SUBSCRIPTION_CHECK_ENABLED or not SUBSCRIPTION_ENFORCEMENT_ENABLED:
|
||
return True
|
||
status = self._subscription_status
|
||
if status.allows_product_workflows:
|
||
return True
|
||
if self._subscription_thread is not None:
|
||
self.show_status("正在验证会员状态，请稍后再%s" % action_name, level="info")
|
||
return False
|
||
self.show_status("%s：%s" % (action_name, status.user_message), level="warning")
|
||
self.open_settings_tab()
|
||
return False
|
||
|
||
def _on_subscription_checked(self, token, payload):
|
||
if self._subscription_closed or token != self._subscription_request_token:
|
||
return
|
||
result = payload.get("subscription") if isinstance(payload, dict) else None
|
||
if not isinstance(result, subscription.SubscriptionStatus):
|
||
result = subscription.SubscriptionStatus(subscription.STATUS_UNAVAILABLE)
|
||
self._subscription_status = result
|
||
self._apply_subscription_status(result)
|
||
|
||
def _forget_subscription_thread(self, thread):
|
||
if self._subscription_thread is thread:
|
||
self._subscription_thread = None
|
||
self._subscription_worker = None
|
||
|
||
def _apply_subscription_status(self, status):
|
||
self._subscription_status = status
|
||
if status.state == subscription.STATUS_ACTIVE:
|
||
expiry = subscription.format_expiry(status.expires_at)
|
||
self._set_membership_window_title(
|
||
status.account_name,
|
||
status.plan_name,
|
||
"有效至%s" % expiry,
|
||
)
|
||
elif status.state == subscription.STATUS_GRACE:
|
||
expiry = subscription.format_expiry(
|
||
status.grace_expires_at
|
||
) or subscription.format_expiry(status.expires_at)
|
||
self._set_membership_window_title(
|
||
status.account_name,
|
||
status.plan_name,
|
||
"宽限至%s" % expiry,
|
||
)
|
||
else:
|
||
self._set_membership_window_title()
|
||
|
||
access_allowed = (
|
||
status.allows_product_workflows
|
||
or not SUBSCRIPTION_ENFORCEMENT_ENABLED
|
||
)
|
||
self._set_product_access(access_allowed)
|
||
if status.allows_product_workflows:
|
||
if status.state == subscription.STATUS_LEGACY:
|
||
self.show_status(status.user_message, level="muted")
|
||
else:
|
||
message = (
|
||
"会员状态已验证"
|
||
if SUBSCRIPTION_ENFORCEMENT_ENABLED
|
||
else "会员状态已验证，当前处于观察模式"
|
||
)
|
||
self.show_status(message, level="success")
|
||
if SUBSCRIPTION_ENFORCEMENT_ENABLED:
|
||
self._show_subscription_notice_once(status)
|
||
return
|
||
if not SUBSCRIPTION_ENFORCEMENT_ENABLED:
|
||
self.show_status(
|
||
"订阅观察：%s，当前不影响使用" % status.user_message,
|
||
level=self._subscription_level(status),
|
||
)
|
||
return
|
||
self.open_settings_tab()
|
||
self.show_status(status.user_message, level=self._subscription_level(status))
|
||
|
||
@staticmethod
|
||
def _subscription_level(status):
|
||
if status.state in {subscription.STATUS_UNAVAILABLE, subscription.STATUS_KEY_INVALID}:
|
||
return "warning"
|
||
return "danger"
|
||
|
||
def _set_membership_window_title(
|
||
self,
|
||
account_name="",
|
||
plan_name="",
|
||
expiry_text="",
|
||
):
|
||
account = " ".join(str(account_name or "").split())
|
||
plan = " ".join(str(plan_name or "").split())
|
||
expiry = " ".join(str(expiry_text or "").split())
|
||
if account and plan and expiry:
|
||
self.setWindowTitle(
|
||
"%s | %s · %s · %s" % (display_name(), account, plan, expiry)
|
||
)
|
||
return
|
||
self.setWindowTitle(display_name())
|
||
|
||
def _set_product_access(self, enabled):
|
||
for index, title in enumerate(TAB_TITLES):
|
||
self.tabs.setTabEnabled(index, bool(enabled) or title == "设置")
|
||
|
||
def _show_subscription_notice_once(self, status):
|
||
notice_id = str(status.notice_id or "").strip()
|
||
if not notice_id or notice_id == appconfig.subscription_notice_id(self.config):
|
||
return
|
||
box = QMessageBox(self)
|
||
box.setIcon(QMessageBox.Information)
|
||
box.setWindowTitle("已启用会员订阅")
|
||
expiry = subscription.format_expiry(
|
||
status.grace_expires_at if status.state == subscription.STATUS_GRACE else status.expires_at
|
||
)
|
||
text = "已启用蝦皮圈会员订阅服务。当前账号：%s；套餐：%s。" % (
|
||
status.account_name,
|
||
status.plan_name,
|
||
)
|
||
if expiry:
|
||
text += "当前有效至%s。" % expiry
|
||
text += "请在到期前前往线上会员中心选择或续费套餐。"
|
||
box.setText(text)
|
||
manage_button = None
|
||
if status.manage_url:
|
||
manage_button = box.addButton("前往会员中心", QMessageBox.ActionRole)
|
||
close_button = box.addButton("关闭", QMessageBox.AcceptRole)
|
||
box.setDefaultButton(close_button)
|
||
box.exec()
|
||
if manage_button is not None and box.clickedButton() is manage_button:
|
||
QDesktopServices.openUrl(QUrl(status.manage_url))
|
||
try:
|
||
saved = appconfig.save_subscription_notice_id(
|
||
notice_id,
|
||
path=self.config_path,
|
||
)
|
||
except Exception:
|
||
return
|
||
self.config.clear()
|
||
self.config.update(saved)
|
||
|
||
def _on_tab_changed(self, index):
|
||
if self._reverting_tab_change:
|
||
self._last_tab_index = index
|
||
self.show_status(f"当前：{self.tabs.tabText(index)}", level="muted")
|
||
return
|
||
previous = self._last_tab_index
|
||
if previous == self._settings_tab_index and index != previous:
|
||
if not self._confirm_leave_settings_tab():
|
||
self._reverting_tab_change = True
|
||
try:
|
||
self.tabs.setCurrentIndex(previous)
|
||
finally:
|
||
self._reverting_tab_change = False
|
||
self._last_tab_index = previous
|
||
self.show_status(f"当前：{self.tabs.tabText(previous)}", level="muted")
|
||
return
|
||
self._last_tab_index = index
|
||
self.show_status(f"当前：{self.tabs.tabText(index)}", level="muted")
|
||
|
||
def _settings_tab(self):
|
||
return self.tabs.widget(self._settings_tab_index)
|
||
|
||
def _confirm_leave_settings_tab(self):
|
||
settings_tab = self._settings_tab()
|
||
if not hasattr(settings_tab, "is_dirty") or not settings_tab.is_dirty():
|
||
return True
|
||
box = QMessageBox(self)
|
||
box.setWindowTitle("未保存更改")
|
||
box.setText("设置有未保存更改。要先保存再离开吗？")
|
||
save_button = box.addButton("保存", QMessageBox.AcceptRole)
|
||
discard_button = box.addButton("放弃", QMessageBox.DestructiveRole)
|
||
box.addButton("取消", QMessageBox.RejectRole)
|
||
box.setDefaultButton(save_button)
|
||
box.exec()
|
||
clicked = box.clickedButton()
|
||
if clicked is save_button:
|
||
return bool(settings_tab.save_app_settings())
|
||
if clicked is discard_button:
|
||
return bool(settings_tab.discard_unsaved_changes())
|
||
return False
|
||
|
||
def closeEvent(self, event):
|
||
if self._confirm_leave_settings_tab():
|
||
self._subscription_closed = True
|
||
if self._subscription_worker is not None:
|
||
self._subscription_worker.cancel()
|
||
event.accept()
|
||
return
|
||
event.ignore()
|
||
|
||
def open_accounts_tab(self):
|
||
self.tabs.setCurrentIndex(TAB_TITLES.index("账号管理"))
|
||
|
||
def open_settings_tab(self):
|
||
self.tabs.setCurrentIndex(TAB_TITLES.index("设置"))
|