320655b0a0
- Remove apps/dispatch/ (100% replaced by ops dispatch module, unmaintained) - Commit services/targo-hub/lib/ (24 modules, 6290 lines — was never tracked) - Commit services/docuseal + services/legacy-db docker-compose configs - Extract client app composables: useOTP, useAddressSearch, catalog data, format utils - Refactor CartPage.vue 630→175 lines, CatalogPage.vue 375→95 lines - Clean hardcoded credentials from config.js fallback values - Add client portal: catalog, cart, checkout, OTP verification, address search - Add ops: NetworkPage, AgentFlowsPage, ConversationPanel, UnifiedCreateModal - Add ops composables: useBestTech, useConversations, usePermissions, useScanner - Add field app: scanner composable, docker/nginx configs Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
202 lines
8.3 KiB
Python
202 lines
8.3 KiB
Python
"""
|
|
Fix OLT port data in Service Locations.
|
|
Replaces 'None/x/y ONT:z' with the actual OLT IP from legacy DB.
|
|
Also adds custom fields for OLT data if not present.
|
|
|
|
Run inside erpnext-backend container:
|
|
bench --site erp.gigafibre.ca execute fix_olt_port_ip.run
|
|
"""
|
|
import frappe
|
|
import pymysql
|
|
|
|
import os
|
|
|
|
# Use local legacy-db container if available, else remote
|
|
LEGACY_DB = {
|
|
"host": os.environ.get("LEGACY_DB_HOST", "legacy-db"),
|
|
"port": int(os.environ.get("LEGACY_DB_PORT", "3306")),
|
|
"user": "facturation",
|
|
"password": "VD67owoj",
|
|
"database": "gestionclient",
|
|
"charset": "utf8mb4",
|
|
"cursorclass": pymysql.cursors.DictCursor,
|
|
}
|
|
|
|
|
|
def ensure_custom_fields():
|
|
"""Create missing custom fields on Service Location for OLT data."""
|
|
fields = [
|
|
{"fieldname": "olt_ip", "label": "OLT IP", "fieldtype": "Data", "insert_after": "olt_port"},
|
|
{"fieldname": "olt_name", "label": "Nom OLT", "fieldtype": "Data", "insert_after": "olt_ip"},
|
|
{"fieldname": "ont_id", "label": "ONT ID", "fieldtype": "Int", "insert_after": "olt_name"},
|
|
{"fieldname": "ont_serial", "label": "ONT Serial", "fieldtype": "Data", "insert_after": "ont_id"},
|
|
{"fieldname": "terrain_type", "label": "Type terrain", "fieldtype": "Data", "insert_after": "ont_serial"},
|
|
{"fieldname": "fibre_distance_m","label": "Distance fibre (m)", "fieldtype": "Float", "insert_after": "terrain_type"},
|
|
{"fieldname": "fibre_spans", "label": "Nombre de portées", "fieldtype": "Int", "insert_after": "fibre_distance_m"},
|
|
{"fieldname": "vlan_manage", "label": "VLAN Manage", "fieldtype": "Int", "insert_after": "network_id"},
|
|
{"fieldname": "vlan_internet", "label": "VLAN Internet", "fieldtype": "Int", "insert_after": "vlan_manage"},
|
|
{"fieldname": "vlan_telephone", "label": "VLAN Téléphone", "fieldtype": "Int", "insert_after": "vlan_internet"},
|
|
{"fieldname": "vlan_tv", "label": "VLAN TV", "fieldtype": "Int", "insert_after": "vlan_telephone"},
|
|
{"fieldname": "legacy_fibre_id", "label": "Legacy Fibre ID", "fieldtype": "Int", "insert_after": "legacy_delivery_id", "hidden": 1},
|
|
]
|
|
|
|
created = 0
|
|
for f in fields:
|
|
existing = frappe.db.exists("Custom Field", {"dt": "Service Location", "fieldname": f["fieldname"]})
|
|
if existing:
|
|
continue
|
|
doc = frappe.new_doc("Custom Field")
|
|
doc.dt = "Service Location"
|
|
doc.fieldname = f["fieldname"]
|
|
doc.label = f["label"]
|
|
doc.fieldtype = f["fieldtype"]
|
|
doc.insert_after = f["insert_after"]
|
|
if f.get("hidden"):
|
|
doc.hidden = 1
|
|
doc.insert(ignore_permissions=True)
|
|
created += 1
|
|
print(" Created custom field: {}".format(f["fieldname"]))
|
|
|
|
if created:
|
|
frappe.db.commit()
|
|
frappe.clear_cache(doctype="Service Location")
|
|
print("Created {} custom fields".format(created))
|
|
else:
|
|
print("All custom fields already exist")
|
|
|
|
|
|
def run():
|
|
print("=" * 60)
|
|
print("1. Ensuring custom fields exist...")
|
|
print("=" * 60)
|
|
ensure_custom_fields()
|
|
|
|
print("\n" + "=" * 60)
|
|
print("2. Fetching fibre data from legacy DB...")
|
|
print("=" * 60)
|
|
|
|
conn = pymysql.connect(**LEGACY_DB)
|
|
|
|
with conn.cursor() as cur:
|
|
cur.execute("""
|
|
SELECT f.id as fibre_id, f.service_id, f.sn as ont_sn,
|
|
f.info_connect as olt_ip, f.ontid, f.frame, f.slot, f.port,
|
|
f.tech as terrain, f.distance, f.nb_portees,
|
|
f.vlan_manage, f.vlan_internet, f.vlan_telephone, f.vlan_tele,
|
|
fo.description as olt_name,
|
|
s.delivery_id
|
|
FROM fibre f
|
|
LEFT JOIN fibre_olt fo ON f.info_connect = fo.ip
|
|
LEFT JOIN service s ON f.service_id = s.id
|
|
WHERE s.delivery_id IS NOT NULL
|
|
""")
|
|
fibres = cur.fetchall()
|
|
|
|
conn.close()
|
|
print("Found {} fibre records".format(len(fibres)))
|
|
|
|
# Build delivery_id → location name map
|
|
locs = frappe.db.sql(
|
|
'SELECT name, legacy_delivery_id FROM "tabService Location" WHERE legacy_delivery_id IS NOT NULL',
|
|
as_dict=True
|
|
)
|
|
loc_map = {int(l["legacy_delivery_id"]): l["name"] for l in locs if l["legacy_delivery_id"]}
|
|
print("Location map: {} entries".format(len(loc_map)))
|
|
|
|
print("\n" + "=" * 60)
|
|
print("3. Updating Service Locations...")
|
|
print("=" * 60)
|
|
|
|
updated = 0
|
|
for fb in fibres:
|
|
loc_name = loc_map.get(fb["delivery_id"])
|
|
if not loc_name:
|
|
continue
|
|
|
|
# Build proper olt_port with IP
|
|
olt_ip = str(fb["olt_ip"]) if fb["olt_ip"] else ""
|
|
olt_name = str(fb["olt_name"]) if fb["olt_name"] else ""
|
|
|
|
frame = fb["frame"] if fb["frame"] is not None else 0
|
|
slot = fb["slot"] if fb["slot"] is not None else 0
|
|
port = fb["port"] if fb["port"] is not None else 0
|
|
|
|
port_parts = []
|
|
if olt_ip:
|
|
port_parts.append(olt_ip)
|
|
port_parts.append("{}/{}/{}".format(frame, slot, port))
|
|
if fb["ontid"]:
|
|
port_parts[-1] += " ONT:{}".format(fb["ontid"])
|
|
olt_port = " ".join(port_parts)
|
|
|
|
vlans = []
|
|
if fb["vlan_internet"]:
|
|
vlans.append("inet:{}".format(fb["vlan_internet"]))
|
|
if fb["vlan_manage"]:
|
|
vlans.append("mgmt:{}".format(fb["vlan_manage"]))
|
|
if fb["vlan_telephone"]:
|
|
vlans.append("tel:{}".format(fb["vlan_telephone"]))
|
|
if fb["vlan_tele"]:
|
|
vlans.append("tv:{}".format(fb["vlan_tele"]))
|
|
network_id = " ".join(vlans) if vlans else None
|
|
|
|
sets = ['olt_port = %(olt_port)s', 'connection_type = %(conn_type)s']
|
|
params = {"name": loc_name, "olt_port": olt_port, "conn_type": "Fibre FTTH"}
|
|
|
|
if network_id:
|
|
sets.append("network_id = %(network_id)s")
|
|
params["network_id"] = network_id
|
|
if olt_ip:
|
|
sets.append("olt_ip = %(olt_ip)s")
|
|
params["olt_ip"] = olt_ip
|
|
if olt_name:
|
|
sets.append("olt_name = %(olt_name)s")
|
|
params["olt_name"] = olt_name
|
|
sets.append("network_notes = %(network_notes)s")
|
|
params["network_notes"] = "OLT: {}".format(olt_name)
|
|
if fb["ontid"]:
|
|
sets.append("ont_id = %(ont_id)s")
|
|
params["ont_id"] = int(fb["ontid"])
|
|
if fb["ont_sn"]:
|
|
sets.append("ont_serial = %(ont_sn)s")
|
|
params["ont_sn"] = str(fb["ont_sn"])
|
|
if fb["terrain"]:
|
|
sets.append("terrain_type = %(terrain)s")
|
|
params["terrain"] = str(fb["terrain"])
|
|
if fb["distance"]:
|
|
sets.append("fibre_distance_m = %(distance)s")
|
|
params["distance"] = float(fb["distance"])
|
|
if fb["nb_portees"]:
|
|
sets.append("fibre_spans = %(spans)s")
|
|
params["spans"] = int(fb["nb_portees"])
|
|
if fb["fibre_id"]:
|
|
sets.append("legacy_fibre_id = %(fibre_id)s")
|
|
params["fibre_id"] = int(fb["fibre_id"])
|
|
if fb["vlan_manage"]:
|
|
sets.append("vlan_manage = %(vlan_m)s")
|
|
params["vlan_m"] = int(fb["vlan_manage"])
|
|
if fb["vlan_internet"]:
|
|
sets.append("vlan_internet = %(vlan_i)s")
|
|
params["vlan_i"] = int(fb["vlan_internet"])
|
|
if fb["vlan_telephone"]:
|
|
sets.append("vlan_telephone = %(vlan_t)s")
|
|
params["vlan_t"] = int(fb["vlan_telephone"])
|
|
if fb["vlan_tele"]:
|
|
sets.append("vlan_tv = %(vlan_tv)s")
|
|
params["vlan_tv"] = int(fb["vlan_tele"])
|
|
|
|
sql = 'UPDATE "tabService Location" SET {} WHERE name = %(name)s'.format(", ".join(sets))
|
|
frappe.db.sql(sql, params)
|
|
updated += 1
|
|
|
|
frappe.db.commit()
|
|
print("Updated {} Service Locations with fibre data".format(updated))
|
|
|
|
# Show sample of C-MARIL fix
|
|
result = frappe.db.sql(
|
|
'SELECT name, customer, olt_port, olt_ip, olt_name FROM "tabService Location" WHERE customer LIKE %s AND connection_type = %s',
|
|
('%MARIL5145574140%', 'Fibre FTTH'), as_dict=True
|
|
)
|
|
for r in result:
|
|
print(" VERIFY: {} → olt_port={}, olt_ip={}, olt_name={}".format(r["name"], r["olt_port"], r["olt_ip"], r["olt_name"]))
|