""" Consultas de solo lectura sobre data/shopify_sap.db (y logs/events.jsonl). Uso (desde la raíz del proyecto): .venv/bin/python -m tools.consultar resumen .venv/bin/python -m tools.consultar descargadas [--store snobbycl] [--limit 50] .venv/bin/python -m tools.consultar integradas [--store snobbycl] [--limit 50] .venv/bin/python -m tools.consultar errores [--store snobbycl] [--limit 50] .venv/bin/python -m tools.consultar orden --name "#8064" [--store snobbycl] - resumen: órdenes por tienda y estado, con barras. - descargadas: últimas órdenes descargadas (por fecha de descarga). - integradas: últimas órdenes integradas, con los documentos SAP creados. - errores: órdenes en error, documentos con error y eventos de error del log. Las horas de la BD (UTC) se muestran en hora local (APP_TIMEZONE o la del sistema); las del log de eventos ya vienen en hora local. - orden: cabecera, fulfillments y todos los pasos (documentos) de una orden. """ import argparse import json import os import sqlite3 from datetime import datetime, timezone from pathlib import Path from typing import Any, List, Sequence from zoneinfo import ZoneInfo from core.config import cargar_env ROOT = Path(__file__).resolve().parent.parent DB_PATH = ROOT / "data" / "shopify_sap.db" EVENTS_PATH = ROOT / "logs" / "events.jsonl" def _zona(): cargar_env() nombre = os.getenv("APP_TIMEZONE") return ZoneInfo(nombre) if nombre else None # None = zona horaria del sistema def hora_local(valor: Any) -> Any: """ La BD guarda time_insert/time_update en UTC (strftime 'now'); se muestran en hora local. Zona: variable APP_TIMEZONE (ej. America/Santiago) si está definida; si no, la del sistema. """ if not valor: return valor try: dt = datetime.strptime(str(valor), "%Y-%m-%d %H:%M:%S.%f" if "." in str(valor) else "%Y-%m-%d %H:%M:%S") return dt.replace(tzinfo=timezone.utc).astimezone(_zona()).strftime("%Y-%m-%d %H:%M:%S") except ValueError: return valor def conectar() -> sqlite3.Connection: # Solo lectura: no bloquea ni modifica la BD aunque el scheduler esté corriendo. conn = sqlite3.connect(f"file:{DB_PATH}?mode=ro", uri=True) conn.row_factory = sqlite3.Row return conn def tabla(filas: Sequence[Sequence[Any]], columnas: List[str], max_ancho: int = 60) -> None: if not filas: print("(sin resultados)") return celdas = [[("" if v is None else str(v))[:max_ancho] for v in fila] for fila in filas] anchos = [max(len(c), *(len(f[i]) for f in celdas)) for i, c in enumerate(columnas)] linea = " ".join(c.ljust(anchos[i]) for i, c in enumerate(columnas)) print(linea) print(" ".join("-" * a for a in anchos)) for f in celdas: print(" ".join(v.ljust(anchos[i]) for i, v in enumerate(f))) print(f"\n{len(filas)} fila(s)") def filtro_tienda(store: str | None, alias: str = "o") -> tuple[str, list]: return (f"AND {alias}.store_code = ?", [store]) if store else ("", []) # ============================================================================ def resumen(conn: sqlite3.Connection, args) -> None: filas = conn.execute( "SELECT store_code, status, COUNT(*) n FROM orders GROUP BY store_code, status ORDER BY store_code, n DESC" ).fetchall() maximo = max((r["n"] for r in filas), default=1) salida = [(r["store_code"], r["status"], r["n"], "█" * max(1, round(r["n"] / maximo * 40))) for r in filas] tabla(salida, ["tienda", "estado", "órdenes", ""]) def descargadas(conn: sqlite3.Connection, args) -> None: extra, params = filtro_tienda(args.store) filas = conn.execute( f""" SELECT o.time_insert, o.store_code, o.order_number, o.shopify_order_id, o.financial_status, o.fulfillment_status, o.status FROM orders o WHERE 1=1 {extra} ORDER BY o.time_insert DESC, o.sys_id DESC LIMIT ? """, params + [args.limit], ).fetchall() tabla([(hora_local(r[0]),) + tuple(r)[1:] for r in filas], ["descargada (local)", "tienda", "orden", "shopify_id", "financiero", "fulfillment", "estado"]) def integradas(conn: sqlite3.Connection, args) -> None: extra, params = filtro_tienda(args.store) filas = conn.execute( f""" SELECT o.time_update, o.store_code, o.order_number, o.card_code, (SELECT group_concat(d.doc_type || ' ' || d.sap_doc_num, ' | ') FROM order_documents d WHERE d.order_sys_id = o.sys_id AND d.status = 'created') docs FROM orders o WHERE o.status = 'integrated' {extra} ORDER BY o.time_update DESC LIMIT ? """, params + [args.limit], ).fetchall() tabla([(hora_local(r[0]),) + tuple(r)[1:] for r in filas], ["integrada (local)", "tienda", "orden", "cliente", "documentos SAP"], max_ancho=110) def errores(conn: sqlite3.Connection, args) -> None: extra, params = filtro_tienda(args.store) print("== Órdenes en estado error ==") filas = conn.execute( f"SELECT time_update, store_code, order_number, last_error FROM orders o WHERE status = 'error' {extra} ORDER BY time_update DESC LIMIT ?", params + [args.limit], ).fetchall() tabla([(hora_local(r[0]),) + tuple(r)[1:] for r in filas], ["actualizada (local)", "tienda", "orden", "último error"], max_ancho=110) print("\n== Documentos con error ==") filas = conn.execute( f""" SELECT d.time_update, o.store_code, o.order_number, d.doc_type, d.attempts, d.error_message FROM order_documents d JOIN orders o ON o.sys_id = d.order_sys_id WHERE d.status = 'error' {extra} ORDER BY d.time_update DESC LIMIT ? """, params + [args.limit], ).fetchall() tabla([(hora_local(r[0]),) + tuple(r)[1:] for r in filas], ["actualizado (local)", "tienda", "orden", "documento", "intentos", "error"], max_ancho=90) print("\n== Eventos de error recientes (logs/events.jsonl) ==") eventos = [] if EVENTS_PATH.exists(): for linea in EVENTS_PATH.read_text(encoding="utf-8").splitlines(): try: e = json.loads(linea) except json.JSONDecodeError: continue if e.get("level") in ("ERROR", "CRITICAL") and (not args.store or e.get("store_code") == args.store): eventos.append((e.get("timestamp"), e.get("store_code"), e.get("shopify_order_id"), e.get("event"), e.get("message"))) tabla(eventos[-args.limit:][::-1], ["cuándo", "tienda", "shopify_id", "evento", "mensaje"], max_ancho=90) def orden(conn: sqlite3.Connection, args) -> None: extra, params = filtro_tienda(args.store) nombre = args.name if args.name.startswith("#") else f"#{args.name}" ordenes = conn.execute(f"SELECT * FROM orders o WHERE o.order_number = ? {extra}", [nombre] + params).fetchall() if not ordenes: print(f"No hay órdenes con name {nombre}") return for o in ordenes: print(f"\n===== {o['order_number']} | {o['store_code']} | shopify_id {o['shopify_order_id']} =====") for campo in ("status", "last_error", "financial_status", "fulfillment_status", "payment_gateway_names", "card_code", "tags", "shopify_created_at", "time_insert", "time_update"): print(f" {campo:<22} {hora_local(o[campo]) if campo in ('time_insert', 'time_update') else o[campo]}") print("\n Fulfillments") fulfillments = conn.execute("SELECT * FROM order_fulfillments WHERE order_sys_id = ?", (o["sys_id"],)).fetchall() tabla([(f["sys_id"], f["shopify_fulfillment_id"], f["location_id"], "tienda" if f["is_store"] else "CD", f["warehouse_code"]) for f in fulfillments], ["id", "shopify_fulfillment", "location", "tipo", "bodega"]) print("\n Pasos (documentos SAP)") docs = conn.execute( "SELECT * FROM order_documents WHERE order_sys_id = ? ORDER BY sequence, fulfillment_sys_id, sys_id", (o["sys_id"],) ).fetchall() tabla([(d["sequence"], d["doc_type"], d["fulfillment_sys_id"], d["status"], d["sap_doc_entry"], d["sap_doc_num"], d["attempts"], hora_local(d["time_update"]), d["error_message"]) for d in docs], ["seq", "documento", "fulfillment", "estado", "DocEntry", "DocNum", "intentos", "actualizado (local)", "error"], max_ancho=70) # ============================================================================ def main() -> None: parser = argparse.ArgumentParser(description="Consultas de solo lectura sobre la BD de Shopify → SAP") sub = parser.add_subparsers(dest="comando", required=True) sub.add_parser("resumen") for nombre in ("descargadas", "integradas", "errores"): p = sub.add_parser(nombre) p.add_argument("--store") p.add_argument("--limit", type=int, default=50) p = sub.add_parser("orden") p.add_argument("--name", required=True) p.add_argument("--store") args = parser.parse_args() with conectar() as conn: {"resumen": resumen, "descargadas": descargadas, "integradas": integradas, "errores": errores, "orden": orden}[args.comando](conn, args) if __name__ == "__main__": main()