consultoria-as
7b2a904498
- Migrate from SQLite to PostgreSQL with normalized schema - Add 11 lookup tables (fuel_type, body_type, drivetrain, transmission, materials, position_part, manufacture_type, quality_tier, countries, reference_type, shapes) - Rewrite dashboard/server.py (76 routes) using SQLAlchemy text() queries - Rewrite console/db.py (27 methods) using SQLAlchemy ORM - Add models.py with 27 SQLAlchemy model definitions - Add config.py for centralized DB_URL configuration - Add migrate_to_postgres.py migration script - Add docs/METABASE_GUIDE.md with complete data entry guide - Rebrand from "AUTOPARTS DB" to "NEXUS AUTOPARTS" - Fill vehicle data gaps via NHTSA API + heuristics: engines (cylinders, power, torque), brands (country, founded_year), models (body_type, production years), MYE (drivetrain, transmission, trim) Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
355 lines
12 KiB
Python
355 lines
12 KiB
Python
"""
|
|
Catalog navigation screen for the NEXUS AUTOPARTS console application.
|
|
|
|
Provides a three-level drill-down through the parts hierarchy:
|
|
Categories -> Groups -> Parts. An optional vehicle filter (mye_id)
|
|
restricts the parts list to those that fit a specific vehicle
|
|
configuration.
|
|
"""
|
|
|
|
from console.core.screens import Screen
|
|
from console.core.keybindings import Key
|
|
from console.config import APP_NAME, VERSION
|
|
from console.utils.formatting import truncate
|
|
|
|
|
|
# Footer labels for each navigation level
|
|
_FOOTER_CATEGORIES = [
|
|
("1-9", "Seleccionar"),
|
|
("Filtro", "Teclear"),
|
|
("F10", "Menu"),
|
|
("ESC", "Atras"),
|
|
]
|
|
|
|
_FOOTER_GROUPS = [
|
|
("1-9", "Seleccionar"),
|
|
("Filtro", "Teclear"),
|
|
("F10", "Menu"),
|
|
("ESC", "Atras"),
|
|
]
|
|
|
|
_FOOTER_PARTS = [
|
|
("1-9", "Ver parte"),
|
|
("ENTER", "Ver parte"),
|
|
("PgUp/Dn", "Paginar"),
|
|
("ESC", "Atras"),
|
|
]
|
|
|
|
|
|
class CatalogoScreen(Screen):
|
|
"""Hierarchical catalog browser: Categories -> Groups -> Parts."""
|
|
|
|
def __init__(self):
|
|
super().__init__(name="catalogo", title="Catalogo")
|
|
self._filter_text = ""
|
|
self._selected = 0
|
|
self._items = []
|
|
self._parts_data = []
|
|
self._selected_part = 0
|
|
|
|
# ------------------------------------------------------------------
|
|
# Helpers
|
|
# ------------------------------------------------------------------
|
|
|
|
def _ensure_defaults(self, context):
|
|
"""Set default context values if missing."""
|
|
context.setdefault("level", "categories")
|
|
context.setdefault("mye_id", None)
|
|
context.setdefault("brand", "")
|
|
context.setdefault("model", "")
|
|
context.setdefault("year", "")
|
|
context.setdefault("engine", "")
|
|
context.setdefault("category_id", None)
|
|
context.setdefault("category_name", "")
|
|
context.setdefault("group_id", None)
|
|
context.setdefault("group_name", "")
|
|
context.setdefault("page", 1)
|
|
context.setdefault("per_page", 15)
|
|
|
|
def _build_header_title(self, context):
|
|
"""Build the header title based on context."""
|
|
level = context["level"]
|
|
parts = []
|
|
|
|
# Vehicle info if available
|
|
if context.get("brand"):
|
|
vehicle = " ".join(
|
|
filter(None, [
|
|
context["brand"],
|
|
context["model"],
|
|
str(context["year"]) if context["year"] else "",
|
|
])
|
|
)
|
|
parts.append(vehicle)
|
|
|
|
if level == "categories":
|
|
parts.append("Categorias")
|
|
elif level == "groups":
|
|
parts.append(context.get("category_name", "Grupos"))
|
|
elif level == "parts":
|
|
cat = context.get("category_name", "")
|
|
grp = context.get("group_name", "")
|
|
if cat and grp:
|
|
parts.append(f"{cat} > {grp}")
|
|
elif grp:
|
|
parts.append(grp)
|
|
|
|
return " — ".join(parts) if parts else "CATALOGO DE CATEGORIAS"
|
|
|
|
def _load_categories(self, db):
|
|
"""Load and filter categories."""
|
|
categories = db.get_categories()
|
|
if self._filter_text:
|
|
ft = self._filter_text.upper()
|
|
categories = [
|
|
c for c in categories
|
|
if ft in (c.get("name_es") or c.get("name") or "").upper()
|
|
or ft in (c.get("name") or "").upper()
|
|
]
|
|
self._items = [
|
|
(str(i + 1), c.get("name_es") or c.get("name", ""), c["id"])
|
|
for i, c in enumerate(categories)
|
|
]
|
|
|
|
def _load_groups(self, db, category_id):
|
|
"""Load and filter groups for a category."""
|
|
groups = db.get_groups(category_id)
|
|
if self._filter_text:
|
|
ft = self._filter_text.upper()
|
|
groups = [
|
|
g for g in groups
|
|
if ft in (g.get("name_es") or g.get("name") or "").upper()
|
|
or ft in (g.get("name") or "").upper()
|
|
]
|
|
self._items = [
|
|
(str(i + 1), g.get("name_es") or g.get("name", ""), g["id"])
|
|
for i, g in enumerate(groups)
|
|
]
|
|
|
|
def _load_parts(self, db, context):
|
|
"""Load parts for the current group/vehicle with pagination."""
|
|
self._parts_data = db.get_parts(
|
|
group_id=context.get("group_id"),
|
|
mye_id=context.get("mye_id"),
|
|
page=context.get("page", 1),
|
|
per_page=context.get("per_page", 15),
|
|
)
|
|
|
|
def _reset_filter(self):
|
|
"""Reset filter text and selection."""
|
|
self._filter_text = ""
|
|
self._selected = 0
|
|
|
|
# ------------------------------------------------------------------
|
|
# Render
|
|
# ------------------------------------------------------------------
|
|
|
|
def render(self, context, db, renderer):
|
|
self._ensure_defaults(context)
|
|
level = context["level"]
|
|
|
|
# Header
|
|
header_title = self._build_header_title(context)
|
|
renderer.draw_header(
|
|
f" {APP_NAME} v{VERSION}",
|
|
f" {header_title} ",
|
|
)
|
|
|
|
if level == "categories":
|
|
self._load_categories(db)
|
|
display_items = [(num, label) for num, label, _id in self._items]
|
|
renderer.draw_filter_list(
|
|
display_items,
|
|
self._filter_text,
|
|
self._selected,
|
|
title="CATALOGO DE CATEGORIAS",
|
|
)
|
|
renderer.draw_footer(_FOOTER_CATEGORIES)
|
|
|
|
elif level == "groups":
|
|
self._load_groups(db, context["category_id"])
|
|
display_items = [(num, label) for num, label, _id in self._items]
|
|
renderer.draw_filter_list(
|
|
display_items,
|
|
self._filter_text,
|
|
self._selected,
|
|
title=context.get("category_name", "GRUPOS"),
|
|
)
|
|
renderer.draw_footer(_FOOTER_GROUPS)
|
|
|
|
elif level == "parts":
|
|
self._load_parts(db, context)
|
|
headers = ["NUMERO OEM", "DESCRIPCION", "GRUPO", "ALT"]
|
|
widths = [18, 30, 18, 5]
|
|
rows = []
|
|
for p in self._parts_data:
|
|
alts = len(db.get_alternatives(p["id"]))
|
|
rows.append((
|
|
truncate(p.get("oem_part_number", ""), 18),
|
|
truncate(
|
|
p.get("name_es") or p.get("name", ""), 30
|
|
),
|
|
truncate(p.get("group_name", ""), 18),
|
|
str(alts) if alts > 0 else "",
|
|
))
|
|
|
|
page = context.get("page", 1)
|
|
renderer.draw_table(
|
|
headers,
|
|
rows,
|
|
widths,
|
|
page_info={"page": page, "total_pages": page, "total_rows": len(rows)},
|
|
selected_row=self._selected_part,
|
|
)
|
|
renderer.draw_footer(_FOOTER_PARTS)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Key handling
|
|
# ------------------------------------------------------------------
|
|
|
|
def on_key(self, key, context, db, renderer, nav):
|
|
self._ensure_defaults(context)
|
|
level = context["level"]
|
|
|
|
if level in ("categories", "groups"):
|
|
return self._handle_filter_level(key, context)
|
|
elif level == "parts":
|
|
return self._handle_parts_level(key, context)
|
|
|
|
return None
|
|
|
|
def _handle_filter_level(self, key, context):
|
|
"""Handle keys for categories and groups levels (filter list)."""
|
|
level = context["level"]
|
|
|
|
# ESC: go back
|
|
if key == Key.ESCAPE:
|
|
if level == "groups":
|
|
context["level"] = "categories"
|
|
context["category_id"] = None
|
|
context["category_name"] = ""
|
|
self._reset_filter()
|
|
return None
|
|
return "back"
|
|
|
|
# Arrow navigation
|
|
if key == Key.UP:
|
|
if self._selected > 0:
|
|
self._selected -= 1
|
|
return None
|
|
|
|
if key == Key.DOWN:
|
|
if self._items and self._selected < len(self._items) - 1:
|
|
self._selected += 1
|
|
return None
|
|
|
|
# ENTER: select current item
|
|
if key == Key.ENTER:
|
|
if self._items and 0 <= self._selected < len(self._items):
|
|
return self._select_item(context, self._selected)
|
|
return None
|
|
|
|
# Number keys: direct selection (1-9)
|
|
if 49 <= key <= 57: # '1'..'9'
|
|
idx = key - 49 # 0-based
|
|
if 0 <= idx < len(self._items):
|
|
return self._select_item(context, idx)
|
|
return None
|
|
|
|
# Backspace: remove last filter character
|
|
if key in (Key.BACKSPACE, 8):
|
|
if self._filter_text:
|
|
self._filter_text = self._filter_text[:-1]
|
|
self._selected = 0
|
|
elif context["level"] == "groups":
|
|
context["level"] = "categories"
|
|
context["category_id"] = None
|
|
context["category_name"] = ""
|
|
self._reset_filter()
|
|
else:
|
|
return "back"
|
|
return None
|
|
|
|
# Printable characters: add to filter
|
|
if 32 <= key <= 126:
|
|
self._filter_text += chr(key)
|
|
self._selected = 0
|
|
return None
|
|
|
|
return None
|
|
|
|
def _select_item(self, context, idx):
|
|
"""Handle selection of an item at the given index."""
|
|
_num, label, item_id = self._items[idx]
|
|
level = context["level"]
|
|
|
|
if level == "categories":
|
|
context["level"] = "groups"
|
|
context["category_id"] = item_id
|
|
context["category_name"] = label
|
|
self._reset_filter()
|
|
return None
|
|
|
|
elif level == "groups":
|
|
context["level"] = "parts"
|
|
context["group_id"] = item_id
|
|
context["group_name"] = label
|
|
context["page"] = 1
|
|
self._selected_part = 0
|
|
self._reset_filter()
|
|
return None
|
|
|
|
return None
|
|
|
|
def _handle_parts_level(self, key, context):
|
|
"""Handle keys for the parts table level."""
|
|
# ESC: go back to groups
|
|
if key == Key.ESCAPE:
|
|
context["level"] = "groups"
|
|
context["group_id"] = None
|
|
context["group_name"] = ""
|
|
self._selected_part = 0
|
|
self._reset_filter()
|
|
return None
|
|
|
|
# Arrow navigation
|
|
if key == Key.UP:
|
|
if self._selected_part > 0:
|
|
self._selected_part -= 1
|
|
return None
|
|
|
|
if key == Key.DOWN:
|
|
if self._parts_data and self._selected_part < len(self._parts_data) - 1:
|
|
self._selected_part += 1
|
|
return None
|
|
|
|
# ENTER: view selected part detail
|
|
if key == Key.ENTER:
|
|
if self._parts_data and 0 <= self._selected_part < len(self._parts_data):
|
|
part = self._parts_data[self._selected_part]
|
|
return ("parte_detalle", {"part_id": part["id"]}, "Parte")
|
|
return None
|
|
|
|
# Number keys: direct selection (1-9)
|
|
if 49 <= key <= 57: # '1'..'9'
|
|
idx = key - 49 # 0-based
|
|
if 0 <= idx < len(self._parts_data):
|
|
part = self._parts_data[idx]
|
|
return ("parte_detalle", {"part_id": part["id"]}, "Parte")
|
|
return None
|
|
|
|
# PgDn: next page
|
|
if key == Key.PGDN:
|
|
context["page"] = context.get("page", 1) + 1
|
|
self._selected_part = 0
|
|
return None
|
|
|
|
# PgUp: previous page
|
|
if key == Key.PGUP:
|
|
if context.get("page", 1) > 1:
|
|
context["page"] = context["page"] - 1
|
|
self._selected_part = 0
|
|
return None
|
|
|
|
return None
|