324 lines
13 KiB
Python
324 lines
13 KiB
Python
"""
|
|
Table view + model for displaying orders.
|
|
|
|
Using QAbstractTableModel instead of QTableWidget on purpose: it scales
|
|
to thousands of rows, and features like sorting and the company filter
|
|
below are straightforward extensions of this model rather than
|
|
rewrites. There's no "Source" filter - every row is a JIRA ticket;
|
|
ShipStation only enriches rows with tracking numbers, it doesn't add
|
|
its own.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from typing import List, Any, Optional
|
|
|
|
from PyQt6.QtCore import Qt, QAbstractTableModel, QModelIndex, QSortFilterProxyModel, pyqtSignal
|
|
from PyQt6.QtGui import QColor
|
|
from PyQt6.QtWidgets import (
|
|
QTableView,
|
|
QAbstractItemView,
|
|
QHeaderView,
|
|
QWidget,
|
|
QVBoxLayout,
|
|
QHBoxLayout,
|
|
QComboBox,
|
|
QLineEdit,
|
|
QLabel,
|
|
)
|
|
|
|
from app.models import Order
|
|
from app.return_labels import is_emailed_label_order, get_emailed_label_skus
|
|
from app.schedule import get_cutoff_time, is_past_cutoff_today
|
|
from app.status_rules import get_cancelled_statuses, status_in
|
|
from app.ticket_validation import validate_ticket, make_validation_context
|
|
from app.tracking import get_fulfilled_statuses
|
|
|
|
CANCELLED_TEXT_COLOR = QColor(180, 0, 0)
|
|
FULFILLED_ROW_COLOR = QColor(210, 240, 210)
|
|
ISSUE_TEXT_COLOR = QColor(170, 100, 0)
|
|
CHECK_MARK = "\u2713"
|
|
WARNING_MARK = "\u26a0"
|
|
|
|
# Order here is display order, left to right.
|
|
COLUMNS = [
|
|
("company", "Company"),
|
|
("ticket_number", "Ticket #"),
|
|
("status", "Status"),
|
|
("issues_display", "Issues"),
|
|
("packed_display", "Done"),
|
|
("skus_display", "SKUs"),
|
|
("summary", "Summary"),
|
|
("assignee", "Assignee"),
|
|
("serials_display", "Serial #s"),
|
|
("return_label_display", "Return Label"),
|
|
("outgoing_tracking_display", "Outgoing Tracking"),
|
|
("return_tracking_display", "Return Tracking"),
|
|
("shipping_method", "Shipping Method"),
|
|
("uploaded_display", "Uploaded"),
|
|
("past_cutoff_display", "Past Cutoff"),
|
|
("created_display", "Created"),
|
|
]
|
|
|
|
ALL_COMPANIES = "All Companies"
|
|
|
|
|
|
def _format_tracking_numbers(tracking_numbers, is_return: bool) -> str:
|
|
if not tracking_numbers:
|
|
return ""
|
|
return ", ".join(
|
|
t.get("number", "") for t in tracking_numbers if bool(t.get("is_return")) == is_return
|
|
)
|
|
|
|
|
|
def _format_created(source_created_at) -> str:
|
|
if not source_created_at:
|
|
return ""
|
|
return source_created_at.strftime("%b %d, %I:%M %p").replace(" 0", " ")
|
|
|
|
|
|
class OrdersTableModel(QAbstractTableModel):
|
|
def __init__(self, orders: List[Order] | None = None, parent=None):
|
|
super().__init__(parent)
|
|
self._orders: List[Order] = orders or []
|
|
self._cancelled_statuses: set[str] = set()
|
|
self._fulfilled_statuses: set[str] = set()
|
|
self._cutoff_time = None
|
|
self._issues_by_id: dict[int, list] = {}
|
|
self._is_emailed_label_by_id: dict[int, bool] = {}
|
|
|
|
def set_orders(self, orders: List[Order]) -> None:
|
|
# Re-read status lists and cutoff time each refresh, in case Settings changed.
|
|
self._cancelled_statuses = get_cancelled_statuses()
|
|
self._fulfilled_statuses = get_fulfilled_statuses()
|
|
self._cutoff_time = get_cutoff_time()
|
|
# Computed ONCE per refresh here, not per cell paint - both of
|
|
# these read Settings from disk internally, and Qt calls data()
|
|
# extremely frequently (every cell, every repaint, constantly
|
|
# during scrolling) - doing this per-cell instead of per-refresh
|
|
# was a severe, real performance bug, not just a minor slowdown.
|
|
# validation_context is ALSO fetched once here rather than once
|
|
# per order inside validate_ticket() - same fix, applied to the
|
|
# cost of computing the cache itself, not just using it.
|
|
validation_context = make_validation_context()
|
|
self._issues_by_id = {
|
|
order.id: validate_ticket(order.skus, order.line_items, **validation_context)
|
|
for order in orders
|
|
}
|
|
emailed_skus = get_emailed_label_skus()
|
|
self._is_emailed_label_by_id = {
|
|
order.id: is_emailed_label_order(order.skus or [], emailed_skus) for order in orders
|
|
}
|
|
self.beginResetModel()
|
|
self._orders = orders
|
|
self.endResetModel()
|
|
|
|
def rowCount(self, parent: QModelIndex = QModelIndex()) -> int:
|
|
return len(self._orders)
|
|
|
|
def columnCount(self, parent: QModelIndex = QModelIndex()) -> int:
|
|
return len(COLUMNS)
|
|
|
|
def headerData(self, section, orientation, role=Qt.ItemDataRole.DisplayRole):
|
|
if role != Qt.ItemDataRole.DisplayRole:
|
|
return None
|
|
if orientation == Qt.Orientation.Horizontal:
|
|
return COLUMNS[section][1]
|
|
return str(section + 1)
|
|
|
|
def data(self, index: QModelIndex, role: int = Qt.ItemDataRole.DisplayRole) -> Any:
|
|
if not index.isValid():
|
|
return None
|
|
|
|
order = self._orders[index.row()]
|
|
is_cancelled = status_in(order.status, self._cancelled_statuses)
|
|
field_name, _ = COLUMNS[index.column()]
|
|
|
|
if role == Qt.ItemDataRole.ForegroundRole:
|
|
if is_cancelled:
|
|
return CANCELLED_TEXT_COLOR
|
|
if field_name == "issues_display" and self._issues_by_id.get(order.id):
|
|
return ISSUE_TEXT_COLOR
|
|
return None
|
|
|
|
if role == Qt.ItemDataRole.BackgroundRole:
|
|
# Cancelled rows use red TEXT (above) instead of a background,
|
|
# so it stays readable regardless of theme. Fulfilled rows
|
|
# still get a background tint since there's no text-color
|
|
# signal already in use for them.
|
|
if not is_cancelled and status_in(order.status, self._fulfilled_statuses):
|
|
return FULFILLED_ROW_COLOR
|
|
return None
|
|
|
|
if role == Qt.ItemDataRole.ToolTipRole and field_name == "issues_display":
|
|
issues = self._issues_by_id.get(order.id, [])
|
|
return "\n".join(i.message for i in issues) if issues else None
|
|
|
|
if role != Qt.ItemDataRole.DisplayRole:
|
|
return None
|
|
|
|
if field_name == "issues_display":
|
|
issues = self._issues_by_id.get(order.id, [])
|
|
return f"{WARNING_MARK} ({len(issues)})" if issues else ""
|
|
if field_name == "skus_display":
|
|
return ", ".join(order.skus or [])
|
|
if field_name == "packed_display":
|
|
return CHECK_MARK if order.packed else ""
|
|
if field_name == "serials_display":
|
|
serials = order.serial_numbers or {}
|
|
filled = sum(1 for v in serials.values() if (v or "").strip())
|
|
return f"{filled} entered" if filled else ""
|
|
if field_name == "return_label_display":
|
|
return CHECK_MARK if self._is_emailed_label_by_id.get(order.id) else ""
|
|
if field_name == "outgoing_tracking_display":
|
|
return _format_tracking_numbers(order.tracking_numbers, is_return=False)
|
|
if field_name == "return_tracking_display":
|
|
return _format_tracking_numbers(order.tracking_numbers, is_return=True)
|
|
if field_name == "uploaded_display":
|
|
return CHECK_MARK if order.shipstation_sent_at else ""
|
|
if field_name == "created_display":
|
|
return _format_created(order.source_created_at)
|
|
if field_name == "past_cutoff_display":
|
|
return CHECK_MARK if is_past_cutoff_today(order.source_created_at, self._cutoff_time) else ""
|
|
|
|
value = getattr(order, field_name)
|
|
return "" if value is None else str(value)
|
|
|
|
def order_at(self, row: int) -> Order:
|
|
return self._orders[row]
|
|
|
|
|
|
class OrdersFilterProxyModel(QSortFilterProxyModel):
|
|
"""Filters by company and a free-text search across ticket/SKU/summary/tracking."""
|
|
|
|
def __init__(self, parent=None):
|
|
super().__init__(parent)
|
|
self.company_filter: str = ALL_COMPANIES
|
|
self.search_text: str = ""
|
|
|
|
def set_company_filter(self, company: str) -> None:
|
|
self.company_filter = company
|
|
self.invalidateFilter()
|
|
|
|
def set_search_text(self, text: str) -> None:
|
|
self.search_text = text.strip().lower()
|
|
self.invalidateFilter()
|
|
|
|
def filterAcceptsRow(self, source_row: int, source_parent: QModelIndex) -> bool:
|
|
model: OrdersTableModel = self.sourceModel()
|
|
order = model.order_at(source_row)
|
|
|
|
if self.company_filter != ALL_COMPANIES and order.company != self.company_filter:
|
|
return False
|
|
|
|
if self.search_text:
|
|
tracking_text = " ".join(
|
|
t.get("number", "") for t in (order.tracking_numbers or [])
|
|
)
|
|
haystack = " ".join(
|
|
[
|
|
order.ticket_number or "",
|
|
order.summary or "",
|
|
order.status or "",
|
|
" ".join(order.skus or []),
|
|
tracking_text,
|
|
]
|
|
).lower()
|
|
if self.search_text not in haystack:
|
|
return False
|
|
|
|
return True
|
|
|
|
|
|
class OrdersTableView(QWidget):
|
|
"""Filter bar + sortable, read-only, full-row-selection table."""
|
|
|
|
order_double_clicked = pyqtSignal(object) # emits the Order that was double-clicked
|
|
|
|
def __init__(self, parent=None):
|
|
super().__init__(parent)
|
|
|
|
self._source_model = OrdersTableModel()
|
|
self._proxy_model = OrdersFilterProxyModel()
|
|
self._proxy_model.setSourceModel(self._source_model)
|
|
|
|
layout = QVBoxLayout(self)
|
|
layout.setContentsMargins(0, 0, 0, 0)
|
|
|
|
filter_bar = QHBoxLayout()
|
|
filter_bar.addWidget(QLabel("Company:"))
|
|
self.company_combo = QComboBox()
|
|
self.company_combo.addItem(ALL_COMPANIES)
|
|
self.company_combo.currentTextChanged.connect(self._proxy_model.set_company_filter)
|
|
filter_bar.addWidget(self.company_combo)
|
|
|
|
filter_bar.addWidget(QLabel("Search:"))
|
|
self.search_box = QLineEdit()
|
|
self.search_box.setPlaceholderText("Ticket #, SKU, status, or tracking #...")
|
|
self.search_box.textChanged.connect(self._proxy_model.set_search_text)
|
|
filter_bar.addWidget(self.search_box, stretch=1)
|
|
|
|
layout.addLayout(filter_bar)
|
|
|
|
self.table = QTableView()
|
|
self.table.setModel(self._proxy_model)
|
|
self.table.setSelectionBehavior(QAbstractItemView.SelectionBehavior.SelectRows)
|
|
self.table.setEditTriggers(QAbstractItemView.EditTrigger.NoEditTriggers)
|
|
self.table.setSortingEnabled(True)
|
|
# Interactive (not Stretch) so columns size to their own content and
|
|
# stay individually resizable by dragging, rather than every column
|
|
# being force-stretched to equal widths regardless of content.
|
|
self.table.horizontalHeader().setSectionResizeMode(QHeaderView.ResizeMode.Interactive)
|
|
self.table.horizontalHeader().setStretchLastSection(True)
|
|
self.table.doubleClicked.connect(self._on_row_double_clicked)
|
|
layout.addWidget(self.table)
|
|
|
|
def _on_row_double_clicked(self, index) -> None:
|
|
source_index = self._proxy_model.mapToSource(index)
|
|
order = self._source_model.order_at(source_index.row())
|
|
self.order_double_clicked.emit(order)
|
|
|
|
def set_orders(self, orders: List[Order]) -> None:
|
|
self._source_model.set_orders(orders)
|
|
self.table.resizeColumnsToContents()
|
|
self._refresh_filter_options(orders)
|
|
|
|
def _refresh_filter_options(self, orders: List[Order]) -> None:
|
|
current = self.company_combo.currentText()
|
|
values = sorted({o.company for o in orders if o.company})
|
|
self.company_combo.blockSignals(True)
|
|
self.company_combo.clear()
|
|
self.company_combo.addItem(ALL_COMPANIES)
|
|
self.company_combo.addItems(values)
|
|
restore_index = self.company_combo.findText(current)
|
|
self.company_combo.setCurrentIndex(restore_index if restore_index >= 0 else 0)
|
|
self.company_combo.blockSignals(False)
|
|
|
|
def selected_order(self) -> Optional[Order]:
|
|
indexes = self.table.selectionModel().selectedRows()
|
|
if not indexes:
|
|
return None
|
|
source_index = self._proxy_model.mapToSource(indexes[0])
|
|
return self._source_model.order_at(source_index.row())
|
|
|
|
def select_ticket(self, ticket_number: str) -> bool:
|
|
"""Re-selects a row by ticket number - a table refresh (set_orders)
|
|
clears whatever was selected, since the underlying model resets.
|
|
Used after a step in a multi-step action (like the return-label
|
|
dummy-then-return flow) so the next step doesn't silently find
|
|
nothing selected. Returns whether the ticket was found/selected."""
|
|
for row in range(self._proxy_model.rowCount()):
|
|
proxy_index = self._proxy_model.index(row, 0)
|
|
source_index = self._proxy_model.mapToSource(proxy_index)
|
|
if self._source_model.order_at(source_index.row()).ticket_number == ticket_number:
|
|
self.table.selectRow(proxy_index.row())
|
|
return True
|
|
return False
|
|
|
|
def visible_orders(self) -> List[Order]:
|
|
"""Orders currently passing the active filters - used for export."""
|
|
result = []
|
|
for row in range(self._proxy_model.rowCount()):
|
|
source_index = self._proxy_model.mapToSource(self._proxy_model.index(row, 0))
|
|
result.append(self._source_model.order_at(source_index.row()))
|
|
return result
|