96 lines
3.9 KiB
Python
96 lines
3.9 KiB
Python
|
"""采购记录的只读 Qt Model/View 数据源。"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from dataclasses import dataclass, field
|
||
|
|
from typing import Protocol
|
||
|
|
|
||
|
|
from PySide6.QtCore import QAbstractTableModel, QModelIndex, Qt
|
||
|
|
|
||
|
|
from cmbuyer_client.logging_policy import redact_text
|
||
|
|
from cmbuyer_client.core.errors import ValidationError
|
||
|
|
from cmbuyer_client.core.validation import rfc3339_z_nanoseconds
|
||
|
|
|
||
|
|
|
||
|
|
@dataclass(frozen=True)
|
||
|
|
class PurchaseRecord:
|
||
|
|
record_id: str
|
||
|
|
title: str
|
||
|
|
status: str
|
||
|
|
created_at: str
|
||
|
|
original_text: str = ""
|
||
|
|
result_text: str = ""
|
||
|
|
image_description: str = ""
|
||
|
|
created_at_nanoseconds: int = field(init=False, repr=False)
|
||
|
|
|
||
|
|
def __post_init__(self) -> None:
|
||
|
|
# 记录 provider 只能注入可显示摘要；最终 UI 边界仍统一脱敏，避免
|
||
|
|
# consumer bug 把 Bearer/裸 token 放入 model、详情或可见日志。
|
||
|
|
for field in ("title", "status", "original_text", "result_text", "image_description"):
|
||
|
|
object.__setattr__(self, field, redact_text(getattr(self, field)))
|
||
|
|
if not isinstance(self.created_at, str):
|
||
|
|
raise ValueError("noncanonical_record_timestamp")
|
||
|
|
fraction = self.created_at[20:-1] if len(self.created_at) > 20 and self.created_at.endswith("Z") else ""
|
||
|
|
if fraction and fraction.endswith("0"):
|
||
|
|
raise ValueError("noncanonical_record_timestamp")
|
||
|
|
try:
|
||
|
|
timestamp = rfc3339_z_nanoseconds(self.created_at)
|
||
|
|
except ValidationError:
|
||
|
|
raise ValueError("noncanonical_record_timestamp") from None
|
||
|
|
object.__setattr__(self, "created_at_nanoseconds", timestamp)
|
||
|
|
|
||
|
|
|
||
|
|
class PurchaseRecordProvider(Protocol):
|
||
|
|
"""只返回已准备好的无秘密 View DTO；不得在 GUI 线程查询 SQLite/HTTP。"""
|
||
|
|
|
||
|
|
def snapshot(self) -> list[PurchaseRecord]: ...
|
||
|
|
|
||
|
|
|
||
|
|
class PurchaseRecordModel(QAbstractTableModel):
|
||
|
|
RECORD_ID_ROLE = int(Qt.ItemDataRole.UserRole) + 1
|
||
|
|
|
||
|
|
def __init__(self, records: list[PurchaseRecord] | None = None, parent=None) -> None:
|
||
|
|
super().__init__(parent)
|
||
|
|
self._records: list[PurchaseRecord] = []
|
||
|
|
self.set_records(records or [])
|
||
|
|
|
||
|
|
def rowCount(self, parent: QModelIndex = QModelIndex()) -> int:
|
||
|
|
return 0 if parent.isValid() else len(self._records)
|
||
|
|
|
||
|
|
def columnCount(self, parent: QModelIndex = QModelIndex()) -> int:
|
||
|
|
return 0 if parent.isValid() else 2
|
||
|
|
|
||
|
|
def data(self, index: QModelIndex, role: int = int(Qt.ItemDataRole.DisplayRole)):
|
||
|
|
if not index.isValid() or not 0 <= index.row() < len(self._records):
|
||
|
|
return None
|
||
|
|
record = self._records[index.row()]
|
||
|
|
if role == int(Qt.ItemDataRole.DisplayRole):
|
||
|
|
return record.title if index.column() == 0 else record.status
|
||
|
|
if role == self.RECORD_ID_ROLE:
|
||
|
|
return record.record_id
|
||
|
|
if role == int(Qt.ItemDataRole.ToolTipRole):
|
||
|
|
return f"{record.title}\n{record.created_at}"
|
||
|
|
if role == int(Qt.ItemDataRole.TextAlignmentRole) and index.column() == 1:
|
||
|
|
return int(Qt.AlignmentFlag.AlignCenter)
|
||
|
|
return None
|
||
|
|
|
||
|
|
def headerData(self, section: int, orientation: Qt.Orientation, role: int = int(Qt.ItemDataRole.DisplayRole)):
|
||
|
|
if role != int(Qt.ItemDataRole.DisplayRole) or orientation != Qt.Orientation.Horizontal:
|
||
|
|
return None
|
||
|
|
return ("标题", "状态")[section] if 0 <= section < 2 else None
|
||
|
|
|
||
|
|
def set_records(self, records: list[PurchaseRecord]) -> None:
|
||
|
|
self.beginResetModel()
|
||
|
|
self._records = sorted(
|
||
|
|
records,
|
||
|
|
key=lambda item: (item.created_at_nanoseconds, item.record_id),
|
||
|
|
reverse=True,
|
||
|
|
)
|
||
|
|
self.endResetModel()
|
||
|
|
|
||
|
|
def record_at(self, row: int) -> PurchaseRecord | None:
|
||
|
|
return self._records[row] if 0 <= row < len(self._records) else None
|
||
|
|
|
||
|
|
def row_for_id(self, record_id: str) -> int:
|
||
|
|
return next((row for row, item in enumerate(self._records) if item.record_id == record_id), -1)
|