182 lines
4.1 KiB
Python
Executable File

#!/usr/bin/env python3
import os
import sys
import xmlrpc.client
from pathlib import Path
BASE_DIR = Path(__file__).resolve().parents[2]
ENV_FILE = BASE_DIR / ".env.odoo-sync"
def load_env(path):
if not path.exists():
return
for line in path.read_text().splitlines():
line = line.strip()
if not line or line.startswith("#") or "=" not in line:
continue
key, value = line.split("=", 1)
os.environ.setdefault(key.strip(), value.strip().strip('"').strip("'"))
def env(name, default=None, required=False):
value = os.getenv(name, default)
if required and not value:
print(f"[ERROR] Falta variable: {name}", file=sys.stderr)
sys.exit(2)
return value
def existing_fields(models, db, uid, password, model, wanted):
fields = models.execute_kw(
db,
uid,
password,
model,
"fields_get",
[],
{"attributes": ["string", "type"]},
)
return [f for f in wanted if f in fields]
def search_read(models, db, uid, password, model, domain, wanted_fields, limit=20):
fields = existing_fields(models, db, uid, password, model, wanted_fields)
return models.execute_kw(
db,
uid,
password,
model,
"search_read",
[domain],
{
"fields": fields,
"limit": limit,
"order": "id desc",
},
)
def main():
load_env(ENV_FILE)
url = env("ODOO_URL", required=True).rstrip("/")
db = env("ODOO_DB", required=True)
user = env("ODOO_USER", required=True)
password = env("ODOO_PASSWORD", required=True)
workcenter_code = env("ODOO_WORKCENTER_CODE", "CORT-01")
common = xmlrpc.client.ServerProxy(f"{url}/xmlrpc/2/common")
models = xmlrpc.client.ServerProxy(f"{url}/xmlrpc/2/object")
uid = common.authenticate(db, user, password, {})
if not uid:
print("[ERROR] Autenticación fallida")
return 1
print(f"[OK] Autenticado en Odoo uid={uid}")
print()
print("=== Centros de trabajo ===")
workcenters = search_read(
models,
db,
uid,
password,
"mrp.workcenter",
[],
["id", "name", "code", "tag_ids"],
limit=50,
)
for wc in workcenters:
print(wc)
print()
print(f"=== Centro de trabajo con código/nombre {workcenter_code} ===")
# Algunos Odoo tienen campo code, otros no. Probamos primero code si existe.
wc_fields = existing_fields(models, db, uid, password, "mrp.workcenter", ["id", "name", "code"])
if "code" in wc_fields:
domain = ["|", ("code", "=", workcenter_code), ("name", "ilike", workcenter_code)]
else:
domain = [("name", "ilike", workcenter_code)]
target_wcs = search_read(
models,
db,
uid,
password,
"mrp.workcenter",
domain,
["id", "name", "code"],
limit=10,
)
for wc in target_wcs:
print(wc)
print()
print("=== Órdenes de trabajo recientes ===")
workorders = search_read(
models,
db,
uid,
password,
"mrp.workorder",
[],
[
"id",
"name",
"state",
"workcenter_id",
"production_id",
"product_id",
"date_start",
"date_finished",
"duration_expected",
"duration",
"qty_production",
],
limit=20,
)
for wo in workorders:
print(wo)
print()
print("=== Órdenes de fabricación recientes ===")
productions = search_read(
models,
db,
uid,
password,
"mrp.production",
[],
[
"id",
"name",
"state",
"product_id",
"product_qty",
"qty_producing",
"date_start",
"date_finished",
"workorder_ids",
],
limit=20,
)
for mo in productions:
print(mo)
return 0
if __name__ == "__main__":
raise SystemExit(main())