romfast/gomag-vending
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(dashboard): redesign UI with smart polling, unified sync card, filter bar

Browse Source 
Replace SSE with smart polling (30s idle / 3s when running). Unify sync
panel into single two-row card with live progress text. Add unified filter
bar (period dropdown, status pills, search) with period-total counts.
Add Client/Cont tooltip for different shipping/billing persons. Add SKU
mappings pct_total badges + complete/incomplete filter + 409 duplicate
check. Add missing SKUs search + rescan progress UX. Migrate SQLite
orders schema (shipping_name, billing_name, payment_method,
delivery_method). Fix JSON_OUTPUT_DIR path for server running from
project root. Fix pagination controls showing top+bottom with per-page
selector (25/50/100/250).

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Marius Mutu
2026-03-13 17:55:36 +02:00
parent 82196b9dc0
commit 5f8b9b6003
14 changed files with 1235 additions and 648 deletions
Download Patch File Download Diff File Expand all files Collapse all files

159
api/app/services/sync_service.py
View File

@@ -13,28 +13,10 @@ logger = logging.getLogger(__name__)
_sync_lock = asyncio.Lock()
_current_sync = None # dict with run_id, status, progress info
# SSE subscriber system
_subscribers: list[asyncio.Queue] = []
# In-memory text log buffer per run
_run_logs: dict[str, list[str]] = {}
def subscribe() -> asyncio.Queue:
"""Subscribe to sync events. Returns a queue that will receive event dicts."""
q = asyncio.Queue()
_subscribers.append(q)
return q
def unsubscribe(q: asyncio.Queue):
"""Unsubscribe from sync events."""
try:
_subscribers.remove(q)
except ValueError:
pass
def _log_line(run_id: str, message: str):
"""Append a timestamped line to the in-memory log buffer."""
if run_id not in _run_logs:
@@ -51,13 +33,17 @@ def get_run_text_log(run_id: str) -> str | None:
return "\n".join(lines)
async def _emit(event: dict):
"""Push an event to all subscriber queues."""
for q in _subscribers:
try:
q.put_nowait(event)
except asyncio.QueueFull:
pass
def _update_progress(phase: str, phase_text: str, current: int = 0, total: int = 0,
counts: dict = None):
"""Update _current_sync with progress details for polling."""
global _current_sync
if _current_sync is None:
return
_current_sync["phase"] = phase
_current_sync["phase_text"] = phase_text
_current_sync["progress_current"] = current
_current_sync["progress_total"] = total
_current_sync["counts"] = counts or {"imported": 0, "skipped": 0, "errors": 0}
async def get_sync_status():
@@ -80,7 +66,12 @@ async def prepare_sync(id_pol: int = None, id_sectie: int = None) -> dict:
"run_id": run_id,
"status": "running",
"started_at": datetime.now().isoformat(),
"progress": "Starting..."
"finished_at": None,
"phase": "starting",
"phase_text": "Starting...",
"progress_current": 0,
"progress_total": 0,
"counts": {"imported": 0, "skipped": 0, "errors": 0},
}
return {"run_id": run_id, "status": "starting"}
@@ -100,11 +91,15 @@ async def run_sync(id_pol: int = None, id_sectie: int = None, run_id: str = None
"run_id": run_id,
"status": "running",
"started_at": datetime.now().isoformat(),
"progress": "Reading JSON files..."
"finished_at": None,
"phase": "reading",
"phase_text": "Reading JSON files...",
"progress_current": 0,
"progress_total": 0,
"counts": {"imported": 0, "skipped": 0, "errors": 0},
}
_current_sync["progress"] = "Reading JSON files..."
await _emit({"type": "phase", "run_id": run_id, "message": "Reading JSON files..."})
_update_progress("reading", "Reading JSON files...")
started_dt = datetime.now()
_run_logs[run_id] = [
@@ -119,7 +114,7 @@ async def run_sync(id_pol: int = None, id_sectie: int = None, run_id: str = None
orders, json_count = order_reader.read_json_orders()
orders.sort(key=lambda o: o.date or '')
await sqlite_service.create_sync_run(run_id, json_count)
await _emit({"type": "phase", "run_id": run_id, "message": f"Found {len(orders)} orders in {json_count} files"})
_update_progress("reading", f"Found {len(orders)} orders in {json_count} files", 0, len(orders))
_log_line(run_id, f"Gasite {len(orders)} comenzi in {json_count} fisiere")
# Populate web_products catalog from all orders (R4)
@@ -131,12 +126,11 @@ async def run_sync(id_pol: int = None, id_sectie: int = None, run_id: str = None
if not orders:
_log_line(run_id, "Nicio comanda gasita.")
await sqlite_service.update_sync_run(run_id, "completed", 0, 0, 0, 0)
_update_progress("completed", "No orders found")
summary = {"run_id": run_id, "status": "completed", "message": "No orders found", "json_files": json_count}
await _emit({"type": "completed", "run_id": run_id, "summary": summary})
return summary
_current_sync["progress"] = f"Validating {len(orders)} orders..."
await _emit({"type": "phase", "run_id": run_id, "message": f"Validating {len(orders)} orders..."})
_update_progress("validation", f"Validating {len(orders)} orders...", 0, len(orders))
# Step 2a: Find new orders (not yet in Oracle)
all_order_numbers = [o.number for o in orders]
@@ -149,7 +143,8 @@ async def run_sync(id_pol: int = None, id_sectie: int = None, run_id: str = None
validation = await asyncio.to_thread(validation_service.validate_skus, all_skus)
importable, skipped = validation_service.classify_orders(orders, validation)
await _emit({"type": "phase", "run_id": run_id, "message": f"{len(importable)} importable, {len(skipped)} skipped (missing SKUs)"})
_update_progress("validation", f"{len(importable)} importable, {len(skipped)} skipped (missing SKUs)",
0, len(importable))
_log_line(run_id, f"Validare SKU-uri: {len(importable)} importabile, {len(skipped)} nemapate")
# Step 2c: Build SKU context from skipped orders
@@ -189,8 +184,7 @@ async def run_sync(id_pol: int = None, id_sectie: int = None, run_id: str = None
logger.info(f"Sync params: ID_POL={id_pol}, ID_SECTIE={id_sectie}")
_log_line(run_id, f"Parametri import: ID_POL={id_pol}, ID_SECTIE={id_sectie}")
if id_pol and importable:
_current_sync["progress"] = "Validating prices..."
await _emit({"type": "phase", "run_id": run_id, "message": "Validating prices..."})
_update_progress("validation", "Validating prices...", 0, len(importable))
_log_line(run_id, "Validare preturi...")
# Gather all CODMATs from importable orders
all_codmats = set()
@@ -216,10 +210,21 @@ async def run_sync(id_pol: int = None, id_sectie: int = None, run_id: str = None
price_result["missing_price"], id_pol
)
# Step 3: Record skipped orders + emit events + store items
# Step 3: Record skipped orders + store items
skipped_count = 0
for order, missing_skus in skipped:
customer = order.billing.company_name or \
f"{order.billing.firstname} {order.billing.lastname}"
skipped_count += 1
# Derive shipping / billing names
shipping_name = ""
if order.shipping:
shipping_name = f"{getattr(order.shipping, 'firstname', '') or ''} {getattr(order.shipping, 'lastname', '') or ''}".strip()
billing_name = f"{getattr(order.billing, 'firstname', '') or ''} {getattr(order.billing, 'lastname', '') or ''}".strip()
if not shipping_name:
shipping_name = billing_name
customer = shipping_name or order.billing.company_name or billing_name
payment_method = getattr(order, 'payment_name', None) or None
delivery_method = getattr(order, 'delivery_name', None) or None
await sqlite_service.upsert_order(
sync_run_id=run_id,
order_number=order.number,
@@ -227,7 +232,11 @@ async def run_sync(id_pol: int = None, id_sectie: int = None, run_id: str = None
customer_name=customer,
status="SKIPPED",
missing_skus=missing_skus,
items_count=len(order.items)
items_count=len(order.items),
shipping_name=shipping_name,
billing_name=billing_name,
payment_method=payment_method,
delivery_method=delivery_method,
)
await sqlite_service.add_sync_run_order(run_id, order.number, "SKIPPED")
# Store order items with mapping status (R9)
@@ -243,28 +252,35 @@ async def run_sync(id_pol: int = None, id_sectie: int = None, run_id: str = None
})
await sqlite_service.add_order_items(order.number, order_items_data)
_log_line(run_id, f"#{order.number} [{order.date or '?'}] {customer} → OMIS (lipsa: {', '.join(missing_skus)})")
await _emit({
"type": "order_result", "run_id": run_id,
"order_number": order.number, "customer_name": customer,
"order_date": order.date,
"status": "SKIPPED", "missing_skus": missing_skus,
"items_count": len(order.items), "progress": f"0/{len(importable)}"
})
_update_progress("skipped", f"Skipped {skipped_count}/{len(skipped)}: #{order.number} {customer}",
0, len(importable),
{"imported": 0, "skipped": skipped_count, "errors": 0})
# Step 4: Import valid orders
imported_count = 0
error_count = 0
for i, order in enumerate(importable):
progress_str = f"{i+1}/{len(importable)}"
_current_sync["progress"] = f"Importing {progress_str}: #{order.number}"
# Derive shipping / billing names
shipping_name = ""
if order.shipping:
shipping_name = f"{getattr(order.shipping, 'firstname', '') or ''} {getattr(order.shipping, 'lastname', '') or ''}".strip()
billing_name = f"{getattr(order.billing, 'firstname', '') or ''} {getattr(order.billing, 'lastname', '') or ''}".strip()
if not shipping_name:
shipping_name = billing_name
customer = shipping_name or order.billing.company_name or billing_name
payment_method = getattr(order, 'payment_name', None) or None
delivery_method = getattr(order, 'delivery_name', None) or None
_update_progress("import",
f"Import {i+1}/{len(importable)}: #{order.number} {customer}",
i + 1, len(importable),
{"imported": imported_count, "skipped": len(skipped), "errors": error_count})
result = await asyncio.to_thread(
import_service.import_single_order,
order, id_pol=id_pol, id_sectie=id_sectie
)
customer = order.billing.company_name or \
f"{order.billing.firstname} {order.billing.lastname}"
# Build order items data for storage (R9)
order_items_data = []
@@ -287,7 +303,11 @@ async def run_sync(id_pol: int = None, id_sectie: int = None, run_id: str = None
status="IMPORTED",
id_comanda=result["id_comanda"],
id_partener=result["id_partener"],
items_count=len(order.items)
items_count=len(order.items),
shipping_name=shipping_name,
billing_name=billing_name,
payment_method=payment_method,
delivery_method=delivery_method,
)
await sqlite_service.add_sync_run_order(run_id, order.number, "IMPORTED")
# Store ROA address IDs (R9)
@@ -298,13 +318,6 @@ async def run_sync(id_pol: int = None, id_sectie: int = None, run_id: str = None
)
await sqlite_service.add_order_items(order.number, order_items_data)
_log_line(run_id, f"#{order.number} [{order.date or '?'}] {customer} → IMPORTAT (ID: {result['id_comanda']})")
await _emit({
"type": "order_result", "run_id": run_id,
"order_number": order.number, "customer_name": customer,
"order_date": order.date,
"status": "IMPORTED", "items_count": len(order.items),
"id_comanda": result["id_comanda"], "progress": progress_str
})
else:
error_count += 1
await sqlite_service.upsert_order(
@@ -315,18 +328,15 @@ async def run_sync(id_pol: int = None, id_sectie: int = None, run_id: str = None
status="ERROR",
id_partener=result.get("id_partener"),
error_message=result["error"],
items_count=len(order.items)
items_count=len(order.items),
shipping_name=shipping_name,
billing_name=billing_name,
payment_method=payment_method,
delivery_method=delivery_method,
)
await sqlite_service.add_sync_run_order(run_id, order.number, "ERROR")
await sqlite_service.add_order_items(order.number, order_items_data)
_log_line(run_id, f"#{order.number} [{order.date or '?'}] {customer} → EROARE: {result['error']}")
await _emit({
"type": "order_result", "run_id": run_id,
"order_number": order.number, "customer_name": customer,
"order_date": order.date,
"status": "ERROR", "error_message": result["error"],
"items_count": len(order.items), "progress": progress_str
})
# Safety: stop if too many errors
if error_count > 10:
@@ -351,11 +361,18 @@ async def run_sync(id_pol: int = None, id_sectie: int = None, run_id: str = None
"missing_skus": len(validation["missing"])
}
_update_progress("completed",
f"Completed: {imported_count} imported, {len(skipped)} skipped, {error_count} errors",
len(importable), len(importable),
{"imported": imported_count, "skipped": len(skipped), "errors": error_count})
if _current_sync:
_current_sync["status"] = status
_current_sync["finished_at"] = datetime.now().isoformat()
logger.info(
f"Sync {run_id} completed: {imported_count} imported, "
f"{len(skipped)} skipped, {error_count} errors"
)
await _emit({"type": "completed", "run_id": run_id, "summary": summary})
duration = (datetime.now() - started_dt).total_seconds()
_log_line(run_id, "")
@@ -367,8 +384,10 @@ async def run_sync(id_pol: int = None, id_sectie: int = None, run_id: str = None
logger.error(f"Sync {run_id} failed: {e}")
_log_line(run_id, f"EROARE FATALA: {e}")
await sqlite_service.update_sync_run(run_id, "failed", 0, 0, 0, 1, error_message=str(e))
_current_sync["error"] = str(e)
await _emit({"type": "failed", "run_id": run_id, "error": str(e)})
if _current_sync:
_current_sync["status"] = "failed"
_current_sync["finished_at"] = datetime.now().isoformat()
_current_sync["error"] = str(e)
return {"run_id": run_id, "status": "failed", "error": str(e)}
finally:
# Keep _current_sync for 10 seconds so status endpoint can show final result