From 73544c0a85387066d256f9c00932ebd58d7663ac Mon Sep 17 00:00:00 2001 From: Victor Fraile Garcia Date: Sun, 19 Jul 2026 10:10:54 +0200 Subject: [PATCH] feat(ucepsa): add Odoo Shop Floor sessions shadow --- ...7_odoo_shopfloor_sessions_shadow_v0.3.3.md | 244 +++ .../ops/manage_odoo_shopfloor_sync_v033.sh | 132 ++ ...te_odoo_shopfloor_sessions_shadow_v033.sql | 207 ++ ...sa_odoo_shopfloor_sessions_shadow_v033.sql | 613 ++++++ .../tools/odoo_shopfloor_session_admin.py | 529 +++++ .../tools/odoo_shopfloor_session_sync.py | 1866 +++++++++++++++++ 6 files changed, 3591 insertions(+) create mode 100644 ucepsa/edge-oee-demo/docs/runbooks/097_odoo_shopfloor_sessions_shadow_v0.3.3.md create mode 100644 ucepsa/edge-oee-demo/ops/manage_odoo_shopfloor_sync_v033.sh create mode 100644 ucepsa/edge-oee-demo/ops/validate_odoo_shopfloor_sessions_shadow_v033.sql create mode 100644 ucepsa/edge-oee-demo/sql/versions/104_ucepsa_odoo_shopfloor_sessions_shadow_v033.sql create mode 100644 ucepsa/edge-oee-demo/tools/odoo_shopfloor_session_admin.py create mode 100644 ucepsa/edge-oee-demo/tools/odoo_shopfloor_session_sync.py diff --git a/ucepsa/edge-oee-demo/docs/runbooks/097_odoo_shopfloor_sessions_shadow_v0.3.3.md b/ucepsa/edge-oee-demo/docs/runbooks/097_odoo_shopfloor_sessions_shadow_v0.3.3.md new file mode 100644 index 0000000..fdb96a2 --- /dev/null +++ b/ucepsa/edge-oee-demo/docs/runbooks/097_odoo_shopfloor_sessions_shadow_v0.3.3.md @@ -0,0 +1,244 @@ +# UCEPSA — Odoo Shop Floor Sessions SHADOW v0.3.3 + +## Objetivo + +Construir sesiones canónicas de producción a partir del flujo real de Odoo +Shop Floor sin pedir un segundo login a la operaria y sin alterar Odoo, balizas, +RevPi, WISE, la cola oficial ni el Loss Ledger. + +## Evidencia base + +### Workorder 87 + +```text +WH/MO/00081-001 +CORT-02 +11:48:31 → 13:48:58 hora local +operaria 43 / M. Angeles Martinez +85 kg contabilizados +``` + +La operaria está confirmada por dos filas de +`mrp.workcenter.productivity`. + +### Workorder 810 + +```text +WH/MO/00081-002 +CORT-02 +14:01:56 → 15:50:59 hora local +candidata 20 / MARIA CIBÚ +76 kg candidatos +``` + +La persona y la cantidad quedan `PENDING_HUMAN`. El intervalo de sesión, +máquina, orden y workorder sí están confirmados. + +## Alcance de v0.3.3 + +Crea: + +- snapshot actual por workorder; +- eventos secuenciales por transición; +- sesión canónica por workorder; +- resolución de operaria con confianza; +- separación entre cantidad candidata y contabilizada; +- cola de revisión SHADOW; +- contexto paralelo `ODOO_SHOPFLOOR`; +- herramienta posterior de confirmación humana. + +No cambia: + +- `production_context_sessions`; +- vistas de candidatos de paro v0.3.1; +- cola oficial; +- Odoo; +- MQTT de balizas; +- RevPi; +- WISE; +- balizas. + +La asociación de paros a las sesiones nuevas se hará en una versión posterior, +cuando la captura haya sido validada en operación continua. + +## Reglas de operaria + +1. Una única persona real en productividad: + - operaria confirmada; + - confianza alta. + +2. Sin productividad y una única persona real asignada, excluyendo perfiles + técnicos: + - candidata; + - confianza media; + - confirmación humana pendiente. + +3. Más de una persona real o ninguna evidencia: + - sin resolver; + - confianza baja. + +`employee_id=35 / CORTADORAS` se excluye porque es el perfil técnico compartido +de la tablet. + +## Reglas de cantidad + +- Producción `done` y cantidad contabilizada: + `FINAL_POSTED`. +- Producción `to_close` con `qty_producing`: + `PROVISIONAL_TO_CLOSE`. +- La cantidad candidata no se transforma en cantidad confirmada sin evidencia + de stock final o confirmación humana. + +## Tiempo + +Los datetimes XML-RPC de Odoo se interpretan como UTC. + +La sesión usa: + +```text +mrp.workorder.date_start +mrp.workorder.date_finished +``` + +No usa `duration` como reloj de presencia. + +## Cadena productiva + +La clave se obtiene quitando el sufijo de backorder: + +```text +WH/MO/00081-001 → ODOO:WH/MO/00081 +WH/MO/00081-002 → ODOO:WH/MO/00081 +``` + +`parent_production_id` y `child_production_ids` no se usan para los relevos, +porque en este caso representan una relación con la fabricación del tubo. + +## Despliegue + +### 1. Aplicar migración + +```bash +FILE=/srv/mesavault/40-clients/ucepsa/edge-oee-demo/sql/versions/104_ucepsa_odoo_shopfloor_sessions_shadow_v033.sql + +docker exec -i mv_ucepsa_postgres_hot sh -lc \ + 'psql -v ON_ERROR_STOP=1 -U "$POSTGRES_USER" -d "$POSTGRES_DB"' \ + < "$FILE" +``` + +No es necesario detener servicios. + +### 2. Instalar herramientas en runtime + +```bash +REPO=/srv/mesavault/40-clients/ucepsa/edge-oee-demo +RUNTIME=/srv/mesavault/edge-oee-ucepsa + +install -m 0755 \ + "$REPO/tools/odoo_shopfloor_session_sync.py" \ + "$RUNTIME/tools/odoo_shopfloor_session_sync.py" + +install -m 0755 \ + "$REPO/tools/odoo_shopfloor_session_admin.py" \ + "$RUNTIME/tools/odoo_shopfloor_session_admin.py" + +install -m 0755 \ + "$REPO/ops/manage_odoo_shopfloor_sync_v033.sh" \ + "$RUNTIME/ops/manage_odoo_shopfloor_sync_v033.sh" +``` + +### 3. Ensayo de los dos workorders conocidos + +```bash +cd /srv/mesavault/edge-oee-ucepsa + +ops/manage_odoo_shopfloor_sync_v033.sh dry-run-known +``` + +Debe mostrar: + +```text +87 → operaria 43 / HIGH / 85 kg FINAL_POSTED +810 → candidata 20 / MEDIUM / 76 kg PROVISIONAL_TO_CLOSE +``` + +### 4. Backfill + +```bash +ops/manage_odoo_shopfloor_sync_v033.sh backfill-known +``` + +### 5. Validación + +```bash +VALIDATION=/srv/mesavault/40-clients/ucepsa/edge-oee-demo/ops/validate_odoo_shopfloor_sessions_shadow_v033.sql + +docker exec -i mv_ucepsa_postgres_hot sh -lc \ + 'psql -v ON_ERROR_STOP=1 -U "$POSTGRES_USER" -d "$POSTGRES_DB"' \ + < "$VALIDATION" +``` + +Resultados obligatorios: + +```text +official_rows = 0 +invalid_session_intervals = 0 +known_backfill_ok = true +consecutive_duplicate_event_count = 0 +generated_production_context_rows = 0 +official_operator_queue_rows = 0 +``` + +### 6. Activar sincronización continua + +Solo después de validar el backfill: + +```bash +cd /srv/mesavault/edge-oee-ucepsa +ops/manage_odoo_shopfloor_sync_v033.sh start +ops/manage_odoo_shopfloor_sync_v033.sh status +docker logs --tail 100 mv_ucepsa_odoo_shopfloor_session_sync +``` + +## Confirmación posterior de Juan Pablo + +No ejecutar hasta recibir confirmación. + +### Confirmar operaria candidata + +```bash +docker exec -i mv_ucepsa_odoo_order_state_pg_sink \ + python tools/odoo_shopfloor_session_admin.py \ + confirm-operator \ + --workorder-id 810 \ + --use-candidate \ + --confirmed-by "Juan Pablo" \ + --notes "Confirmado el segundo tramo de CORT-02." +``` + +### Confirmar cantidad + +```bash +docker exec -i mv_ucepsa_odoo_order_state_pg_sink \ + python tools/odoo_shopfloor_session_admin.py \ + confirm-quantity \ + --workorder-id 810 \ + --quantity 76 \ + --confirmed-by "Juan Pablo" \ + --notes "Confirmados 76 kg declarados en el segundo tramo." +``` + +La sincronización posterior conserva las confirmaciones humanas. + +## Parada y rollback operativo + +Parar el sincronizador: + +```bash +/srv/mesavault/edge-oee-ucepsa/ops/manage_odoo_shopfloor_sync_v033.sh stop +``` + +La parada no afecta Odoo, captura de máquina, MQTT ni balizas. + +Las tablas nuevas pueden mantenerse para diagnóstico. No se recomienda borrarlas +durante el piloto. diff --git a/ucepsa/edge-oee-demo/ops/manage_odoo_shopfloor_sync_v033.sh b/ucepsa/edge-oee-demo/ops/manage_odoo_shopfloor_sync_v033.sh new file mode 100644 index 0000000..3efab4e --- /dev/null +++ b/ucepsa/edge-oee-demo/ops/manage_odoo_shopfloor_sync_v033.sh @@ -0,0 +1,132 @@ +#!/usr/bin/env bash +set -euo pipefail + +ACTION="${1:-status}" +ROOT="${SHOPFLOOR_RUNTIME_ROOT:-/srv/mesavault/edge-oee-ucepsa}" +CONTAINER="${SHOPFLOOR_SYNC_CONTAINER:-mv_ucepsa_odoo_shopfloor_session_sync}" +NETWORK="${SHOPFLOOR_DOCKER_NETWORK:-mv_ucepsa_net}" +INTERVAL="${SHOPFLOOR_SYNC_INTERVAL_S:-15}" +LOOKBACK="${SHOPFLOOR_SYNC_LOOKBACK_HOURS:-168}" +OUTPUT="${SHOPFLOOR_SYNC_OUTPUT:-/tmp/ucepsa_odoo_shopfloor_session_sync.jsonl}" + +cd "$ROOT" + +image() { + docker inspect \ + mv_ucepsa_odoo_order_state_pg_sink \ + --format '{{.Config.Image}}' +} + +start_sync() { + local sync_image + sync_image="$(image)" + + docker rm -f "$CONTAINER" >/dev/null 2>&1 || true + + docker run -d \ + --name "$CONTAINER" \ + --restart unless-stopped \ + --network "$NETWORK" \ + --env-file .env \ + -e ODOO_DB_OVERRIDE=ucepsa_prod \ + -v "$ROOT:/app:ro" \ + -v /tmp:/probe-output \ + -w /app \ + --entrypoint python \ + "$sync_image" \ + tools/odoo_shopfloor_session_sync.py \ + --watch \ + --interval "$INTERVAL" \ + --lookback-hours "$LOOKBACK" \ + --output "/probe-output/$(basename "$OUTPUT")" + + docker ps --format 'table {{.Names}}\t{{.Status}}' \ + | grep -E "$CONTAINER|NAMES" +} + +one_shot() { + local sync_image + sync_image="$(image)" + + docker run --rm -i \ + --network "$NETWORK" \ + --env-file .env \ + -e ODOO_DB_OVERRIDE=ucepsa_prod \ + -v "$ROOT:/app:ro" \ + -w /app \ + --entrypoint python \ + "$sync_image" \ + tools/odoo_shopfloor_session_sync.py \ + --once \ + --lookback-hours "$LOOKBACK" +} + +dry_run_known() { + local sync_image + sync_image="$(image)" + + docker run --rm -i \ + --network "$NETWORK" \ + --env-file .env \ + -e ODOO_DB_OVERRIDE=ucepsa_prod \ + -v "$ROOT:/app:ro" \ + -w /app \ + --entrypoint python \ + "$sync_image" \ + tools/odoo_shopfloor_session_sync.py \ + --once \ + --dry-run \ + --workorder-id 87 \ + --workorder-id 810 +} + +backfill_known() { + local sync_image + sync_image="$(image)" + + docker run --rm -i \ + --network "$NETWORK" \ + --env-file .env \ + -e ODOO_DB_OVERRIDE=ucepsa_prod \ + -v "$ROOT:/app:ro" \ + -w /app \ + --entrypoint python \ + "$sync_image" \ + tools/odoo_shopfloor_session_sync.py \ + --once \ + --workorder-id 87 \ + --workorder-id 810 +} + +case "$ACTION" in + start) + start_sync + ;; + stop) + docker stop "$CONTAINER" + ;; + restart) + docker rm -f "$CONTAINER" >/dev/null 2>&1 || true + start_sync + ;; + status) + docker ps -a --format 'table {{.Names}}\t{{.Status}}' \ + | grep -E "$CONTAINER|NAMES" + ;; + logs) + docker logs --tail 200 -f "$CONTAINER" + ;; + once) + one_shot + ;; + dry-run-known) + dry_run_known + ;; + backfill-known) + backfill_known + ;; + *) + echo "Uso: $0 {start|stop|restart|status|logs|once|dry-run-known|backfill-known}" >&2 + exit 2 + ;; +esac diff --git a/ucepsa/edge-oee-demo/ops/validate_odoo_shopfloor_sessions_shadow_v033.sql b/ucepsa/edge-oee-demo/ops/validate_odoo_shopfloor_sessions_shadow_v033.sql new file mode 100644 index 0000000..de94082 --- /dev/null +++ b/ucepsa/edge-oee-demo/ops/validate_odoo_shopfloor_sessions_shadow_v033.sql @@ -0,0 +1,207 @@ +\pset pager off + +\echo '=== 1. Objetos v0.3.3 ===' + +SELECT + to_regclass( + 'mv_hot.odoo_shopfloor_workorder_current' + ) AS current_table, + to_regclass( + 'mv_hot.odoo_shopfloor_workorder_events' + ) AS events_table, + to_regclass( + 'mv_loss_intelligence.odoo_shopfloor_sessions' + ) AS sessions_table, + to_regclass( + 'mv_reports_ucepsa_prod.li_odoo_shopfloor_sessions_v1' + ) AS sessions_view, + to_regclass( + 'mv_reports_ucepsa_prod.li_odoo_shopfloor_context_intervals_v1' + ) AS context_view, + to_regclass( + 'mv_reports_ucepsa_prod.li_odoo_shopfloor_session_review_queue_v1' + ) AS review_queue_view; + +\echo '=== 2. SHADOW obligatorio ===' + +SELECT + count(*) FILTER ( + WHERE official_eligible + ) AS official_rows, + count(*) AS total_sessions +FROM mv_loss_intelligence.odoo_shopfloor_sessions; + +\echo '=== 3. Intervalos inválidos ===' + +SELECT count(*) AS invalid_session_intervals +FROM mv_loss_intelligence.odoo_shopfloor_sessions +WHERE ended_at IS NOT NULL + AND started_at IS NOT NULL + AND ended_at < started_at; + +\echo '=== 4. Sesiones conocidas 87 y 810 ===' + +SELECT + odoo_workorder_id, + machine_id, + production_order, + production_chain_key, + started_at, + ended_at, + session_status, + operator_employee_id, + operator_name, + operator_candidate_employee_id, + operator_candidate_name, + operator_resolution, + operator_confidence, + operator_confirmation_status, + planned_quantity, + declared_quantity, + declared_quantity_candidate, + posted_quantity, + remaining_planned_quantity, + quantity_status, + quantity_confirmation_status, + confidence, + official_eligible +FROM mv_loss_intelligence.odoo_shopfloor_sessions +WHERE odoo_workorder_id IN (87, 810) +ORDER BY odoo_workorder_id; + +\echo '=== 5. Validación esperada de los dos tramos ===' + +WITH expected AS ( + SELECT + count(*) FILTER ( + WHERE odoo_workorder_id = 87 + AND machine_id = 'CORT-02' + AND operator_employee_id = 43 + AND operator_confirmation_status = + 'EVIDENCE_CONFIRMED' + AND declared_quantity = 85 + AND posted_quantity = 85 + AND quantity_status = 'FINAL_POSTED' + AND official_eligible = false + ) AS workorder_87_ok, + count(*) FILTER ( + WHERE odoo_workorder_id = 810 + AND machine_id = 'CORT-02' + AND operator_employee_id IS NULL + AND operator_candidate_employee_id = 20 + AND operator_confirmation_status = + 'PENDING_HUMAN' + AND declared_quantity IS NULL + AND declared_quantity_candidate = 76 + AND posted_quantity = 0 + AND quantity_status = + 'PROVISIONAL_TO_CLOSE' + AND official_eligible = false + ) AS workorder_810_ok + FROM mv_loss_intelligence.odoo_shopfloor_sessions + WHERE odoo_workorder_id IN (87, 810) +) +SELECT + workorder_87_ok, + workorder_810_ok, + ( + workorder_87_ok = 1 + AND workorder_810_ok = 1 + ) AS known_backfill_ok +FROM expected; + +\echo '=== 6. Contextos paralelos ===' + +SELECT + context_session_id, + machine_id, + source_type, + source_ref, + external_order_ref, + odoo_workorder_id, + operator_employee_id, + operator_candidate_employee_id, + effective_from, + effective_to, + status, + confidence, + context_priority, + context_time_usable, + individual_attribution_eligible, + official_eligible +FROM + mv_reports_ucepsa_prod + .li_odoo_shopfloor_context_intervals_v1 +WHERE odoo_workorder_id IN (87, 810) +ORDER BY odoo_workorder_id; + +\echo '=== 7. Cola de revisión SHADOW ===' + +SELECT + review_rank, + machine_id, + production_order, + odoo_workorder_id, + operator_candidate_employee_id, + operator_candidate_name, + operator_confirmation_status, + declared_quantity_candidate, + quantity_confirmation_status, + confidence +FROM + mv_reports_ucepsa_prod + .li_odoo_shopfloor_session_review_queue_v1 +ORDER BY review_rank +LIMIT 50; + +\echo '=== 8. Eventos consecutivos duplicados ===' + +WITH ordered AS ( + SELECT + event_id, + odoo_workorder_id, + state_hash, + lag(state_hash) OVER ( + PARTITION BY tenant, site, odoo_workorder_id + ORDER BY event_id + ) AS previous_hash + FROM mv_hot.odoo_shopfloor_workorder_events +) +SELECT count(*) AS consecutive_duplicate_event_count +FROM ordered +WHERE state_hash = previous_hash; + +\echo '=== 9. No se han insertado contextos oficiales ===' + +SELECT count(*) AS generated_production_context_rows +FROM mv_loss_intelligence.production_context_sessions +WHERE source_type = 'ODOO_SHOPFLOOR' + AND evidence_json ->> 'source' = + 'odoo_shopfloor_session_sync'; + +\echo '=== 10. Cola oficial sigue vacía ===' + +SELECT count(*) AS official_operator_queue_rows +FROM mv_reports_ucepsa_prod.li_operator_stop_queue_v2; + +\echo '=== 11. Cobertura ===' + +SELECT * +FROM + mv_reports_ucepsa_prod + .li_odoo_shopfloor_session_coverage_v1 +ORDER BY machine_id; + +\echo '=== 12. Sincronización ===' + +SELECT + tenant, + site, + sync_name, + last_success_at, + last_odoo_cursor_at, + last_processed_count, + last_error, + updated_at +FROM mv_loss_intelligence.odoo_shopfloor_sync_state +ORDER BY tenant, site, sync_name; diff --git a/ucepsa/edge-oee-demo/sql/versions/104_ucepsa_odoo_shopfloor_sessions_shadow_v033.sql b/ucepsa/edge-oee-demo/sql/versions/104_ucepsa_odoo_shopfloor_sessions_shadow_v033.sql new file mode 100644 index 0000000..80a4a3e --- /dev/null +++ b/ucepsa/edge-oee-demo/sql/versions/104_ucepsa_odoo_shopfloor_sessions_shadow_v033.sql @@ -0,0 +1,613 @@ +BEGIN; + +CREATE SCHEMA IF NOT EXISTS mv_hot; +CREATE SCHEMA IF NOT EXISTS mv_loss_intelligence; +CREATE SCHEMA IF NOT EXISTS mv_reports_ucepsa_prod; + +CREATE TABLE IF NOT EXISTS +mv_hot.odoo_shopfloor_workorder_current ( + tenant text NOT NULL, + site text NOT NULL, + machine_id text NOT NULL, + workcenter_id bigint NOT NULL, + workcenter_name text, + odoo_workorder_id bigint NOT NULL, + workorder_name text, + workorder_state text, + odoo_production_id bigint, + production_order text, + production_chain_key text, + production_state text, + origin text, + product_id bigint, + product_name text, + product_uom_id bigint, + product_uom_name text, + planned_quantity numeric(18, 4), + qty_producing numeric(18, 4), + qty_produced numeric(18, 4), + workorder_started_at timestamptz, + workorder_ended_at timestamptz, + production_started_at timestamptz, + production_ended_at timestamptz, + productivity_employee_ids bigint[] NOT NULL + DEFAULT '{}'::bigint[], + assigned_employee_ids bigint[] NOT NULL + DEFAULT '{}'::bigint[], + technical_employee_ids bigint[] NOT NULL + DEFAULT '{}'::bigint[], + output_move_ids bigint[] NOT NULL + DEFAULT '{}'::bigint[], + output_done_quantity numeric(18, 4), + state_hash text NOT NULL, + payload_json jsonb NOT NULL, + odoo_write_date timestamptz, + observed_at timestamptz NOT NULL DEFAULT now(), + created_at timestamptz NOT NULL DEFAULT now(), + updated_at timestamptz NOT NULL DEFAULT now(), + PRIMARY KEY ( + tenant, + site, + odoo_workorder_id + ) +); + +CREATE INDEX IF NOT EXISTS +idx_odoo_shopfloor_current_machine_state +ON mv_hot.odoo_shopfloor_workorder_current ( + tenant, + site, + machine_id, + workorder_state +); + +CREATE INDEX IF NOT EXISTS +idx_odoo_shopfloor_current_write_date +ON mv_hot.odoo_shopfloor_workorder_current ( + tenant, + site, + odoo_write_date +); + +CREATE TABLE IF NOT EXISTS +mv_hot.odoo_shopfloor_workorder_events ( + event_id bigserial PRIMARY KEY, + tenant text NOT NULL, + site text NOT NULL, + machine_id text NOT NULL, + workcenter_id bigint NOT NULL, + odoo_workorder_id bigint NOT NULL, + odoo_production_id bigint, + production_order text, + production_chain_key text, + event_kind text NOT NULL, + effective_event_at timestamptz, + observed_at timestamptz NOT NULL DEFAULT now(), + state_hash text NOT NULL, + payload_json jsonb NOT NULL, + created_at timestamptz NOT NULL DEFAULT now() +); + +CREATE INDEX IF NOT EXISTS +idx_odoo_shopfloor_events_workorder_last +ON mv_hot.odoo_shopfloor_workorder_events ( + tenant, + site, + odoo_workorder_id, + event_id DESC +); + +CREATE INDEX IF NOT EXISTS +idx_odoo_shopfloor_events_machine_effective +ON mv_hot.odoo_shopfloor_workorder_events ( + tenant, + site, + machine_id, + effective_event_at +); + +CREATE TABLE IF NOT EXISTS +mv_loss_intelligence.odoo_shopfloor_technical_employees ( + tenant text NOT NULL, + site text NOT NULL, + employee_id bigint NOT NULL, + employee_name text, + reason text NOT NULL, + active boolean NOT NULL DEFAULT true, + created_at timestamptz NOT NULL DEFAULT now(), + updated_at timestamptz NOT NULL DEFAULT now(), + PRIMARY KEY ( + tenant, + site, + employee_id + ) +); + +INSERT INTO +mv_loss_intelligence.odoo_shopfloor_technical_employees ( + tenant, + site, + employee_id, + employee_name, + reason, + active +) +VALUES ( + 'ucepsa', + 'ucepsa_onpremise', + 35, + 'CORTADORAS', + 'Perfil técnico compartido de la tablet; no representa a una operaria.', + true +) +ON CONFLICT ( + tenant, + site, + employee_id +) +DO UPDATE SET + employee_name = EXCLUDED.employee_name, + reason = EXCLUDED.reason, + active = EXCLUDED.active, + updated_at = now(); + +CREATE TABLE IF NOT EXISTS +mv_loss_intelligence.odoo_shopfloor_sessions ( + shopfloor_session_id bigserial PRIMARY KEY, + tenant text NOT NULL, + site text NOT NULL, + session_key text NOT NULL, + production_chain_key text, + machine_id text NOT NULL, + workcenter_id bigint NOT NULL, + workcenter_name text, + odoo_production_id bigint, + production_order text, + odoo_workorder_id bigint NOT NULL, + workorder_name text, + workorder_state text, + production_state text, + origin text, + product_id bigint, + product_name text, + product_uom_id bigint, + product_uom_name text, + started_at timestamptz, + ended_at timestamptz, + session_status text NOT NULL, + operator_employee_id bigint, + operator_name text, + operator_candidate_employee_id bigint, + operator_candidate_name text, + operator_resolution text NOT NULL, + operator_confidence text NOT NULL, + operator_confirmation_status text NOT NULL, + planned_quantity numeric(18, 4), + declared_quantity numeric(18, 4), + declared_quantity_candidate numeric(18, 4), + posted_quantity numeric(18, 4), + remaining_planned_quantity numeric(18, 4), + quantity_status text NOT NULL, + quantity_confirmation_status text NOT NULL, + confidence text NOT NULL, + official_eligible boolean NOT NULL DEFAULT false, + evidence_json jsonb NOT NULL, + reviewed_by text, + reviewed_at timestamptz, + review_notes text, + source_observed_at timestamptz, + source_odoo_write_date timestamptz, + created_at timestamptz NOT NULL DEFAULT now(), + updated_at timestamptz NOT NULL DEFAULT now(), + CONSTRAINT ux_odoo_shopfloor_sessions_workorder + UNIQUE ( + tenant, + site, + odoo_workorder_id + ), + CONSTRAINT ux_odoo_shopfloor_sessions_key + UNIQUE ( + tenant, + site, + session_key + ), + CONSTRAINT ck_odoo_shopfloor_session_interval + CHECK ( + ended_at IS NULL + OR started_at IS NULL + OR ended_at >= started_at + ), + CONSTRAINT ck_odoo_shopfloor_session_status + CHECK ( + session_status IN ( + 'OPEN', + 'CLOSED', + 'PENDING' + ) + ), + CONSTRAINT ck_odoo_shopfloor_operator_confidence + CHECK ( + operator_confidence IN ( + 'HIGH', + 'MEDIUM', + 'LOW' + ) + ), + CONSTRAINT ck_odoo_shopfloor_operator_confirmation + CHECK ( + operator_confirmation_status IN ( + 'EVIDENCE_CONFIRMED', + 'PENDING_HUMAN', + 'CONFIRMED_HUMAN', + 'UNRESOLVED', + 'REJECTED' + ) + ), + CONSTRAINT ck_odoo_shopfloor_quantity_status + CHECK ( + quantity_status IN ( + 'FINAL_POSTED', + 'POSTED_PARTIAL', + 'PROVISIONAL_TO_CLOSE', + 'PENDING', + 'NOT_APPLICABLE' + ) + ), + CONSTRAINT ck_odoo_shopfloor_quantity_confirmation + CHECK ( + quantity_confirmation_status IN ( + 'EVIDENCE_CONFIRMED', + 'PENDING_HUMAN', + 'CONFIRMED_HUMAN', + 'UNRESOLVED' + ) + ), + CONSTRAINT ck_odoo_shopfloor_confidence + CHECK ( + confidence IN ( + 'HIGH', + 'MEDIUM', + 'LOW' + ) + ), + CONSTRAINT ck_odoo_shopfloor_shadow_only + CHECK (official_eligible = false) +); + +CREATE INDEX IF NOT EXISTS +idx_odoo_shopfloor_sessions_machine_time +ON mv_loss_intelligence.odoo_shopfloor_sessions ( + tenant, + site, + machine_id, + started_at, + ended_at +); + +CREATE INDEX IF NOT EXISTS +idx_odoo_shopfloor_sessions_review +ON mv_loss_intelligence.odoo_shopfloor_sessions ( + tenant, + site, + operator_confirmation_status, + quantity_confirmation_status +); + +CREATE TABLE IF NOT EXISTS +mv_loss_intelligence.odoo_shopfloor_session_reviews ( + review_id bigserial PRIMARY KEY, + shopfloor_session_id bigint NOT NULL + REFERENCES + mv_loss_intelligence.odoo_shopfloor_sessions ( + shopfloor_session_id + ) + ON DELETE CASCADE, + review_type text NOT NULL, + previous_values jsonb NOT NULL, + new_values jsonb NOT NULL, + reviewed_by text NOT NULL, + reviewed_at timestamptz NOT NULL DEFAULT now(), + notes text +); + +CREATE INDEX IF NOT EXISTS +idx_odoo_shopfloor_reviews_session +ON mv_loss_intelligence.odoo_shopfloor_session_reviews ( + shopfloor_session_id, + reviewed_at DESC +); + +CREATE TABLE IF NOT EXISTS +mv_loss_intelligence.odoo_shopfloor_sync_state ( + tenant text NOT NULL, + site text NOT NULL, + sync_name text NOT NULL, + last_success_at timestamptz, + last_odoo_cursor_at timestamptz, + last_processed_count integer, + last_error text, + updated_at timestamptz NOT NULL DEFAULT now(), + PRIMARY KEY ( + tenant, + site, + sync_name + ) +); + +CREATE OR REPLACE VIEW +mv_reports_ucepsa_prod.li_odoo_shopfloor_sessions_v1 +AS +SELECT + s.shopfloor_session_id, + s.tenant, + s.site, + s.session_key, + s.production_chain_key, + s.machine_id, + s.workcenter_id, + s.workcenter_name, + s.odoo_production_id, + s.production_order, + s.odoo_workorder_id, + s.workorder_name, + s.workorder_state, + s.production_state, + s.origin, + s.product_id, + s.product_name, + s.product_uom_id, + s.product_uom_name, + s.started_at, + s.ended_at, + s.session_status, + s.operator_employee_id, + s.operator_name, + s.operator_candidate_employee_id, + s.operator_candidate_name, + s.operator_resolution, + s.operator_confidence, + s.operator_confirmation_status, + s.planned_quantity, + s.declared_quantity, + s.declared_quantity_candidate, + s.posted_quantity, + s.remaining_planned_quantity, + s.quantity_status, + s.quantity_confirmation_status, + s.confidence, + s.official_eligible, + ( + s.started_at IS NOT NULL + AND ( + s.session_status = 'OPEN' + OR s.ended_at IS NOT NULL + ) + ) AS context_time_usable, + ( + s.operator_employee_id IS NOT NULL + AND s.operator_confirmation_status IN ( + 'EVIDENCE_CONFIRMED', + 'CONFIRMED_HUMAN' + ) + ) AS individual_attribution_eligible, + s.evidence_json, + s.reviewed_by, + s.reviewed_at, + s.review_notes, + s.source_observed_at, + s.source_odoo_write_date, + s.created_at, + s.updated_at +FROM mv_loss_intelligence.odoo_shopfloor_sessions s; + +COMMENT ON VIEW +mv_reports_ucepsa_prod.li_odoo_shopfloor_sessions_v1 IS +'Sesiones canónicas Odoo Shop Floor en SHADOW. Tiempo de sesión y atribución individual se gobiernan por separado.'; + +CREATE OR REPLACE VIEW +mv_reports_ucepsa_prod.li_odoo_shopfloor_context_intervals_v1 +AS +SELECT + s.shopfloor_session_id AS context_session_id, + s.tenant, + s.site, + s.machine_id, + 'ODOO_SHOPFLOOR'::text AS source_type, + s.session_key AS source_ref, + s.production_order AS external_order_ref, + s.odoo_production_id, + s.odoo_workorder_id, + s.product_id, + NULL::text AS product_default_code, + s.product_name, + s.operator_employee_id, + s.operator_name, + s.operator_candidate_employee_id, + s.operator_candidate_name, + s.started_at AS effective_from, + s.ended_at AS effective_to, + s.session_status AS status, + 'AUTHORIZED'::text AS authorization_status, + s.confidence, + false AS official_eligible, + 100::integer AS context_priority, + s.context_time_usable, + s.individual_attribution_eligible, + s.operator_resolution, + s.operator_confirmation_status, + s.quantity_status, + s.quantity_confirmation_status, + s.production_chain_key, + s.evidence_json +FROM mv_reports_ucepsa_prod.li_odoo_shopfloor_sessions_v1 s +WHERE s.context_time_usable; + +COMMENT ON VIEW +mv_reports_ucepsa_prod.li_odoo_shopfloor_context_intervals_v1 IS +'Contextos paralelos ODOO_SHOPFLOOR. No sustituyen todavía las vistas v1/v0.3.1 ni alimentan la cola oficial.'; + +CREATE OR REPLACE VIEW +mv_reports_ucepsa_prod.li_odoo_shopfloor_session_review_queue_v1 +AS +SELECT + row_number() OVER ( + ORDER BY + CASE + WHEN s.session_status = 'OPEN' THEN 0 + ELSE 1 + END, + COALESCE( + s.ended_at, + s.started_at, + s.updated_at + ) DESC, + s.shopfloor_session_id DESC + ) AS review_rank, + s.* +FROM mv_reports_ucepsa_prod.li_odoo_shopfloor_sessions_v1 s +WHERE s.operator_confirmation_status IN ( + 'PENDING_HUMAN', + 'UNRESOLVED', + 'REJECTED' + ) + OR s.quantity_confirmation_status IN ( + 'PENDING_HUMAN', + 'UNRESOLVED' + ); + +COMMENT ON VIEW +mv_reports_ucepsa_prod.li_odoo_shopfloor_session_review_queue_v1 IS +'Cola SHADOW de sesiones con operaria o cantidad pendiente. No es una cola para operarias ni modifica Odoo.'; + +CREATE OR REPLACE VIEW +mv_reports_ucepsa_prod.li_odoo_shopfloor_session_coverage_v1 +AS +SELECT + s.tenant, + s.site, + s.machine_id, + count(*) AS session_count, + count(*) FILTER ( + WHERE s.session_status = 'OPEN' + ) AS open_session_count, + count(*) FILTER ( + WHERE s.session_status = 'CLOSED' + ) AS closed_session_count, + count(*) FILTER ( + WHERE s.operator_confirmation_status IN ( + 'EVIDENCE_CONFIRMED', + 'CONFIRMED_HUMAN' + ) + ) AS operator_confirmed_count, + count(*) FILTER ( + WHERE s.operator_confirmation_status = + 'PENDING_HUMAN' + ) AS operator_pending_count, + count(*) FILTER ( + WHERE s.quantity_confirmation_status IN ( + 'EVIDENCE_CONFIRMED', + 'CONFIRMED_HUMAN' + ) + ) AS quantity_confirmed_count, + count(*) FILTER ( + WHERE s.quantity_confirmation_status = + 'PENDING_HUMAN' + ) AS quantity_pending_count, + round( + sum( + EXTRACT( + EPOCH FROM ( + COALESCE(s.ended_at, now()) + - s.started_at + ) + ) + ) FILTER ( + WHERE s.started_at IS NOT NULL + )::numeric / 3600, + 2 + ) AS covered_hours +FROM mv_loss_intelligence.odoo_shopfloor_sessions s +GROUP BY + s.tenant, + s.site, + s.machine_id; + +COMMENT ON VIEW +mv_reports_ucepsa_prod.li_odoo_shopfloor_session_coverage_v1 IS +'Cobertura de sesiones Odoo Shop Floor por máquina y estado de resolución.'; + +CREATE OR REPLACE VIEW +mv_reports_ucepsa_prod.li_odoo_shopfloor_vs_order_context_v1 +AS +SELECT + s.shopfloor_session_id, + s.machine_id, + s.production_order, + s.odoo_workorder_id, + s.started_at, + s.ended_at, + s.session_status, + s.operator_confirmation_status, + s.quantity_confirmation_status, + count(o.id) AS overlapping_order_interval_count, + round( + COALESCE( + sum( + EXTRACT( + EPOCH FROM ( + LEAST( + COALESCE(s.ended_at, now()), + COALESCE(o.effective_to, now()) + ) + - GREATEST( + s.started_at, + o.effective_from + ) + ) + ) + ) FILTER ( + WHERE o.id IS NOT NULL + AND LEAST( + COALESCE(s.ended_at, now()), + COALESCE(o.effective_to, now()) + ) > GREATEST( + s.started_at, + o.effective_from + ) + ), + 0 + )::numeric, + 2 + ) AS overlapping_order_seconds, + count(*) FILTER ( + WHERE o.id IS NOT NULL + AND o.workorder_id IS DISTINCT FROM + s.odoo_workorder_id::text + ) AS mismatching_workorder_intervals +FROM mv_reports_ucepsa_prod.li_odoo_shopfloor_sessions_v1 s +LEFT JOIN +mv_reports_ucepsa_prod.li_order_state_intervals_v1 o + ON o.tenant = s.tenant + AND o.site = s.site + AND o.machine_id = s.machine_id + AND o.effective_from < + COALESCE(s.ended_at, now()) + AND COALESCE(o.effective_to, now()) > + s.started_at +WHERE s.started_at IS NOT NULL +GROUP BY + s.shopfloor_session_id, + s.machine_id, + s.production_order, + s.odoo_workorder_id, + s.started_at, + s.ended_at, + s.session_status, + s.operator_confirmation_status, + s.quantity_confirmation_status; + +COMMENT ON VIEW +mv_reports_ucepsa_prod.li_odoo_shopfloor_vs_order_context_v1 IS +'Diagnóstico paralelo entre sesiones Shop Floor y el histórico order_state. No altera la atribución de paros.'; + +COMMIT; diff --git a/ucepsa/edge-oee-demo/tools/odoo_shopfloor_session_admin.py b/ucepsa/edge-oee-demo/tools/odoo_shopfloor_session_admin.py new file mode 100644 index 0000000..e1fd274 --- /dev/null +++ b/ucepsa/edge-oee-demo/tools/odoo_shopfloor_session_admin.py @@ -0,0 +1,529 @@ +#!/usr/bin/env python3 +"""Revisión humana de sesiones Odoo Shop Floor SHADOW.""" + +from __future__ import annotations + +import argparse +import json +import os +import sys +from decimal import Decimal +from typing import Any + +import psycopg2 +from psycopg2.extras import Json, RealDictCursor + + +def env_first(*names: str, default: str | None = None) -> str | None: + for name in names: + value = os.getenv(name) + if value not in (None, ""): + return value + return default + + +def connect_pg(): + database = env_first("PGDATABASE", "POSTGRES_DB") + user = env_first("PGUSER", "POSTGRES_USER") + password = env_first("PGPASSWORD", "POSTGRES_PASSWORD") + if not database or not user or not password: + raise RuntimeError("Faltan credenciales PostgreSQL") + connection = psycopg2.connect( + host=env_first( + "PGHOST", + default="mv_ucepsa_postgres_hot", + ), + port=int(env_first("PGPORT", default="5432") or "5432"), + dbname=database, + user=user, + password=password, + ) + connection.autocommit = False + return connection + + +def tenant_site() -> tuple[str, str]: + return ( + env_first("TENANT", "TENANT_OVERRIDE", default="ucepsa") + or "ucepsa", + env_first( + "SITE", + "SITE_OVERRIDE", + default="ucepsa_onpremise", + ) + or "ucepsa_onpremise", + ) + + +def load_session(cur, workorder_id: int) -> dict[str, Any]: + tenant, site = tenant_site() + cur.execute( + """ + SELECT * + FROM mv_loss_intelligence.odoo_shopfloor_sessions + WHERE tenant = %s + AND site = %s + AND odoo_workorder_id = %s + FOR UPDATE + """, + (tenant, site, workorder_id), + ) + row = cur.fetchone() + if not row: + raise RuntimeError( + f"No existe sesión para workorder {workorder_id}" + ) + return dict(row) + + +def recalculate_confidence(row: dict[str, Any]) -> str: + interval_ok = ( + row.get("started_at") is not None + and ( + row.get("session_status") == "OPEN" + or row.get("ended_at") is not None + ) + ) + operator_ok = row.get( + "operator_confirmation_status" + ) in { + "EVIDENCE_CONFIRMED", + "CONFIRMED_HUMAN", + } + quantity_ok = row.get( + "quantity_confirmation_status" + ) in { + "EVIDENCE_CONFIRMED", + "CONFIRMED_HUMAN", + } + if interval_ok and operator_ok and quantity_ok: + return "HIGH" + if interval_ok: + return "MEDIUM" + return "LOW" + + +def audit( + cur, + session_id: int, + review_type: str, + before: dict[str, Any], + after: dict[str, Any], + reviewed_by: str, + notes: str | None, +) -> None: + cur.execute( + """ + INSERT INTO + mv_loss_intelligence.odoo_shopfloor_session_reviews ( + shopfloor_session_id, + review_type, + previous_values, + new_values, + reviewed_by, + notes + ) + VALUES (%s, %s, %s, %s, %s, %s) + """, + ( + session_id, + review_type, + Json(before), + Json(after), + reviewed_by, + notes, + ), + ) + + +def command_list(connection, args: argparse.Namespace) -> None: + tenant, site = tenant_site() + where = "" + params: list[Any] = [tenant, site] + if args.pending: + where = """ + AND ( + operator_confirmation_status IN ( + 'PENDING_HUMAN', + 'UNRESOLVED', + 'REJECTED' + ) + OR quantity_confirmation_status IN ( + 'PENDING_HUMAN', + 'UNRESOLVED' + ) + ) + """ + with connection.cursor(cursor_factory=RealDictCursor) as cur: + cur.execute( + f""" + SELECT + shopfloor_session_id, + machine_id, + production_order, + odoo_workorder_id, + started_at, + ended_at, + operator_employee_id, + operator_name, + operator_candidate_employee_id, + operator_candidate_name, + operator_resolution, + operator_confirmation_status, + declared_quantity, + declared_quantity_candidate, + posted_quantity, + quantity_status, + quantity_confirmation_status, + confidence, + official_eligible + FROM + mv_loss_intelligence.odoo_shopfloor_sessions + WHERE tenant = %s + AND site = %s + {where} + ORDER BY + COALESCE(ended_at, started_at, updated_at) DESC, + shopfloor_session_id DESC + LIMIT %s + """, + params + [args.limit], + ) + rows = cur.fetchall() + print( + json.dumps( + rows, + ensure_ascii=False, + indent=2, + default=str, + ) + ) + + +def command_confirm_operator( + connection, + args: argparse.Namespace, +) -> None: + with connection.cursor(cursor_factory=RealDictCursor) as cur: + row = load_session(cur, args.workorder_id) + before = { + key: row.get(key) + for key in ( + "operator_employee_id", + "operator_name", + "operator_candidate_employee_id", + "operator_candidate_name", + "operator_resolution", + "operator_confidence", + "operator_confirmation_status", + "confidence", + ) + } + + employee_id = args.employee_id + employee_name = args.employee_name + if args.use_candidate: + employee_id = row.get( + "operator_candidate_employee_id" + ) + employee_name = row.get( + "operator_candidate_name" + ) + + if employee_id is None: + raise RuntimeError( + "No existe empleado candidato y no se indicó " + "--employee-id." + ) + + row.update( + { + "operator_employee_id": int(employee_id), + "operator_name": employee_name, + "operator_candidate_employee_id": None, + "operator_candidate_name": None, + "operator_resolution": "HUMAN_CONFIRMED", + "operator_confidence": "HIGH", + "operator_confirmation_status": "CONFIRMED_HUMAN", + } + ) + row["confidence"] = recalculate_confidence(row) + + cur.execute( + """ + UPDATE + mv_loss_intelligence.odoo_shopfloor_sessions + SET + operator_employee_id = %s, + operator_name = %s, + operator_candidate_employee_id = NULL, + operator_candidate_name = NULL, + operator_resolution = 'HUMAN_CONFIRMED', + operator_confidence = 'HIGH', + operator_confirmation_status = + 'CONFIRMED_HUMAN', + confidence = %s, + reviewed_by = %s, + reviewed_at = now(), + review_notes = %s, + updated_at = now() + WHERE shopfloor_session_id = %s + """, + ( + int(employee_id), + employee_name, + row["confidence"], + args.confirmed_by, + args.notes, + row["shopfloor_session_id"], + ), + ) + + after = { + key: row.get(key) + for key in before + } + audit( + cur, + row["shopfloor_session_id"], + "CONFIRM_OPERATOR", + before, + after, + args.confirmed_by, + args.notes, + ) + connection.commit() + print(json.dumps(after, ensure_ascii=False, indent=2, default=str)) + + +def command_confirm_quantity( + connection, + args: argparse.Namespace, +) -> None: + with connection.cursor(cursor_factory=RealDictCursor) as cur: + row = load_session(cur, args.workorder_id) + before = { + key: row.get(key) + for key in ( + "declared_quantity", + "declared_quantity_candidate", + "quantity_status", + "quantity_confirmation_status", + "remaining_planned_quantity", + "confidence", + ) + } + + quantity = ( + Decimal(str(args.quantity)) + if args.quantity is not None + else row.get("declared_quantity_candidate") + ) + if quantity is None: + raise RuntimeError( + "No existe cantidad candidata y no se indicó " + "--quantity." + ) + + planned = row.get("planned_quantity") + remaining = None + if planned is not None: + remaining = max( + Decimal(str(planned)) - Decimal(str(quantity)), + Decimal("0"), + ) + + row.update( + { + "declared_quantity": quantity, + "declared_quantity_candidate": None, + "quantity_status": "FINAL_POSTED", + "quantity_confirmation_status": "CONFIRMED_HUMAN", + "remaining_planned_quantity": remaining, + } + ) + row["confidence"] = recalculate_confidence(row) + + cur.execute( + """ + UPDATE + mv_loss_intelligence.odoo_shopfloor_sessions + SET + declared_quantity = %s, + declared_quantity_candidate = NULL, + quantity_status = 'FINAL_POSTED', + quantity_confirmation_status = + 'CONFIRMED_HUMAN', + remaining_planned_quantity = %s, + confidence = %s, + reviewed_by = %s, + reviewed_at = now(), + review_notes = %s, + updated_at = now() + WHERE shopfloor_session_id = %s + """, + ( + quantity, + remaining, + row["confidence"], + args.confirmed_by, + args.notes, + row["shopfloor_session_id"], + ), + ) + + after = { + key: row.get(key) + for key in before + } + audit( + cur, + row["shopfloor_session_id"], + "CONFIRM_QUANTITY", + before, + after, + args.confirmed_by, + args.notes, + ) + connection.commit() + print(json.dumps(after, ensure_ascii=False, indent=2, default=str)) + + +def command_reject_operator( + connection, + args: argparse.Namespace, +) -> None: + with connection.cursor(cursor_factory=RealDictCursor) as cur: + row = load_session(cur, args.workorder_id) + before = { + key: row.get(key) + for key in ( + "operator_employee_id", + "operator_name", + "operator_candidate_employee_id", + "operator_candidate_name", + "operator_resolution", + "operator_confidence", + "operator_confirmation_status", + "confidence", + ) + } + row.update( + { + "operator_employee_id": None, + "operator_name": None, + "operator_candidate_employee_id": None, + "operator_candidate_name": None, + "operator_resolution": "HUMAN_REJECTED", + "operator_confidence": "LOW", + "operator_confirmation_status": "REJECTED", + } + ) + row["confidence"] = recalculate_confidence(row) + + cur.execute( + """ + UPDATE + mv_loss_intelligence.odoo_shopfloor_sessions + SET + operator_employee_id = NULL, + operator_name = NULL, + operator_candidate_employee_id = NULL, + operator_candidate_name = NULL, + operator_resolution = 'HUMAN_REJECTED', + operator_confidence = 'LOW', + operator_confirmation_status = 'REJECTED', + confidence = %s, + reviewed_by = %s, + reviewed_at = now(), + review_notes = %s, + updated_at = now() + WHERE shopfloor_session_id = %s + """, + ( + row["confidence"], + args.confirmed_by, + args.notes, + row["shopfloor_session_id"], + ), + ) + after = { + key: row.get(key) + for key in before + } + audit( + cur, + row["shopfloor_session_id"], + "REJECT_OPERATOR_CANDIDATE", + before, + after, + args.confirmed_by, + args.notes, + ) + connection.commit() + print(json.dumps(after, ensure_ascii=False, indent=2, default=str)) + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + description=( + "Administra revisiones humanas de sesiones " + "Odoo Shop Floor SHADOW." + ) + ) + sub = parser.add_subparsers(dest="command", required=True) + + list_parser = sub.add_parser("list") + list_parser.add_argument("--pending", action="store_true") + list_parser.add_argument("--limit", type=int, default=100) + + operator = sub.add_parser("confirm-operator") + operator.add_argument("--workorder-id", type=int, required=True) + operator.add_argument("--employee-id", type=int) + operator.add_argument("--employee-name") + operator.add_argument("--use-candidate", action="store_true") + operator.add_argument("--confirmed-by", required=True) + operator.add_argument("--notes") + + quantity = sub.add_parser("confirm-quantity") + quantity.add_argument("--workorder-id", type=int, required=True) + quantity.add_argument("--quantity", type=Decimal) + quantity.add_argument("--confirmed-by", required=True) + quantity.add_argument("--notes") + + reject = sub.add_parser("reject-operator-candidate") + reject.add_argument("--workorder-id", type=int, required=True) + reject.add_argument("--confirmed-by", required=True) + reject.add_argument("--notes") + + return parser + + +def main() -> int: + args = build_parser().parse_args() + connection = connect_pg() + try: + if args.command == "list": + command_list(connection, args) + elif args.command == "confirm-operator": + command_confirm_operator(connection, args) + elif args.command == "confirm-quantity": + command_confirm_quantity(connection, args) + elif args.command == "reject-operator-candidate": + command_reject_operator(connection, args) + else: + raise RuntimeError(f"Comando no soportado: {args.command}") + return 0 + except Exception as exc: + connection.rollback() + print( + f"ERROR {type(exc).__name__}: {exc}", + file=sys.stderr, + ) + return 1 + finally: + connection.close() + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/ucepsa/edge-oee-demo/tools/odoo_shopfloor_session_sync.py b/ucepsa/edge-oee-demo/tools/odoo_shopfloor_session_sync.py new file mode 100644 index 0000000..a74a015 --- /dev/null +++ b/ucepsa/edge-oee-demo/tools/odoo_shopfloor_session_sync.py @@ -0,0 +1,1866 @@ +#!/usr/bin/env python3 +"""Sincroniza sesiones Odoo Shop Floor hacia PostgreSQL en modo SHADOW. + +No escribe en Odoo. No cambia balizas. No alimenta la cola oficial. +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import re +import signal +import sys +import time +import xmlrpc.client +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone +from decimal import Decimal +from typing import Any, Iterable + +import psycopg2 +from psycopg2.extras import Json, RealDictCursor + + +VERSION = "0.3.3" +SYNC_NAME = "odoo_shopfloor_sessions_v033" +RUNNING = True + + +def stop_handler(signum: int, frame: Any) -> None: + del signum, frame + global RUNNING + RUNNING = False + + +signal.signal(signal.SIGTERM, stop_handler) +signal.signal(signal.SIGINT, stop_handler) + + +def env_first(*names: str, default: str | None = None) -> str | None: + for name in names: + value = os.getenv(name) + if value not in (None, ""): + return value + return default + + +def parse_odoo_datetime(value: Any) -> datetime | None: + if value in (None, False, ""): + return None + text = str(value).strip() + try: + parsed = datetime.fromisoformat(text.replace("Z", "+00:00")) + except ValueError: + parsed = datetime.strptime(text, "%Y-%m-%d %H:%M:%S") + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=timezone.utc) + return parsed.astimezone(timezone.utc) + + +def odoo_datetime_string(value: datetime) -> str: + return value.astimezone(timezone.utc).strftime("%Y-%m-%d %H:%M:%S") + + +def m2o_id(value: Any) -> int | None: + if isinstance(value, (list, tuple)) and value: + return int(value[0]) + if isinstance(value, int): + return value + return None + + +def m2o_name(value: Any) -> str | None: + if isinstance(value, (list, tuple)) and len(value) > 1: + return str(value[1]) + return None + + +def decimal_or_none(value: Any) -> Decimal | None: + if value in (None, False, ""): + return None + try: + return Decimal(str(value)) + except Exception: + return None + + +def decimal_max(*values: Decimal | None) -> Decimal | None: + present = [value for value in values if value is not None] + return max(present) if present else None + + +def decimal_sum(values: Iterable[Decimal | None]) -> Decimal: + total = Decimal("0") + for value in values: + if value is not None: + total += value + return total + + +def canonical_hash(payload: dict[str, Any]) -> str: + raw = json.dumps( + payload, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + default=str, + ) + return hashlib.sha256(raw.encode("utf-8")).hexdigest() + + +def production_chain_key(name: str | None) -> str | None: + if not name: + return None + root = re.sub(r"-\d{3}$", "", name) + return f"ODOO:{root}" + + +def json_safe(value: Any) -> Any: + if isinstance(value, Decimal): + return str(value) + if isinstance(value, datetime): + return value.isoformat() + if isinstance(value, dict): + return {key: json_safe(item) for key, item in value.items()} + if isinstance(value, (list, tuple, set)): + return [json_safe(item) for item in value] + return value + + +@dataclass(frozen=True) +class Config: + tenant: str + site: str + odoo_url: str + odoo_db: str + odoo_user: str + odoo_password: str + pg_host: str + pg_port: int + pg_db: str + pg_user: str + pg_password: str + + +def load_config() -> Config: + required = { + "ODOO_URL": env_first("ODOO_URL"), + "ODOO_DB": env_first("ODOO_DB_OVERRIDE", "ODOO_DB"), + "ODOO_USER": env_first("ODOO_USER"), + "ODOO_PASSWORD": env_first("ODOO_PASSWORD"), + "PGDATABASE": env_first("PGDATABASE", "POSTGRES_DB"), + "PGUSER": env_first("PGUSER", "POSTGRES_USER"), + "PGPASSWORD": env_first("PGPASSWORD", "POSTGRES_PASSWORD"), + } + missing = [name for name, value in required.items() if not value] + if missing: + raise RuntimeError( + "Faltan variables obligatorias: " + ", ".join(missing) + ) + + return Config( + tenant=env_first("TENANT", "TENANT_OVERRIDE", default="ucepsa") + or "ucepsa", + site=env_first( + "SITE", + "SITE_OVERRIDE", + default="ucepsa_onpremise", + ) + or "ucepsa_onpremise", + odoo_url=str(required["ODOO_URL"]).rstrip("/"), + odoo_db=str(required["ODOO_DB"]), + odoo_user=str(required["ODOO_USER"]), + odoo_password=str(required["ODOO_PASSWORD"]), + pg_host=env_first( + "PGHOST", + default="mv_ucepsa_postgres_hot", + ) + or "mv_ucepsa_postgres_hot", + pg_port=int(env_first("PGPORT", default="5432") or "5432"), + pg_db=str(required["PGDATABASE"]), + pg_user=str(required["PGUSER"]), + pg_password=str(required["PGPASSWORD"]), + ) + + +class OdooClient: + def __init__(self, config: Config) -> None: + common = xmlrpc.client.ServerProxy( + f"{config.odoo_url}/xmlrpc/2/common", + allow_none=True, + ) + uid = common.authenticate( + config.odoo_db, + config.odoo_user, + config.odoo_password, + {}, + ) + if not uid: + raise RuntimeError("No se pudo autenticar en Odoo") + + self.config = config + self.uid = uid + self.models = xmlrpc.client.ServerProxy( + f"{config.odoo_url}/xmlrpc/2/object", + allow_none=True, + ) + self._field_cache: dict[str, set[str]] = {} + + def execute( + self, + model: str, + method: str, + args: list[Any], + kwargs: dict[str, Any] | None = None, + ) -> Any: + return self.models.execute_kw( + self.config.odoo_db, + self.uid, + self.config.odoo_password, + model, + method, + args, + kwargs or {}, + ) + + def existing_fields( + self, + model: str, + candidates: list[str], + ) -> list[str]: + if model not in self._field_cache: + metadata = self.execute( + model, + "fields_get", + [], + {"attributes": ["type", "relation"]}, + ) + self._field_cache[model] = set(metadata) + return [ + field + for field in candidates + if field in self._field_cache[model] + ] + + def search_read( + self, + model: str, + domain: list[Any], + fields: list[str], + *, + order: str = "id", + limit: int = 0, + ) -> list[dict[str, Any]]: + kwargs: dict[str, Any] = { + "fields": self.existing_fields(model, fields), + "order": order, + } + if limit: + kwargs["limit"] = limit + return self.execute( + model, + "search_read", + [domain], + kwargs, + ) + + +def connect_pg(config: Config): + connection = psycopg2.connect( + host=config.pg_host, + port=config.pg_port, + dbname=config.pg_db, + user=config.pg_user, + password=config.pg_password, + ) + connection.autocommit = False + return connection + + +def load_machine_map( + connection, + config: Config, + cli_pairs: list[str], +) -> dict[int, str]: + machine_map: dict[int, str] = {} + + with connection.cursor(cursor_factory=RealDictCursor) as cur: + cur.execute( + """ + SELECT + workcenter_id, + machine_id + FROM mv_hot.odoo_order_state_current + WHERE tenant = %s + AND site = %s + AND workcenter_id IS NOT NULL + ORDER BY machine_id + """, + (config.tenant, config.site), + ) + for row in cur.fetchall(): + machine_map[int(row["workcenter_id"])] = str( + row["machine_id"] + ) + + raw_json = env_first("SHOPFLOOR_MACHINE_MAP_JSON") + if raw_json: + parsed = json.loads(raw_json) + if not isinstance(parsed, dict): + raise ValueError( + "SHOPFLOOR_MACHINE_MAP_JSON debe ser un objeto JSON" + ) + for key, value in parsed.items(): + if str(key).isdigit(): + machine_map[int(key)] = str(value) + elif str(value).isdigit(): + machine_map[int(value)] = str(key) + else: + raise ValueError( + "Mapa JSON no reconocido. Use " + '{"54":"CORT-02"} o {"CORT-02":54}.' + ) + + for pair in cli_pairs: + if "=" not in pair: + raise ValueError( + f"Mapa inválido {pair!r}; use WORKCENTER_ID=MACHINE_ID" + ) + left, right = pair.split("=", 1) + machine_map[int(left.strip())] = right.strip() + + if not machine_map: + raise RuntimeError( + "No existe mapa workcenter→máquina. " + "Revise mv_hot.odoo_order_state_current o use " + "--machine-map 54=CORT-02." + ) + return machine_map + + +def load_technical_employees( + connection, + config: Config, +) -> set[int]: + with connection.cursor() as cur: + cur.execute( + """ + SELECT employee_id + FROM + mv_loss_intelligence.odoo_shopfloor_technical_employees + WHERE tenant = %s + AND site = %s + AND active + """, + (config.tenant, config.site), + ) + return {int(row[0]) for row in cur.fetchall()} + + +WORKORDER_FIELDS = [ + "id", + "display_name", + "name", + "state", + "production_id", + "workcenter_id", + "product_id", + "date_start", + "date_finished", + "duration", + "duration_expected", + "time_ids", + "working_user_ids", + "last_working_user_id", + "employee_id", + "employee_ids", + "employee_assigned_ids", + "connected_employee_ids", + "create_uid", + "write_uid", + "create_date", + "write_date", +] + +PRODUCTION_FIELDS = [ + "id", + "name", + "display_name", + "state", + "origin", + "product_id", + "product_qty", + "qty_producing", + "qty_produced", + "product_uom_id", + "date_start", + "date_finished", + "date_deadline", + "backorder_sequence", + "parent_production_id", + "child_production_ids", + "employee_ids", + "workorder_ids", + "create_uid", + "write_uid", + "create_date", + "write_date", +] + +PRODUCTIVITY_FIELDS = [ + "id", + "display_name", + "description", + "workorder_id", + "workcenter_id", + "date_start", + "date_end", + "duration", + "user_id", + "employee_id", + "loss_id", + "loss_type", + "create_uid", + "write_uid", + "create_date", + "write_date", +] + +EMPLOYEE_FIELDS = [ + "id", + "name", + "display_name", + "active", + "user_id", + "create_date", + "write_date", +] + +STOCK_MOVE_FIELDS = [ + "id", + "state", + "product_id", + "product_uom", + "product_uom_qty", + "quantity", + "production_id", + "raw_material_production_id", + "date", + "create_uid", + "write_uid", + "create_date", + "write_date", +] + + +def fetch_source_data( + odoo: OdooClient, + machine_map: dict[int, str], + workorder_ids: list[int], + since: datetime, +) -> dict[str, Any]: + workcenter_ids = sorted(machine_map) + + if workorder_ids: + domain: list[Any] = [("id", "in", workorder_ids)] + else: + domain = [ + "&", + ("workcenter_id", "in", workcenter_ids), + "|", + ("state", "=", "progress"), + ("write_date", ">=", odoo_datetime_string(since)), + ] + + workorders = odoo.search_read( + "mrp.workorder", + domain, + WORKORDER_FIELDS, + order="write_date, id", + limit=1000, + ) + + workorders = [ + row + for row in workorders + if m2o_id(row.get("workcenter_id")) in machine_map + ] + + production_ids = sorted( + { + production_id + for row in workorders + if ( + production_id := m2o_id(row.get("production_id")) + ) + is not None + } + ) + actual_workorder_ids = [int(row["id"]) for row in workorders] + + productions = ( + odoo.search_read( + "mrp.production", + [("id", "in", production_ids)], + PRODUCTION_FIELDS, + order="id", + limit=1000, + ) + if production_ids + else [] + ) + + productivity = ( + odoo.search_read( + "mrp.workcenter.productivity", + [("workorder_id", "in", actual_workorder_ids)], + PRODUCTIVITY_FIELDS, + order="id", + limit=5000, + ) + if actual_workorder_ids + else [] + ) + + employee_ids: set[int] = set() + for row in workorders: + for field in ( + "employee_ids", + "employee_assigned_ids", + "connected_employee_ids", + ): + employee_ids.update(int(item) for item in row.get(field, [])) + employee_id = m2o_id(row.get("employee_id")) + if employee_id is not None: + employee_ids.add(employee_id) + + for row in productivity: + employee_id = m2o_id(row.get("employee_id")) + if employee_id is not None: + employee_ids.add(employee_id) + + employees = ( + odoo.search_read( + "hr.employee", + [("id", "in", sorted(employee_ids))], + EMPLOYEE_FIELDS, + order="id", + limit=5000, + ) + if employee_ids + else [] + ) + + stock_moves = ( + odoo.search_read( + "stock.move", + [ + "|", + ("production_id", "in", production_ids), + ( + "raw_material_production_id", + "in", + production_ids, + ), + ], + STOCK_MOVE_FIELDS, + order="id", + limit=10000, + ) + if production_ids + else [] + ) + + return { + "workorders": workorders, + "productions": productions, + "productivity": productivity, + "employees": employees, + "stock_moves": stock_moves, + } + + +def resolve_operator( + workorder: dict[str, Any], + productivity_rows: list[dict[str, Any]], + employees_by_id: dict[int, dict[str, Any]], + technical_ids: set[int], +) -> dict[str, Any]: + def is_technical(employee_id: int) -> bool: + employee = employees_by_id.get(employee_id) or {} + name = str(employee.get("name") or "").strip().upper() + return ( + employee_id in technical_ids + or name in {"CORTADORAS", "TABLET CORTADORAS"} + ) + + productivity_ids = sorted( + { + employee_id + for row in productivity_rows + if ( + employee_id := m2o_id(row.get("employee_id")) + ) + is not None + and not is_technical(employee_id) + } + ) + + assigned_ids = sorted( + { + int(employee_id) + for employee_id in workorder.get( + "employee_assigned_ids", + [], + ) + if not is_technical(int(employee_id)) + } + ) + + all_assigned_ids = sorted( + { + int(employee_id) + for employee_id in workorder.get( + "employee_assigned_ids", + [], + ) + } + ) + + technical_assigned = sorted( + set(all_assigned_ids) - set(assigned_ids) + ) + + if len(productivity_ids) == 1: + employee_id = productivity_ids[0] + employee = employees_by_id.get(employee_id) or {} + return { + "operator_employee_id": employee_id, + "operator_name": employee.get("name"), + "operator_candidate_employee_id": None, + "operator_candidate_name": None, + "operator_resolution": "PRODUCTIVITY_EMPLOYEE", + "operator_confidence": "HIGH", + "operator_confirmation_status": "EVIDENCE_CONFIRMED", + "productivity_employee_ids": productivity_ids, + "assigned_employee_ids": all_assigned_ids, + "technical_employee_ids": technical_assigned, + } + + if len(productivity_ids) > 1: + return { + "operator_employee_id": None, + "operator_name": None, + "operator_candidate_employee_id": None, + "operator_candidate_name": None, + "operator_resolution": "MULTIPLE_PRODUCTIVITY_EMPLOYEES", + "operator_confidence": "LOW", + "operator_confirmation_status": "UNRESOLVED", + "productivity_employee_ids": productivity_ids, + "assigned_employee_ids": all_assigned_ids, + "technical_employee_ids": technical_assigned, + } + + if len(assigned_ids) == 1: + employee_id = assigned_ids[0] + employee = employees_by_id.get(employee_id) or {} + return { + "operator_employee_id": None, + "operator_name": None, + "operator_candidate_employee_id": employee_id, + "operator_candidate_name": employee.get("name"), + "operator_resolution": ( + "UNIQUE_NON_TECHNICAL_ASSIGNED_EMPLOYEE" + ), + "operator_confidence": "MEDIUM", + "operator_confirmation_status": "PENDING_HUMAN", + "productivity_employee_ids": [], + "assigned_employee_ids": all_assigned_ids, + "technical_employee_ids": technical_assigned, + } + + if len(assigned_ids) > 1: + resolution = "MULTIPLE_NON_TECHNICAL_ASSIGNED_EMPLOYEES" + elif all_assigned_ids: + resolution = "TECHNICAL_PROFILE_ONLY" + else: + resolution = "NO_OPERATOR_EVIDENCE" + + return { + "operator_employee_id": None, + "operator_name": None, + "operator_candidate_employee_id": None, + "operator_candidate_name": None, + "operator_resolution": resolution, + "operator_confidence": "LOW", + "operator_confirmation_status": "UNRESOLVED", + "productivity_employee_ids": [], + "assigned_employee_ids": all_assigned_ids, + "technical_employee_ids": technical_assigned, + } + + +def resolve_quantity( + production: dict[str, Any], + output_moves: list[dict[str, Any]], +) -> dict[str, Any]: + planned = decimal_or_none(production.get("product_qty")) + qty_producing = decimal_or_none( + production.get("qty_producing") + ) + qty_produced = decimal_or_none( + production.get("qty_produced") + ) + + done_output_quantity = decimal_sum( + decimal_or_none(move.get("quantity")) + for move in output_moves + if move.get("state") == "done" + ) + + posted = decimal_max( + qty_produced, + done_output_quantity, + ) + if posted == Decimal("0"): + posted = Decimal("0") + + production_state = str(production.get("state") or "") + + declared: Decimal | None = None + candidate: Decimal | None = None + + if production_state == "done" and posted and posted > 0: + declared = posted + quantity_status = "FINAL_POSTED" + confirmation = "EVIDENCE_CONFIRMED" + elif posted and posted > 0: + declared = posted + quantity_status = "POSTED_PARTIAL" + confirmation = "EVIDENCE_CONFIRMED" + elif ( + production_state == "to_close" + and qty_producing is not None + and qty_producing > 0 + ): + candidate = qty_producing + quantity_status = "PROVISIONAL_TO_CLOSE" + confirmation = "PENDING_HUMAN" + elif planned is None: + quantity_status = "NOT_APPLICABLE" + confirmation = "UNRESOLVED" + else: + quantity_status = "PENDING" + confirmation = "UNRESOLVED" + + basis = declared + if basis is None: + basis = candidate + if basis is None and posted is not None: + basis = posted + + remaining: Decimal | None = None + if planned is not None and basis is not None: + remaining = max(planned - basis, Decimal("0")) + + return { + "planned_quantity": planned, + "declared_quantity": declared, + "declared_quantity_candidate": candidate, + "posted_quantity": posted, + "remaining_planned_quantity": remaining, + "quantity_status": quantity_status, + "quantity_confirmation_status": confirmation, + "qty_producing": qty_producing, + "qty_produced": qty_produced, + "output_done_quantity": done_output_quantity, + "output_move_ids": [ + int(move["id"]) + for move in output_moves + ], + } + + +def build_records( + source: dict[str, Any], + machine_map: dict[int, str], + technical_ids: set[int], +) -> list[dict[str, Any]]: + productions_by_id = { + int(row["id"]): row + for row in source["productions"] + } + employees_by_id = { + int(row["id"]): row + for row in source["employees"] + } + + productivity_by_workorder: dict[int, list[dict[str, Any]]] = {} + for row in source["productivity"]: + workorder_id = m2o_id(row.get("workorder_id")) + if workorder_id is not None: + productivity_by_workorder.setdefault( + workorder_id, + [], + ).append(row) + + moves_by_production: dict[int, list[dict[str, Any]]] = {} + for row in source["stock_moves"]: + production_id = m2o_id(row.get("production_id")) + if production_id is not None: + moves_by_production.setdefault( + production_id, + [], + ).append(row) + + records: list[dict[str, Any]] = [] + + for workorder in source["workorders"]: + workorder_id = int(workorder["id"]) + production_id = m2o_id(workorder.get("production_id")) + workcenter_id = m2o_id(workorder.get("workcenter_id")) + + if production_id is None or workcenter_id is None: + continue + if workcenter_id not in machine_map: + continue + + production = productions_by_id.get(production_id) + if not production: + continue + + machine_id = machine_map[workcenter_id] + product_id = m2o_id(production.get("product_id")) + output_moves = [ + move + for move in moves_by_production.get(production_id, []) + if m2o_id(move.get("product_id")) == product_id + ] + + operator = resolve_operator( + workorder, + productivity_by_workorder.get(workorder_id, []), + employees_by_id, + technical_ids, + ) + quantity = resolve_quantity( + production, + output_moves, + ) + + started_at = parse_odoo_datetime( + workorder.get("date_start") + ) + ended_at = parse_odoo_datetime( + workorder.get("date_finished") + ) + workorder_state = str(workorder.get("state") or "") + + if workorder_state == "done" and started_at and ended_at: + session_status = "CLOSED" + elif workorder_state == "progress" and started_at: + session_status = "OPEN" + ended_at = None + else: + session_status = "PENDING" + + operator_confirmed = ( + operator["operator_confirmation_status"] + == "EVIDENCE_CONFIRMED" + ) + quantity_confirmed = ( + quantity["quantity_confirmation_status"] + == "EVIDENCE_CONFIRMED" + ) + complete_interval = ( + started_at is not None + and ( + session_status == "OPEN" + or ended_at is not None + ) + ) + + if ( + complete_interval + and operator_confirmed + and quantity_confirmed + ): + confidence = "HIGH" + elif complete_interval: + confidence = "MEDIUM" + else: + confidence = "LOW" + + production_order = str(production.get("name") or "") + chain_key = production_chain_key(production_order) + uom_id = m2o_id(production.get("product_uom_id")) + uom_name = m2o_name(production.get("product_uom_id")) + + evidence = { + "schema_version": VERSION, + "source": "odoo_shopfloor_session_sync", + "workorder": workorder, + "production": production, + "productivity": productivity_by_workorder.get( + workorder_id, + [], + ), + "employees": { + str(employee_id): employees_by_id.get(employee_id) + for employee_id in sorted( + set( + operator["productivity_employee_ids"] + + operator["assigned_employee_ids"] + ) + ) + }, + "output_moves": output_moves, + "operator_resolution": operator, + "quantity_resolution": quantity, + "timezone_rule": ( + "Odoo XML-RPC datetimes interpreted as UTC" + ), + "official_eligible": False, + } + + state_payload = { + "machine_id": machine_id, + "workcenter_id": workcenter_id, + "workorder_id": workorder_id, + "workorder_state": workorder_state, + "production_id": production_id, + "production_order": production_order, + "production_state": production.get("state"), + "product_id": product_id, + "started_at": started_at, + "ended_at": ended_at, + "operator": operator, + "quantity": quantity, + "source_odoo_write_date": parse_odoo_datetime( + workorder.get("write_date") + ), + } + state_hash = canonical_hash(json_safe(state_payload)) + + if workorder_state == "done" and ended_at: + effective_event_at = ended_at + elif workorder_state == "progress" and started_at: + effective_event_at = started_at + else: + effective_event_at = parse_odoo_datetime( + workorder.get("write_date") + ) + + records.append( + { + "tenant": None, + "site": None, + "session_key": f"ODOO_WORKORDER:{workorder_id}", + "production_chain_key": chain_key, + "machine_id": machine_id, + "workcenter_id": workcenter_id, + "workcenter_name": m2o_name( + workorder.get("workcenter_id") + ), + "odoo_production_id": production_id, + "production_order": production_order, + "odoo_workorder_id": workorder_id, + "workorder_name": workorder.get("name"), + "workorder_state": workorder_state, + "production_state": production.get("state"), + "origin": production.get("origin"), + "product_id": product_id, + "product_name": m2o_name( + production.get("product_id") + ), + "product_uom_id": uom_id, + "product_uom_name": uom_name, + "started_at": started_at, + "ended_at": ended_at, + "session_status": session_status, + **operator, + **quantity, + "confidence": confidence, + "official_eligible": False, + "evidence_json": evidence, + "source_observed_at": datetime.now(timezone.utc), + "source_odoo_write_date": parse_odoo_datetime( + workorder.get("write_date") + ), + "state_hash": state_hash, + "effective_event_at": effective_event_at, + } + ) + + return records + + +def get_existing_session( + cur, + tenant: str, + site: str, + workorder_id: int, +) -> dict[str, Any] | None: + cur.execute( + """ + SELECT * + FROM mv_loss_intelligence.odoo_shopfloor_sessions + WHERE tenant = %s + AND site = %s + AND odoo_workorder_id = %s + FOR UPDATE + """, + (tenant, site, workorder_id), + ) + return cur.fetchone() + + +def preserve_human_reviews( + record: dict[str, Any], + existing: dict[str, Any] | None, +) -> dict[str, Any]: + if not existing: + return record + + merged = dict(record) + + if ( + existing.get("operator_confirmation_status") + == "CONFIRMED_HUMAN" + ): + for field in ( + "operator_employee_id", + "operator_name", + "operator_confirmation_status", + "reviewed_by", + "reviewed_at", + "review_notes", + ): + merged[field] = existing.get(field) + merged["operator_resolution"] = "HUMAN_CONFIRMED" + merged["operator_confidence"] = "HIGH" + + if ( + existing.get("quantity_confirmation_status") + == "CONFIRMED_HUMAN" + ): + for field in ( + "declared_quantity", + "quantity_confirmation_status", + "reviewed_by", + "reviewed_at", + "review_notes", + ): + merged[field] = existing.get(field) + merged["declared_quantity_candidate"] = None + merged["quantity_status"] = "FINAL_POSTED" + + interval_complete = ( + merged.get("started_at") is not None + and ( + merged.get("session_status") == "OPEN" + or merged.get("ended_at") is not None + ) + ) + operator_final = merged.get( + "operator_confirmation_status" + ) in { + "EVIDENCE_CONFIRMED", + "CONFIRMED_HUMAN", + } + quantity_final = merged.get( + "quantity_confirmation_status" + ) in { + "EVIDENCE_CONFIRMED", + "CONFIRMED_HUMAN", + } + + if interval_complete and operator_final and quantity_final: + merged["confidence"] = "HIGH" + elif interval_complete: + merged["confidence"] = "MEDIUM" + else: + merged["confidence"] = "LOW" + + return merged + + +def upsert_record( + connection, + config: Config, + record: dict[str, Any], +) -> bool: + record = dict(record) + record["tenant"] = config.tenant + record["site"] = config.site + record["evidence_json"] = json_safe(record["evidence_json"]) + + with connection.cursor(cursor_factory=RealDictCursor) as cur: + lock_key = ( + f"{config.tenant}|{config.site}|" + f"{record['odoo_workorder_id']}" + ) + cur.execute( + """ + SELECT pg_advisory_xact_lock( + hashtextextended(%s, 0) + ) + """, + (lock_key,), + ) + + existing = get_existing_session( + cur, + config.tenant, + config.site, + int(record["odoo_workorder_id"]), + ) + final_record = preserve_human_reviews( + record, + existing, + ) + + cur.execute( + """ + SELECT state_hash + FROM mv_hot.odoo_shopfloor_workorder_events + WHERE tenant = %(tenant)s + AND site = %(site)s + AND odoo_workorder_id = + %(odoo_workorder_id)s + ORDER BY event_id DESC + LIMIT 1 + """, + final_record, + ) + last = cur.fetchone() + transition_inserted = ( + last is None + or last["state_hash"] != final_record["state_hash"] + ) + + if transition_inserted: + cur.execute( + """ + INSERT INTO + mv_hot.odoo_shopfloor_workorder_events ( + tenant, + site, + machine_id, + workcenter_id, + odoo_workorder_id, + odoo_production_id, + production_order, + production_chain_key, + event_kind, + effective_event_at, + observed_at, + state_hash, + payload_json + ) + VALUES ( + %(tenant)s, + %(site)s, + %(machine_id)s, + %(workcenter_id)s, + %(odoo_workorder_id)s, + %(odoo_production_id)s, + %(production_order)s, + %(production_chain_key)s, + %(event_kind)s, + %(effective_event_at)s, + %(source_observed_at)s, + %(state_hash)s, + %(payload_json)s + ) + """, + { + **final_record, + "event_kind": ( + "INITIAL_SNAPSHOT" + if last is None + else "STATE_TRANSITION" + ), + "payload_json": Json( + final_record["evidence_json"] + ), + }, + ) + + cur.execute( + """ + INSERT INTO + mv_hot.odoo_shopfloor_workorder_current ( + tenant, + site, + machine_id, + workcenter_id, + workcenter_name, + odoo_workorder_id, + workorder_name, + workorder_state, + odoo_production_id, + production_order, + production_chain_key, + production_state, + origin, + product_id, + product_name, + product_uom_id, + product_uom_name, + planned_quantity, + qty_producing, + qty_produced, + workorder_started_at, + workorder_ended_at, + production_started_at, + production_ended_at, + productivity_employee_ids, + assigned_employee_ids, + technical_employee_ids, + output_move_ids, + output_done_quantity, + state_hash, + payload_json, + odoo_write_date, + observed_at, + updated_at + ) + VALUES ( + %(tenant)s, + %(site)s, + %(machine_id)s, + %(workcenter_id)s, + %(workcenter_name)s, + %(odoo_workorder_id)s, + %(workorder_name)s, + %(workorder_state)s, + %(odoo_production_id)s, + %(production_order)s, + %(production_chain_key)s, + %(production_state)s, + %(origin)s, + %(product_id)s, + %(product_name)s, + %(product_uom_id)s, + %(product_uom_name)s, + %(planned_quantity)s, + %(qty_producing)s, + %(qty_produced)s, + %(started_at)s, + %(ended_at)s, + %(production_started_at)s, + %(production_ended_at)s, + %(productivity_employee_ids)s, + %(assigned_employee_ids)s, + %(technical_employee_ids)s, + %(output_move_ids)s, + %(output_done_quantity)s, + %(state_hash)s, + %(payload_json)s, + %(source_odoo_write_date)s, + %(source_observed_at)s, + now() + ) + ON CONFLICT ( + tenant, + site, + odoo_workorder_id + ) + DO UPDATE SET + machine_id = EXCLUDED.machine_id, + workcenter_id = EXCLUDED.workcenter_id, + workcenter_name = EXCLUDED.workcenter_name, + workorder_name = EXCLUDED.workorder_name, + workorder_state = EXCLUDED.workorder_state, + odoo_production_id = + EXCLUDED.odoo_production_id, + production_order = + EXCLUDED.production_order, + production_chain_key = + EXCLUDED.production_chain_key, + production_state = + EXCLUDED.production_state, + origin = EXCLUDED.origin, + product_id = EXCLUDED.product_id, + product_name = EXCLUDED.product_name, + product_uom_id = EXCLUDED.product_uom_id, + product_uom_name = + EXCLUDED.product_uom_name, + planned_quantity = + EXCLUDED.planned_quantity, + qty_producing = EXCLUDED.qty_producing, + qty_produced = EXCLUDED.qty_produced, + workorder_started_at = + EXCLUDED.workorder_started_at, + workorder_ended_at = + EXCLUDED.workorder_ended_at, + production_started_at = + EXCLUDED.production_started_at, + production_ended_at = + EXCLUDED.production_ended_at, + productivity_employee_ids = + EXCLUDED.productivity_employee_ids, + assigned_employee_ids = + EXCLUDED.assigned_employee_ids, + technical_employee_ids = + EXCLUDED.technical_employee_ids, + output_move_ids = + EXCLUDED.output_move_ids, + output_done_quantity = + EXCLUDED.output_done_quantity, + state_hash = EXCLUDED.state_hash, + payload_json = EXCLUDED.payload_json, + odoo_write_date = + EXCLUDED.odoo_write_date, + observed_at = EXCLUDED.observed_at, + updated_at = now() + """, + { + **final_record, + "production_started_at": parse_odoo_datetime( + final_record["evidence_json"]["production"].get( + "date_start" + ) + ), + "production_ended_at": parse_odoo_datetime( + final_record["evidence_json"]["production"].get( + "date_finished" + ) + ), + "payload_json": Json( + final_record["evidence_json"] + ), + }, + ) + + cur.execute( + """ + INSERT INTO + mv_loss_intelligence.odoo_shopfloor_sessions ( + tenant, + site, + session_key, + production_chain_key, + machine_id, + workcenter_id, + workcenter_name, + odoo_production_id, + production_order, + odoo_workorder_id, + workorder_name, + workorder_state, + production_state, + origin, + product_id, + product_name, + product_uom_id, + product_uom_name, + started_at, + ended_at, + session_status, + operator_employee_id, + operator_name, + operator_candidate_employee_id, + operator_candidate_name, + operator_resolution, + operator_confidence, + operator_confirmation_status, + planned_quantity, + declared_quantity, + declared_quantity_candidate, + posted_quantity, + remaining_planned_quantity, + quantity_status, + quantity_confirmation_status, + confidence, + official_eligible, + evidence_json, + reviewed_by, + reviewed_at, + review_notes, + source_observed_at, + source_odoo_write_date, + updated_at + ) + VALUES ( + %(tenant)s, + %(site)s, + %(session_key)s, + %(production_chain_key)s, + %(machine_id)s, + %(workcenter_id)s, + %(workcenter_name)s, + %(odoo_production_id)s, + %(production_order)s, + %(odoo_workorder_id)s, + %(workorder_name)s, + %(workorder_state)s, + %(production_state)s, + %(origin)s, + %(product_id)s, + %(product_name)s, + %(product_uom_id)s, + %(product_uom_name)s, + %(started_at)s, + %(ended_at)s, + %(session_status)s, + %(operator_employee_id)s, + %(operator_name)s, + %(operator_candidate_employee_id)s, + %(operator_candidate_name)s, + %(operator_resolution)s, + %(operator_confidence)s, + %(operator_confirmation_status)s, + %(planned_quantity)s, + %(declared_quantity)s, + %(declared_quantity_candidate)s, + %(posted_quantity)s, + %(remaining_planned_quantity)s, + %(quantity_status)s, + %(quantity_confirmation_status)s, + %(confidence)s, + false, + %(evidence_json)s, + %(reviewed_by)s, + %(reviewed_at)s, + %(review_notes)s, + %(source_observed_at)s, + %(source_odoo_write_date)s, + now() + ) + ON CONFLICT ( + tenant, + site, + odoo_workorder_id + ) + DO UPDATE SET + session_key = EXCLUDED.session_key, + production_chain_key = + EXCLUDED.production_chain_key, + machine_id = EXCLUDED.machine_id, + workcenter_id = EXCLUDED.workcenter_id, + workcenter_name = EXCLUDED.workcenter_name, + odoo_production_id = + EXCLUDED.odoo_production_id, + production_order = + EXCLUDED.production_order, + workorder_name = EXCLUDED.workorder_name, + workorder_state = + EXCLUDED.workorder_state, + production_state = + EXCLUDED.production_state, + origin = EXCLUDED.origin, + product_id = EXCLUDED.product_id, + product_name = EXCLUDED.product_name, + product_uom_id = EXCLUDED.product_uom_id, + product_uom_name = + EXCLUDED.product_uom_name, + started_at = EXCLUDED.started_at, + ended_at = EXCLUDED.ended_at, + session_status = EXCLUDED.session_status, + operator_employee_id = + EXCLUDED.operator_employee_id, + operator_name = EXCLUDED.operator_name, + operator_candidate_employee_id = + EXCLUDED.operator_candidate_employee_id, + operator_candidate_name = + EXCLUDED.operator_candidate_name, + operator_resolution = + EXCLUDED.operator_resolution, + operator_confidence = + EXCLUDED.operator_confidence, + operator_confirmation_status = + EXCLUDED.operator_confirmation_status, + planned_quantity = + EXCLUDED.planned_quantity, + declared_quantity = + EXCLUDED.declared_quantity, + declared_quantity_candidate = + EXCLUDED.declared_quantity_candidate, + posted_quantity = + EXCLUDED.posted_quantity, + remaining_planned_quantity = + EXCLUDED.remaining_planned_quantity, + quantity_status = + EXCLUDED.quantity_status, + quantity_confirmation_status = + EXCLUDED.quantity_confirmation_status, + confidence = EXCLUDED.confidence, + official_eligible = false, + evidence_json = EXCLUDED.evidence_json, + reviewed_by = EXCLUDED.reviewed_by, + reviewed_at = EXCLUDED.reviewed_at, + review_notes = EXCLUDED.review_notes, + source_observed_at = + EXCLUDED.source_observed_at, + source_odoo_write_date = + EXCLUDED.source_odoo_write_date, + updated_at = now() + """, + { + **final_record, + "evidence_json": Json( + final_record["evidence_json"] + ), + "reviewed_by": final_record.get("reviewed_by"), + "reviewed_at": final_record.get("reviewed_at"), + "review_notes": final_record.get("review_notes"), + }, + ) + + return transition_inserted + + +def update_sync_state( + connection, + config: Config, + *, + count: int | None = None, + cursor_at: datetime | None = None, + error: str | None = None, +) -> None: + with connection.cursor() as cur: + cur.execute( + """ + INSERT INTO + mv_loss_intelligence.odoo_shopfloor_sync_state ( + tenant, + site, + sync_name, + last_success_at, + last_odoo_cursor_at, + last_processed_count, + last_error, + updated_at + ) + VALUES ( + %s, + %s, + %s, + CASE WHEN %s IS NULL THEN NULL ELSE now() END, + %s, + %s, + %s, + now() + ) + ON CONFLICT ( + tenant, + site, + sync_name + ) + DO UPDATE SET + last_success_at = CASE + WHEN EXCLUDED.last_error IS NULL + THEN now() + ELSE + mv_loss_intelligence + .odoo_shopfloor_sync_state + .last_success_at + END, + last_odoo_cursor_at = COALESCE( + EXCLUDED.last_odoo_cursor_at, + mv_loss_intelligence + .odoo_shopfloor_sync_state + .last_odoo_cursor_at + ), + last_processed_count = COALESCE( + EXCLUDED.last_processed_count, + mv_loss_intelligence + .odoo_shopfloor_sync_state + .last_processed_count + ), + last_error = EXCLUDED.last_error, + updated_at = now() + """, + ( + config.tenant, + config.site, + SYNC_NAME, + None if error else True, + cursor_at, + count, + error, + ), + ) + + +def load_cursor( + connection, + config: Config, + lookback_hours: float, +) -> datetime: + with connection.cursor() as cur: + cur.execute( + """ + SELECT last_odoo_cursor_at + FROM + mv_loss_intelligence.odoo_shopfloor_sync_state + WHERE tenant = %s + AND site = %s + AND sync_name = %s + """, + (config.tenant, config.site, SYNC_NAME), + ) + row = cur.fetchone() + if row and row[0]: + return row[0].astimezone(timezone.utc) - timedelta(minutes=5) + return datetime.now(timezone.utc) - timedelta( + hours=lookback_hours + ) + + +def run_once( + odoo: OdooClient, + connection, + config: Config, + args: argparse.Namespace, +) -> dict[str, Any]: + machine_map = load_machine_map( + connection, + config, + args.machine_map, + ) + technical_ids = load_technical_employees( + connection, + config, + ) + + since = load_cursor( + connection, + config, + args.lookback_hours, + ) + + source = fetch_source_data( + odoo, + machine_map, + args.workorder_id, + since, + ) + records = build_records( + source, + machine_map, + technical_ids, + ) + + summary = { + "version": VERSION, + "dry_run": args.dry_run, + "machine_map": machine_map, + "since": since.isoformat(), + "workorders_fetched": len(source["workorders"]), + "sessions_built": len(records), + "records": [ + { + key: json_safe(record.get(key)) + for key in ( + "odoo_workorder_id", + "odoo_production_id", + "production_order", + "production_chain_key", + "machine_id", + "started_at", + "ended_at", + "session_status", + "operator_employee_id", + "operator_name", + "operator_candidate_employee_id", + "operator_candidate_name", + "operator_resolution", + "operator_confidence", + "operator_confirmation_status", + "planned_quantity", + "declared_quantity", + "declared_quantity_candidate", + "posted_quantity", + "remaining_planned_quantity", + "quantity_status", + "quantity_confirmation_status", + "confidence", + "official_eligible", + ) + } + for record in records + ], + } + + if args.dry_run: + connection.rollback() + return summary + + transition_count = 0 + try: + for record in records: + if upsert_record(connection, config, record): + transition_count += 1 + if not args.workorder_id: + cursor_at = datetime.now(timezone.utc) + update_sync_state( + connection, + config, + count=len(records), + cursor_at=cursor_at, + error=None, + ) + connection.commit() + except Exception: + connection.rollback() + raise + + summary["transitions_inserted"] = transition_count + return summary + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + description=( + "Sincroniza sesiones Odoo Shop Floor a PostgreSQL " + "en modo SHADOW." + ) + ) + parser.add_argument( + "--once", + action="store_true", + help="Ejecuta una iteración y termina.", + ) + parser.add_argument( + "--watch", + action="store_true", + help="Ejecuta de forma continua.", + ) + parser.add_argument( + "--interval", + type=float, + default=15.0, + help="Segundos entre iteraciones en modo watch.", + ) + parser.add_argument( + "--lookback-hours", + type=float, + default=168.0, + help="Ventana inicial si todavía no existe cursor.", + ) + parser.add_argument( + "--workorder-id", + action="append", + type=int, + default=[], + help=( + "Workorder concreto para backfill. " + "Puede repetirse." + ), + ) + parser.add_argument( + "--machine-map", + action="append", + default=[], + help=( + "Mapa adicional WORKCENTER_ID=MACHINE_ID. " + "Ejemplo: 54=CORT-02." + ), + ) + parser.add_argument( + "--dry-run", + action="store_true", + help="Muestra la resolución sin escribir.", + ) + parser.add_argument( + "--output", + help="Guarda cada resumen como JSONL.", + ) + return parser + + +def emit_summary( + summary: dict[str, Any], + output: str | None, +) -> None: + text = json.dumps( + summary, + ensure_ascii=False, + indent=2, + default=str, + ) + print(text, flush=True) + + if output: + with open(output, "a", encoding="utf-8") as handle: + handle.write( + json.dumps( + summary, + ensure_ascii=False, + default=str, + ) + + "\n" + ) + + +def main() -> int: + parser = build_parser() + args = parser.parse_args() + + if not args.once and not args.watch: + args.once = True + + config = load_config() + odoo = OdooClient(config) + connection = connect_pg(config) + + try: + if args.once: + summary = run_once( + odoo, + connection, + config, + args, + ) + emit_summary(summary, args.output) + return 0 + + while RUNNING: + started = time.monotonic() + try: + summary = run_once( + odoo, + connection, + config, + args, + ) + emit_summary(summary, args.output) + except Exception as exc: + connection.rollback() + error = ( + f"{type(exc).__name__}: {exc}" + ) + try: + update_sync_state( + connection, + config, + error=error, + ) + connection.commit() + except Exception: + connection.rollback() + print( + f"ERROR {error}", + file=sys.stderr, + flush=True, + ) + + elapsed = time.monotonic() - started + remaining = max(args.interval - elapsed, 0) + deadline = time.monotonic() + remaining + while RUNNING and time.monotonic() < deadline: + time.sleep(min(0.5, deadline - time.monotonic())) + + return 0 + finally: + connection.close() + + +if __name__ == "__main__": + raise SystemExit(main())