247 lines
8.8 KiB
Python
247 lines
8.8 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 import config
|
|
from app.models import Order
|
|
from app.schedule import get_cutoff_time, is_past_cutoff
|
|
from app.status_rules import parse_status_list, status_in
|
|
from app.tracking import get_fulfilled_statuses
|
|
|
|
CANCELLED_ROW_COLOR = QColor(255, 210, 210)
|
|
FULFILLED_ROW_COLOR = QColor(210, 240, 210)
|
|
|
|
COLUMNS = [
|
|
("company", "Company"),
|
|
("ticket_number", "Ticket #"),
|
|
("skus_display", "SKUs"),
|
|
("summary", "Summary"),
|
|
("status", "Status"),
|
|
("tracking_display", "Tracking #"),
|
|
("created_display", "Created"),
|
|
("past_cutoff_display", "Past Cutoff"),
|
|
]
|
|
|
|
ALL_COMPANIES = "All Companies"
|
|
|
|
|
|
def _format_tracking(tracking_numbers) -> str:
|
|
if not tracking_numbers:
|
|
return ""
|
|
return ", ".join(
|
|
f"{t.get('number')} ({'return' if t.get('is_return') else 'out'})"
|
|
for t in tracking_numbers
|
|
)
|
|
|
|
|
|
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
|
|
|
|
def set_orders(self, orders: List[Order]) -> None:
|
|
# Re-read status lists and cutoff time each refresh, in case Settings changed.
|
|
self._cancelled_statuses = parse_status_list(
|
|
config.get("CANCELLED_STATUSES", config.DEFAULT_CANCELLED_STATUSES)
|
|
)
|
|
self._fulfilled_statuses = get_fulfilled_statuses()
|
|
self._cutoff_time = get_cutoff_time()
|
|
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()]
|
|
|
|
if role == Qt.ItemDataRole.BackgroundRole:
|
|
if status_in(order.status, self._cancelled_statuses):
|
|
return CANCELLED_ROW_COLOR
|
|
if status_in(order.status, self._fulfilled_statuses):
|
|
return FULFILLED_ROW_COLOR
|
|
return None
|
|
|
|
if role != Qt.ItemDataRole.DisplayRole:
|
|
return None
|
|
|
|
field_name, _ = COLUMNS[index.column()]
|
|
|
|
if field_name == "skus_display":
|
|
return ", ".join(order.skus or [])
|
|
if field_name == "tracking_display":
|
|
return _format_tracking(order.tracking_numbers)
|
|
if field_name == "created_display":
|
|
return _format_created(order.source_created_at)
|
|
if field_name == "past_cutoff_display":
|
|
return "Yes" if is_past_cutoff(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:
|
|
haystack = " ".join(
|
|
[
|
|
order.ticket_number or "",
|
|
order.summary or "",
|
|
order.status or "",
|
|
" ".join(order.skus or []),
|
|
_format_tracking(order.tracking_numbers),
|
|
]
|
|
).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)
|
|
self.table.horizontalHeader().setSectionResizeMode(QHeaderView.ResizeMode.Stretch)
|
|
self.table.horizontalHeader().setStretchLastSection(False)
|
|
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 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
|