4908db24 by 柴进

feat(qml): task #12 业务桥层 — 5 个 QObject 暴露给 QML

bridges/
  auth.py        AuthBridge       login/logout + currentUser/loggedIn (PoC 模式接受任意非空)
  imagegen.py    ImageGenBridge   submitTask + 信号转发 (TaskQueueManager → QML)
  history.py     HistoryBridge    暴露 HistoryListModel + refresh/getItem/deleteItem
  taskqueue.py   TaskQueueBridge  自带 _TaskListModel (QAbstractListModel + roleNames)
  jewelry.py     JewelryBridge    词库增删 + previewPrompt
  _icons.py      build_placeholder_icon (旧 ImageGeneratorWindow.create_placeholder_icon
                 是实例方法,桥层独立一份)

main_qml.py 改造:
  - 顶部加载 config.json + 实例化 core 业务 (HistoryManager / JewelryLibraryManager /
    TaskQueueManager) + 启 audit_logger(有 db_config 时)
  - 5 个桥通过 setContextProperty 注入:appState / auth / imageGen / history /
    taskQueue / jewelry
  - imageGen.taskCompleted → history.addNew(timestamp) 串起新生成图自动入历史
  - AppState 保留作 currentTab 等 UI 状态(task #13 后 loggedIn 字段删,QML 改用 auth.loggedIn)

冒烟测试:
  - 5 桥全 import OK,依赖注入构造 OK
  - PoC 模式 login('test','x') 通过 / login('','') 拒绝
  - history.refresh() 加载 2 条历史成功
  - jewelry.previewPrompt({...}) 正确组装中文 prompt
  - QML PoC 启动,登录页 + 主窗口(QML_AUTO_LOGIN=1)渲染完整无视觉回归

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
1 parent 12d6934e
"""QML 桥层。每个 Bridge 是一个 QObject,通过 setContextProperty 注入 QML。
桥层职责:
- 把 core/* 的 Python 业务通过 Property/Signal/Slot 暴露给 QML
- 数据类型转换(Path → str / bytes 走文件而非传值 / 中文字段名透传)
- 不写业务逻辑:所有真业务在 core/* + task_queue.py + audit_logger.py
依赖注入:业务 Manager 在 main_qml.py 实例化一次,作为参数传给桥构造函数。
"""
"""桥层共享的占位图标生成。
旧 ImageGeneratorWindow.create_placeholder_icon 是 QWidget 实例方法(依赖 self),
桥层独立用一份不依赖 widget 实例的版本。
"""
from PySide6.QtCore import Qt
from PySide6.QtGui import QFont, QIcon, QPainter, QPixmap
def build_placeholder_icon(text: str, size: int = 120) -> QIcon:
"""绘制一张 size×size 的浅灰底中央文字占位图,用于历史项缩略图缺失时回退。"""
pixmap = QPixmap(size, size)
pixmap.fill(Qt.lightGray)
painter = QPainter(pixmap)
painter.setPen(Qt.black)
painter.setFont(QFont("Arial", 10))
painter.drawText(pixmap.rect(), Qt.AlignCenter, text)
painter.end()
return QIcon(pixmap)
"""AuthBridge — 登录认证 + 当前用户。
替代旧 LoginDialog。QML LoginScreen 调 auth.login(user, pwd),桥内部走
DatabaseManager.authenticate(同步 MySQL,5s timeout)+ audit_logger.log_login。
PoC 模式(db_config = None):接受任意非空用户名密码,便于无 db 环境调 UI。
"""
import logging
import platform
import socket
from typing import Optional
from PySide6.QtCore import Property, QObject, Signal, Slot
from core.database import DatabaseManager
class AuthBridge(QObject):
loggedInChanged = Signal()
currentUserChanged = Signal()
loginFailed = Signal(str) # error_message
def __init__(self, db_config: Optional[dict] = None, audit_logger=None, parent=None):
super().__init__(parent)
self._logger = logging.getLogger(__name__)
self._db_config = db_config
self._db = DatabaseManager(db_config) if db_config else None
self._audit = audit_logger
self._logged_in = False
self._current_user = ""
@Property(bool, notify=loggedInChanged)
def loggedIn(self) -> bool:
return self._logged_in
@Property(str, notify=currentUserChanged)
def currentUser(self) -> str:
return self._current_user
@Slot(str, str, result=bool)
def login(self, username: str, password: str) -> bool:
username = (username or "").strip()
if not username or not password:
self.loginFailed.emit("用户名和密码不能为空")
return False
# PoC 模式:无 db_config 时接受任意非空
if self._db is None:
self._on_login_success(username)
return True
ok, msg = self._db.authenticate(username, password)
if not ok:
self._logger.warning(f"登录失败: {username} - {msg}")
self.loginFailed.emit(msg)
return False
self._on_login_success(username)
return True
@Slot()
def logout(self) -> None:
self._logged_in = False
self._current_user = ""
self.loggedInChanged.emit()
self.currentUserChanged.emit()
@Slot(result=str)
def deviceName(self) -> str:
"""供 audit 日志和 ImageGenBridge 使用"""
try:
return socket.gethostname() or platform.node() or "unknown"
except Exception:
return "unknown"
# ---- 内部 -----------------------------------------------------------
def _on_login_success(self, username: str) -> None:
self._current_user = username
self._logged_in = True
self.currentUserChanged.emit()
self.loggedInChanged.emit()
self._logger.info(f"登录成功: {username}")
if self._audit is not None:
try:
self._audit.log_login(
user_name=username,
local_ip=self._get_local_ip(),
public_ip=None, # public ip 走慢路径,task #13 再加
device_name=self.deviceName(),
)
except Exception as e:
self._logger.warning(f"audit log_login 失败(不影响登录): {e}")
@staticmethod
def _get_local_ip() -> Optional[str]:
try:
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
s.settimeout(0.5)
s.connect(("8.8.8.8", 80))
return s.getsockname()[0]
except Exception:
return None
"""HistoryBridge — 历史记录 tab 的列表 + 详情桥。
桥层暴露 HistoryListModel 给 QML ListView 直接用(已经是 QAbstractListModel)。
QML 通过 history.refresh() 主动刷新;新生成图片完成时由 ImageGenBridge 串起来调
itemAdded(task #16 wiring 时再补)。
详情查看走 getItem(timestamp) → dict,避免 QML 直接持有 HistoryItem dataclass。
"""
import logging
from typing import Any, Dict
from PySide6.QtCore import Property, QObject, Signal, Slot
from core.history import HistoryListModel
from ._icons import build_placeholder_icon
class HistoryBridge(QObject):
countChanged = Signal()
itemAdded = Signal(str) # timestamp
itemRemoved = Signal(str) # timestamp
def __init__(self, history_manager, parent=None):
super().__init__(parent)
self._logger = logging.getLogger(__name__)
self._history = history_manager
self._model = HistoryListModel(
history_manager=history_manager,
build_placeholder_icon=build_placeholder_icon,
logger=self._logger,
parent=self,
)
# ---- Properties -----------------------------------------------------
@Property(QObject, constant=True)
def model(self) -> HistoryListModel:
"""暴露给 QML ListView 的 model(QAbstractListModel)。"""
return self._model
@Property(int, notify=countChanged)
def count(self) -> int:
return self._model.rowCount()
# ---- Slots ----------------------------------------------------------
@Slot()
def refresh(self) -> None:
"""从磁盘重读 index.json,整体 reset model。"""
items = self._history.load_history_index()
self._model.reset_timestamps([item.timestamp for item in items])
self.countChanged.emit()
self._logger.info(f"history refresh: {len(items)} 条")
@Slot(str)
def deleteItem(self, timestamp: str) -> None:
ok = self._history.delete_history_item(timestamp)
if ok:
self._model.remove_timestamp(timestamp)
self.itemRemoved.emit(timestamp)
self.countChanged.emit()
@Slot(str)
def addNew(self, timestamp: str) -> None:
"""新生成完成时由 ImageGenBridge / 主线程调用,把新 timestamp 插到顶部。"""
self._model.prepend_timestamp(timestamp)
self.itemAdded.emit(timestamp)
self.countChanged.emit()
@Slot(str, result="QVariant")
def getItem(self, timestamp: str) -> Dict[str, Any]:
"""返回单条历史的 QML 友好 dict(路径 → str / Path 不暴露)。"""
item = self._history.load_history_item_fast(timestamp)
if item is None:
return {}
return {
"timestamp": item.timestamp,
"prompt": item.prompt,
"generatedImagePath": str(item.generated_image_path),
"referenceImagePaths": [str(p) for p in item.reference_image_paths],
"aspectRatio": item.aspect_ratio,
"imageSize": item.image_size,
"model": item.model,
"createdAt": item.created_at.strftime("%Y-%m-%d %H:%M:%S"),
}
"""ImageGenBridge — 图片生成 tab 的输入输出桥。
QML 调 imageGen.submitTask(prompt, refs, aspect, size, mode) → 桥层把 mode 中文
名翻译成 Gemini 模型 ID,调 TaskQueueManager.submit_task。任务完成后桥层
监听 TaskQueueManager.task_completed 信号,把 image_bytes 落 HistoryManager.save_generation,
然后把 result_path 通过 taskCompleted 信号转出(QML 只拿到文件路径,不传 bytes)。
"""
import logging
from typing import Optional
from PySide6.QtCore import Property, QObject, Signal, Slot
from core.generation import MODEL_BY_MODE, MODEL_PRO
class ImageGenBridge(QObject):
apiKeyChanged = Signal()
busyChanged = Signal()
taskSubmitted = Signal(str) # task_id
taskCompleted = Signal(str, str, str, str) # task_id, result_path, prompt, model
taskFailed = Signal(str, str) # task_id, error_message
taskProgress = Signal(str, float, str) # task_id, progress, status_text
def __init__(self, task_queue_manager, history_manager, auth_bridge,
api_key: str = "", parent=None):
super().__init__(parent)
self._logger = logging.getLogger(__name__)
self._tqm = task_queue_manager
self._history = history_manager
self._auth = auth_bridge
self._api_key = api_key
# 转发 TaskQueueManager 信号 → 桥层 QML 友好信号
self._tqm.task_added.connect(self._on_task_added)
self._tqm.task_progress.connect(self._on_progress)
self._tqm.task_completed.connect(self._on_completed)
self._tqm.task_failed.connect(self._on_failed)
# ---- Properties -----------------------------------------------------
@Property(str, notify=apiKeyChanged)
def apiKey(self) -> str:
return self._api_key
@Property(bool, notify=busyChanged)
def busy(self) -> bool:
return self._tqm.get_running_count() > 0
# ---- Slots ----------------------------------------------------------
@Slot(str)
def setApiKey(self, key: str) -> None:
if key != self._api_key:
self._api_key = key
self.apiKeyChanged.emit()
@Slot(str, list, str, str, str, result=str)
def submitTask(self, prompt: str, reference_images: list,
aspect_ratio: str, image_size: str, mode: str) -> str:
"""提交一条生成任务,返回 task_id。失败抛 RuntimeError。
Args:
prompt: 中文提示词
reference_images: 参考图本地路径 list[str]
aspect_ratio: '1:1' / '2:3' / ...
image_size: '1K' / '2K' / '4K'
mode: '极速模式' 或 '慢速模式'
"""
from task_queue import TaskType # 局部 import 避免桥层冷启动加载 Qt UI
model = MODEL_BY_MODE.get(mode, MODEL_PRO)
task_id = self._tqm.submit_task(
task_type=TaskType.IMAGE_GENERATION,
prompt=prompt,
api_key=self._api_key,
reference_images=list(reference_images or []),
aspect_ratio=aspect_ratio,
image_size=image_size,
model=model,
user_name=self._auth.currentUser if self._auth else "",
device_name=self._auth.deviceName() if self._auth else "",
)
self._logger.info(f"提交生成任务: {task_id[:8]} - mode={mode}")
return task_id
@Slot(str)
def cancelTask(self, task_id: str) -> None:
self._tqm.cancel_task(task_id)
self.busyChanged.emit()
# ---- 内部信号转发 ----------------------------------------------------
def _on_task_added(self, task) -> None:
self.taskSubmitted.emit(task.id)
self.busyChanged.emit()
def _on_progress(self, task_id: str, progress: float, status_text: str) -> None:
self.taskProgress.emit(task_id, progress, status_text)
def _on_completed(self, task_id: str, image_bytes: bytes, prompt: str,
reference_images: list, aspect_ratio: str,
image_size: str, model: str) -> None:
# 落历史记录(QML 只看路径)
try:
timestamp = self._history.save_generation(
image_bytes=image_bytes,
prompt=prompt,
reference_images=reference_images,
aspect_ratio=aspect_ratio,
image_size=image_size,
model=model,
)
result_path = str(self._history.base_path / timestamp / "generated.png")
except Exception as e:
self._logger.error(f"保存历史失败 {task_id[:8]}: {e}", exc_info=True)
self.taskFailed.emit(task_id, f"图片生成成功但保存历史失败: {e}")
self.busyChanged.emit()
return
self.taskCompleted.emit(task_id, result_path, prompt, model)
self.busyChanged.emit()
def _on_failed(self, task_id: str, error: str) -> None:
self.taskFailed.emit(task_id, error)
self.busyChanged.emit()
"""JewelryBridge — 款式设计 tab 桥层。
QML 需要的能力:
- 拿全部 9 个类别名(categories)
- 按类别拿候选词(getOptions)
- 增删词条 + 重置类别 / 全部
- 用户填完表后预览中文 prompt
后端:core.jewelry.JewelryLibraryManager + PromptAssembler。
"""
import logging
from typing import Any, Dict, List
from PySide6.QtCore import Property, QObject, Signal, Slot
from core.jewelry import PromptAssembler
class JewelryBridge(QObject):
libraryChanged = Signal(str) # category
def __init__(self, library_manager, parent=None):
super().__init__(parent)
self._logger = logging.getLogger(__name__)
self._lib = library_manager
@Property("QVariantList", constant=True)
def categories(self) -> List[str]:
"""9 个类别的中文名(顺序固定,与 QML 表单一致)。"""
return list(self._lib.library.keys())
@Slot(str, result="QVariantList")
def getOptions(self, category: str) -> List[str]:
return list(self._lib.library.get(category, []))
@Slot(str, str)
def addItem(self, category: str, value: str) -> None:
self._lib.add_item(category, value)
self.libraryChanged.emit(category)
@Slot(str, str)
def removeItem(self, category: str, value: str) -> None:
self._lib.remove_item(category, value)
self.libraryChanged.emit(category)
@Slot(str)
def resetCategory(self, category: str) -> None:
self._lib.reset_category(category)
self.libraryChanged.emit(category)
@Slot()
def resetAll(self) -> None:
self._lib.reset_all()
for c in self._lib.library.keys():
self.libraryChanged.emit(c)
@Slot("QVariantMap", result=str)
def previewPrompt(self, form_data: Dict[str, Any]) -> str:
"""form_data: { "主石形状": "圆形", "金属": "18K黄金", ... } —— 空字段允许。"""
return PromptAssembler.assemble({k: str(v or "") for k, v in form_data.items()})
"""TaskQueueBridge — 任务队列 sidebar 桥层。
QML sidebar ListView 直接用 taskQueue.model(QAbstractListModel),
桥层监听 TaskQueueManager 单例信号,按 task_id 增量更新 model。
只持有最近 N 条(与 TaskQueueManager 自身的 _max_history_size 一致),
更老的任务会随 TaskQueueManager._cleanup_old_tasks 自然消失。
"""
import logging
from datetime import datetime
from typing import Dict, List
from PySide6.QtCore import (
Property, QAbstractListModel, QModelIndex, QObject, Qt, Signal, Slot,
)
class _TaskListModel(QAbstractListModel):
"""暴露给 QML ListView 的任务列表模型。
Roles:taskId / prompt / status / progress / statusText / elapsed
"""
TaskIdRole = Qt.UserRole + 1
PromptRole = Qt.UserRole + 2
StatusRole = Qt.UserRole + 3
ProgressRole = Qt.UserRole + 4
StatusTextRole = Qt.UserRole + 5
ElapsedRole = Qt.UserRole + 6
def __init__(self, parent=None):
super().__init__(parent)
self._ids: List[str] = []
self._rows: Dict[str, dict] = {}
def rowCount(self, parent=QModelIndex()) -> int:
if parent.isValid():
return 0
return len(self._ids)
def roleNames(self):
return {
_TaskListModel.TaskIdRole: b"taskId",
_TaskListModel.PromptRole: b"prompt",
_TaskListModel.StatusRole: b"status",
_TaskListModel.ProgressRole: b"progress",
_TaskListModel.StatusTextRole: b"statusText",
_TaskListModel.ElapsedRole: b"elapsed",
}
def data(self, index: QModelIndex, role: int = Qt.DisplayRole):
if not index.isValid():
return None
row = index.row()
if row < 0 or row >= len(self._ids):
return None
record = self._rows.get(self._ids[row], {})
if role == _TaskListModel.TaskIdRole:
return record.get("task_id", "")
if role == _TaskListModel.PromptRole:
return record.get("prompt", "")
if role == _TaskListModel.StatusRole:
return record.get("status", "")
if role == _TaskListModel.ProgressRole:
return float(record.get("progress", 0.0))
if role == _TaskListModel.StatusTextRole:
return record.get("status_text", "")
if role == _TaskListModel.ElapsedRole:
return record.get("elapsed", "")
return None
# ---- 增量操作(桥层调用)---------------------------------------------
def upsert(self, task_id: str, **fields) -> None:
if task_id in self._rows:
self._rows[task_id].update(fields)
row = self._ids.index(task_id)
top = self.index(row, 0)
self.dataChanged.emit(top, top)
else:
self.beginInsertRows(QModelIndex(), 0, 0)
self._ids.insert(0, task_id)
self._rows[task_id] = {"task_id": task_id, **fields}
self.endInsertRows()
class TaskQueueBridge(QObject):
pendingCountChanged = Signal()
runningCountChanged = Signal()
def __init__(self, task_queue_manager, parent=None):
super().__init__(parent)
self._logger = logging.getLogger(__name__)
self._tqm = task_queue_manager
self._model = _TaskListModel(self)
self._tqm.task_added.connect(self._on_task_added)
self._tqm.task_started.connect(self._on_task_started)
self._tqm.task_completed.connect(self._on_task_completed)
self._tqm.task_failed.connect(self._on_task_failed)
self._tqm.task_progress.connect(self._on_progress)
# ---- Properties -----------------------------------------------------
@Property(QObject, constant=True)
def model(self):
return self._model
@Property(int, notify=pendingCountChanged)
def pendingCount(self) -> int:
return self._tqm.get_pending_count()
@Property(int, notify=runningCountChanged)
def runningCount(self) -> int:
return self._tqm.get_running_count()
# ---- Slots ----------------------------------------------------------
@Slot(str)
def cancelTask(self, task_id: str) -> None:
self._tqm.cancel_task(task_id)
self.pendingCountChanged.emit()
self.runningCountChanged.emit()
# ---- 信号转 model 增量 -----------------------------------------------
def _on_task_added(self, task) -> None:
self._model.upsert(
task.id,
prompt=task.prompt,
status="pending",
progress=0.0,
status_text="等待中",
elapsed="",
)
self.pendingCountChanged.emit()
def _on_task_started(self, task_id: str) -> None:
self._model.upsert(task_id, status="running", status_text="生成中…")
self.pendingCountChanged.emit()
self.runningCountChanged.emit()
def _on_progress(self, task_id: str, progress: float, status_text: str) -> None:
self._model.upsert(task_id, progress=progress, status_text=status_text)
def _on_task_completed(self, task_id, *_args) -> None:
elapsed = self._format_elapsed(task_id)
self._model.upsert(task_id, status="completed", progress=1.0,
status_text="已完成", elapsed=elapsed)
self.runningCountChanged.emit()
def _on_task_failed(self, task_id: str, error: str) -> None:
elapsed = self._format_elapsed(task_id)
self._model.upsert(task_id, status="failed", status_text=error or "失败",
elapsed=elapsed)
self.pendingCountChanged.emit()
self.runningCountChanged.emit()
def _format_elapsed(self, task_id: str) -> str:
task = self._tqm.get_task(task_id)
if task and task.started_at and task.completed_at:
secs = (task.completed_at - task.started_at).total_seconds()
return f"{secs:.1f}s"
return ""
"""QML PoC 入口 — 验证 QtQuick + Apple 风格自定义视觉
"""QML 主入口 — 装载业务桥层 + Apple 风格 QtQuick UI
只做 UI 演示:
- 登录页 (LoginScreen.qml)
- 主窗口 (MainWindow.qml) 含 3 tab + 任务队列 sidebar
- 图片生成 tab 的核心 UI 元素
启动序列:
1. 加载 config.json(API key / db_config / 历史最大数)
2. 实例化 core 业务(HistoryManager / JewelryLibraryManager / TaskQueueManager)
3. 实例化 audit_logger(落盘 + 后台 worker,task #18 全量启用)
4. 实例化 5 个桥(auth / imageGen / history / taskQueue / jewelry)+ AppState
5. 把它们注入 QQmlApplicationEngine context
6. load Main.qml
不接业务后端(不真的生成图片 / 调 db),让用户先看视觉。
跑:.venv/Scripts/pythonw.exe qml_poc/main_qml.py
"""
import logging
import os
import sys
from pathlib import Path
from PySide6.QtCore import QObject, QUrl, Property, Signal, Slot
from PySide6.QtCore import Property, QObject, QUrl, Signal, Slot
from PySide6.QtGui import QGuiApplication
from PySide6.QtQml import QQmlApplicationEngine
from PySide6.QtQuickControls2 import QQuickStyle
# 确保 import 路径包含项目根(qml_poc/ 跑 main_qml.py 时也能 import core/bridges)
ROOT = Path(__file__).resolve().parent.parent
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
from audit_logger import AuditLogger # noqa: E402
from bridges.auth import AuthBridge # noqa: E402
from bridges.history import HistoryBridge # noqa: E402
from bridges.imagegen import ImageGenBridge # noqa: E402
from bridges.jewelry import JewelryBridge # noqa: E402
from bridges.taskqueue import TaskQueueBridge # noqa: E402
from config_util import get_config_dir, get_config_path, load_config_safe # noqa: E402
from core.history import HistoryManager # noqa: E402
from core.jewelry import JewelryLibraryManager # noqa: E402
from core.paths import get_app_data_path # noqa: E402
from task_queue import TaskQueueManager # noqa: E402
logger = logging.getLogger(__name__)
class AppState(QObject):
"""暴露给 QML 的全局状态。最小骨架,业务后续再接。"""
"""全局轻量状态。登录态委托给 AuthBridge,本类只管当前 tab 索引等纯 UI 状态。
PoC 调试:QML_AUTO_LOGIN=1 时跳过登录直接进主窗口;task #13 后 QML 改用
auth.loggedIn 替代 appState.loggedIn,本类的 loggedIn 字段也会被删掉。
"""
loggedInChanged = Signal()
currentTabChanged = Signal()
def __init__(self):
def __init__(self, auth_bridge: AuthBridge):
super().__init__()
# PoC 调试:QML_AUTO_LOGIN=1 时直接登录态,免每次重启进程都发回车登录
import os
self._logged_in = os.environ.get("QML_AUTO_LOGIN", "") == "1"
self._auth = auth_bridge
# 兼容现有 QML:env QML_AUTO_LOGIN=1 时强制 loggedIn=True
self._poc_force_login = os.environ.get("QML_AUTO_LOGIN", "") == "1"
self._current_tab = 0
self._auth.loggedInChanged.connect(self.loggedInChanged.emit)
@Property(bool, notify=loggedInChanged)
def loggedIn(self):
return self._logged_in
def loggedIn(self) -> bool:
return self._poc_force_login or self._auth.loggedIn
@Property(int, notify=currentTabChanged)
def currentTab(self):
def currentTab(self) -> int:
return self._current_tab
@Slot(str, str, result=bool)
def login(self, username: str, password: str) -> bool:
# PoC:随便接受,业务不接
if not username or not password:
return False
self._logged_in = True
self.loggedInChanged.emit()
return True
"""兼容旧 QML PoC:转发到 AuthBridge。task #13 后 QML 直接调 auth.login。"""
return self._auth.login(username, password)
@Slot()
def logout(self):
self._logged_in = False
self.loggedInChanged.emit()
def logout(self) -> None:
self._auth.logout()
@Slot(int)
def setTab(self, idx: int):
def setTab(self, idx: int) -> None:
self._current_tab = idx
self.currentTabChanged.emit()
def _setup_logging():
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
)
def main():
QQuickStyle.setStyle("Basic") # 纯净底,不引入 Material/Win11 的视觉污染
_setup_logging()
QQuickStyle.setStyle("Basic")
app = QGuiApplication(sys.argv)
app.setApplicationName("珠宝壹佰图像生成器")
app.setOrganizationName("ZB100")
state = AppState()
# ---- 配置 + 业务核心 ------------------------------------------------
config_path = get_config_path()
config, err = load_config_safe(config_path)
if err:
logger.warning(f"config 加载警告: {err}")
api_key = config.get("api_key", "") or ""
db_config = config.get("db_config") # None 时桥层走 PoC 模式
config_dir = get_config_dir()
history_manager = HistoryManager()
jewelry_manager = JewelryLibraryManager(config_dir)
task_queue_manager = TaskQueueManager() # 单例
audit_logger = None
if db_config:
try:
app_data = get_app_data_path()
audit_logger = AuditLogger(
db_config=db_config,
queue_path=app_data / "audit_queue.ndjson",
logs_dir=app_data / "logs",
)
audit_logger.start()
except Exception as e:
logger.warning(f"audit_logger 启动失败(不影响 UI): {e}")
audit_logger = None
# ---- 桥层 ----------------------------------------------------------
auth_bridge = AuthBridge(db_config=db_config, audit_logger=audit_logger)
image_gen_bridge = ImageGenBridge(
task_queue_manager=task_queue_manager,
history_manager=history_manager,
auth_bridge=auth_bridge,
api_key=api_key,
)
history_bridge = HistoryBridge(history_manager=history_manager)
task_queue_bridge = TaskQueueBridge(task_queue_manager=task_queue_manager)
jewelry_bridge = JewelryBridge(library_manager=jewelry_manager)
app_state = AppState(auth_bridge=auth_bridge)
# 新生成完图后让 history 列表加新行(绕过手动 refresh)
image_gen_bridge.taskCompleted.connect(
lambda task_id, result_path, prompt, model:
history_bridge.addNew(Path(result_path).parent.name)
)
# 启动期预填一次历史
history_bridge.refresh()
# ---- QML 装载 ------------------------------------------------------
engine = QQmlApplicationEngine()
engine.rootContext().setContextProperty("appState", state)
ctx = engine.rootContext()
ctx.setContextProperty("appState", app_state)
ctx.setContextProperty("auth", auth_bridge)
ctx.setContextProperty("imageGen", image_gen_bridge)
ctx.setContextProperty("history", history_bridge)
ctx.setContextProperty("taskQueue", task_queue_bridge)
ctx.setContextProperty("jewelry", jewelry_bridge)
qml_dir = Path(__file__).parent / "qml"
engine.addImportPath(str(qml_dir))
......@@ -75,7 +163,16 @@ def main():
print("QML load failed", file=sys.stderr)
sys.exit(1)
sys.exit(app.exec())
rc = app.exec()
# 退出前 flush audit 日志
if audit_logger:
try:
audit_logger.shutdown(timeout=5.0)
except Exception:
pass
sys.exit(rc)
if __name__ == "__main__":
......