303 lines
9.3 KiB
Python
303 lines
9.3 KiB
Python
"""Order Query Tool — query Azure Table Storage orders by userId via TUI."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import tomllib
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
from azure.data.tables import TableServiceClient
|
|
from textual.app import App, ComposeResult
|
|
from textual.containers import Horizontal, Vertical
|
|
from textual.screen import ModalScreen
|
|
from textual.widgets import Button, DataTable, Input, Static
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Config
|
|
# ---------------------------------------------------------------------------
|
|
|
|
CONFIG_DIR = Path.home() / ".order-query"
|
|
CONFIG_FILE = CONFIG_DIR / "config.toml"
|
|
|
|
|
|
@dataclass
|
|
class Config:
|
|
connection_string: str
|
|
table_name: str = "order"
|
|
|
|
|
|
def load_config() -> Config:
|
|
if not CONFIG_FILE.exists():
|
|
CONFIG_DIR.mkdir(parents=True, exist_ok=True)
|
|
CONFIG_FILE.write_text(
|
|
"# Order Query Tool Configuration\n"
|
|
'connection_string = "DefaultEndpointsProtocol=https;AccountName=...;AccountKey=...;TableEndpoint=https://..."\n'
|
|
"table_name = \"order\"\n"
|
|
)
|
|
raise SystemExit(
|
|
f"Config file created at {CONFIG_FILE}\n"
|
|
"Please edit it with your Azure Storage connection string and run again."
|
|
)
|
|
with open(CONFIG_FILE, "rb") as f:
|
|
data = tomllib.load(f)
|
|
return Config(
|
|
connection_string=data["connection_string"],
|
|
table_name=data.get("table_name", "order"),
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Order model
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@dataclass
|
|
class Order:
|
|
order_id: str
|
|
prod_id: str = ""
|
|
prod_count: int = 0
|
|
prod_price: float = 0.0
|
|
revenue: float = 0.0
|
|
state: int = 0
|
|
order_time: str = ""
|
|
receive_time: str = ""
|
|
platform_order: str = ""
|
|
custom_data: str = ""
|
|
ver: int = 0
|
|
|
|
@property
|
|
def state_text(self) -> str:
|
|
return "Payed" if self.state == 0 else "Recived"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Azure Table Storage client
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def query_orders(config: Config, user_id: str) -> list[Order]:
|
|
client = TableServiceClient.from_connection_string(config.connection_string)
|
|
table = client.get_table_client(config.table_name)
|
|
entities = table.query_entities(
|
|
query_filter=f"PartitionKey eq '{user_id}'"
|
|
)
|
|
orders: list[Order] = []
|
|
for entity in entities:
|
|
orders.append(
|
|
Order(
|
|
order_id=entity.get("RowKey", ""),
|
|
prod_id=entity.get("prodId", ""),
|
|
prod_count=entity.get("prodCount", 0) or 0,
|
|
prod_price=entity.get("prodPrice", 0.0) or 0.0,
|
|
revenue=entity.get("revenue", 0.0) or 0.0,
|
|
state=entity.get("state", 0),
|
|
order_time=str(entity.get("OrderTime", "")),
|
|
receive_time=str(entity.get("ReceiveTime", "")),
|
|
platform_order=entity.get("platformOrder", ""),
|
|
custom_data=entity.get("customData", ""),
|
|
ver=entity.get("ver", 0) or 0,
|
|
)
|
|
)
|
|
return orders
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# TUI Screens
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class InputScreen(ModalScreen[str]):
|
|
"""Screen for entering userId."""
|
|
|
|
def compose(self) -> ComposeResult:
|
|
yield Vertical(
|
|
Static("Enter User ID:", classes="label"),
|
|
Input(placeholder="userId", id="user_id_input"),
|
|
Horizontal(
|
|
Button("Query", variant="primary", id="query_btn"),
|
|
Button("Quit", variant="default", id="quit_btn"),
|
|
classes="buttons",
|
|
),
|
|
classes="input-container",
|
|
)
|
|
|
|
def on_button_pressed(self, event: Button.Pressed) -> None:
|
|
if event.button.id == "query_btn":
|
|
user_id = self.query_one("#user_id_input", Input).value.strip()
|
|
if user_id:
|
|
self.dismiss(user_id)
|
|
elif event.button.id == "quit_btn":
|
|
self.app.exit()
|
|
|
|
def on_input_submitted(self, event: Input.Submitted) -> None:
|
|
user_id = event.value.strip()
|
|
if user_id:
|
|
self.dismiss(user_id)
|
|
|
|
|
|
class OrderListScreen(ModalScreen[str | None]):
|
|
"""Screen showing order list as a DataTable."""
|
|
|
|
def __init__(self, orders: list[Order]) -> None:
|
|
super().__init__()
|
|
self.orders = orders
|
|
|
|
def compose(self) -> ComposeResult:
|
|
yield Vertical(
|
|
Static(f"Found {len(self.orders)} orders. Select one for details."),
|
|
DataTable(id="order_table"),
|
|
Button("Back", id="back_btn"),
|
|
classes="list-container",
|
|
)
|
|
|
|
def on_mount(self) -> None:
|
|
table = self.query_one("#order_table", DataTable)
|
|
table.add_columns(
|
|
"Order ID",
|
|
"Product ID",
|
|
"Count",
|
|
"Price",
|
|
"Revenue",
|
|
"State",
|
|
"Order Time",
|
|
)
|
|
for order in self.orders:
|
|
table.add_row(
|
|
order.order_id,
|
|
order.prod_id,
|
|
str(order.prod_count),
|
|
f"{order.prod_price:.2f}",
|
|
f"{order.revenue:.2f}",
|
|
order.state_text,
|
|
order.order_time[:19] if order.order_time else "",
|
|
key=order.order_id,
|
|
)
|
|
|
|
def on_data_table_row_selected(self, event: DataTable.RowSelected) -> None:
|
|
order_id = event.row_key.value
|
|
self.dismiss(order_id)
|
|
|
|
def on_button_pressed(self, event: Button.Pressed) -> None:
|
|
if event.button.id == "back_btn":
|
|
self.dismiss(None)
|
|
|
|
|
|
class OrderDetailScreen(ModalScreen[None]):
|
|
"""Screen showing full order details."""
|
|
|
|
def __init__(self, order: Order) -> None:
|
|
super().__init__()
|
|
self.order = order
|
|
|
|
def compose(self) -> ComposeResult:
|
|
o = self.order
|
|
detail_text = (
|
|
f"[bold]Order ID:[/bold] {o.order_id}\n"
|
|
f"[bold]Product ID:[/bold] {o.prod_id}\n"
|
|
f"[bold]Product Count:[/bold] {o.prod_count}\n"
|
|
f"[bold]Product Price:[/bold] {o.prod_price:.2f}\n"
|
|
f"[bold]Revenue:[/bold] {o.revenue:.2f}\n"
|
|
f"[bold]State:[/bold] {o.state_text}\n"
|
|
f"[bold]Order Time:[/bold] {o.order_time[:19] if o.order_time else 'N/A'}\n"
|
|
f"[bold]Receive Time:[/bold] {o.receive_time[:19] if o.receive_time else 'N/A'}\n"
|
|
f"[bold]Platform Order:[/bold] {o.platform_order or 'N/A'}\n"
|
|
f"[bold]Version:[/bold] {o.ver}\n"
|
|
f"[bold]Custom Data:[/bold] {o.custom_data or 'N/A'}"
|
|
)
|
|
yield Vertical(
|
|
Static(detail_text),
|
|
Button("Back", id="back_btn"),
|
|
classes="detail-container",
|
|
)
|
|
|
|
def on_button_pressed(self, event: Button.Pressed) -> None:
|
|
self.dismiss(None)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Main App
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class OrderQueryApp(App):
|
|
"""Main TUI application."""
|
|
|
|
CSS = """
|
|
.input-container {
|
|
align: center middle;
|
|
width: 60;
|
|
height: auto;
|
|
padding: 2;
|
|
border: round $primary;
|
|
}
|
|
.label {
|
|
text-align: center;
|
|
margin-bottom: 1;
|
|
text-style: bold;
|
|
}
|
|
.buttons {
|
|
margin-top: 1;
|
|
align: center middle;
|
|
}
|
|
.list-container {
|
|
padding: 1;
|
|
}
|
|
.detail-container {
|
|
align: center middle;
|
|
width: 80;
|
|
height: auto;
|
|
padding: 2;
|
|
border: round $primary;
|
|
}
|
|
#back_btn {
|
|
margin-top: 1;
|
|
}
|
|
"""
|
|
|
|
BINDINGS = [("q", "quit", "Quit")]
|
|
|
|
_orders: list[Order] = []
|
|
|
|
def on_mount(self) -> None:
|
|
self.push_screen(InputScreen(), self._handle_user_id)
|
|
|
|
def _handle_user_id(self, user_id: str | None) -> None:
|
|
if user_id is None:
|
|
return
|
|
try:
|
|
config = load_config()
|
|
self._orders = query_orders(config, user_id)
|
|
except SystemExit:
|
|
raise
|
|
except Exception as e:
|
|
self.notify(str(e), severity="error")
|
|
self.push_screen(InputScreen(), self._handle_user_id)
|
|
return
|
|
if not self._orders:
|
|
self.notify("No orders found for this user", severity="warning")
|
|
self.push_screen(InputScreen(), self._handle_user_id)
|
|
return
|
|
self.push_screen(OrderListScreen(self._orders), self._handle_order_selected)
|
|
|
|
def _handle_order_selected(self, order_id: str | None) -> None:
|
|
if order_id is None:
|
|
self.push_screen(InputScreen(), self._handle_user_id)
|
|
return
|
|
order = next((o for o in self._orders if o.order_id == order_id), None)
|
|
if order:
|
|
self.push_screen(
|
|
OrderDetailScreen(order), lambda _: self._show_list_again()
|
|
)
|
|
|
|
def _show_list_again(self) -> None:
|
|
self.push_screen(OrderListScreen(self._orders), self._handle_order_selected)
|
|
|
|
|
|
def main() -> None:
|
|
app = OrderQueryApp()
|
|
app.run()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|