"""Populate OpenMetadata Data Insights: assign owners (all tables),
descriptions + tier tags (business tables), then the DataInsights app can
produce a meaningful Data Assets dashboard (coverage > 0)."""
import json, os, time, urllib.request, urllib.parse, urllib.error

OM = "http://openmetadata-server:8585/api"
H = {"Authorization": "Bearer " + os.environ["OM_TOKEN"]}
MO = "f0fc4958-5a46-4088-b5d0-9be137dae95d"
BART = "e2849c82-4d42-4e85-b8c7-802181e688be"


def req(method, path, body=None, ct="application/json"):
    data = json.dumps(body).encode() if body is not None else None
    h = dict(H)
    if data is not None:
        h["Content-Type"] = ct
    r = urllib.request.Request(OM + path, data=data, headers=h, method=method)
    try:
        with urllib.request.urlopen(r, timeout=60) as resp:
            return resp.status, json.loads(resp.read().decode() or "{}")
    except urllib.error.HTTPError as e:
        return e.code, e.read().decode()[:200]


def patch(eid, ops):
    return req("PATCH", "/v1/tables/" + eid, ops, "application/json-patch+json")


# ---- map all tables: fqn -> id --------------------------------------------
st, d = req("GET", "/v1/search/query?q=*&index=table_search_index&size=400")
fqn2id = {}
for h in d["hits"]["hits"]:
    fqn2id[h["_source"]["fullyQualifiedName"]] = h["_id"]
print("tables found:", len(fqn2id))

# ---- 1) owner on ALL tables (ownership coverage) --------------------------
owned = 0
for fqn, eid in fqn2id.items():
    st, r = patch(eid, [{"op": "add", "path": "/owners",
                         "value": [{"id": MO, "type": "user"}]}])
    if st == 200:
        owned += 1
    elif st == 400:  # owners already present -> replace
        st2, _ = patch(eid, [{"op": "replace", "path": "/owners",
                              "value": [{"id": MO, "type": "user"}]}])
        owned += 1 if st2 == 200 else 0
print("owners set on:", owned)

# ---- 2) descriptions + tier on business tables ----------------------------
BIZ = {
    "atc_postgres.postgres.public.sales_orders": (
        "Operational sales orders - source of truth in PostgreSQL. Continuously "
        "change-captured by Debezium into Kafka and curated into the lakehouse.", "Tier.Tier1"),
    "atc_mysql.default.hr.employee_events": (
        "HR employee lifecycle events (hire/leave/transfer) in MySQL. CDC source, "
        "masked for PII in the curated layer.", "Tier.Tier1"),
    "atc_mongodb.supplychain.supplychain.events": (
        "Supply-chain events stream in MongoDB. CDC source feeding Kafka and S3.", "Tier.Tier1"),
    "atc_trino.iceberg.curated_masked.sales_orders_masked": (
        "Curated, PII-masked sales orders (Iceberg on S3). Governed analytics layer.", "Tier.Tier1"),
    "atc_trino.iceberg.curated_masked.employee_events_masked": (
        "Curated, PII-masked HR employee events (Iceberg on S3).", "Tier.Tier1"),
    "atc_trino.iceberg.hadoop.historical_sales": (
        "Historical sales offloaded from Hadoop/HDFS into Iceberg for long-term analytics.", "Tier.Tier2"),
    "atc_trino.iceberg.hadoop.historical_sales_hdfs": (
        "Historical sales materialized from HDFS via the hadoop_to_trino pipeline.", "Tier.Tier2"),
    "atc_trino.cassandra_telemetry.telemetry.device_metrics": (
        "Device telemetry time-series in Cassandra, queryable via Trino.", "Tier.Tier2"),
    'atc_trino.kafka.default."postgres-sales.public.sales_orders"': (
        "Debezium CDC stream of sales_orders (Kafka), queryable via Trino.", "Tier.Tier2"),
    'atc_trino.kafka.default."mongodb-supplychain.supplychain.events"': (
        "Debezium CDC stream of supply-chain events (Kafka), queryable via Trino.", "Tier.Tier2"),
    "atc_trino.postgres_sales.public.sales_orders": (
        "Trino read-through view of the PostgreSQL sales_orders source.", "Tier.Tier2"),
    "atc_trino.mysql_hr.hr.employee_events": (
        "Trino read-through view of the MySQL employee_events source.", "Tier.Tier2"),
    "atc_trino.mongodb_supplychain.supplychain.events": (
        "Trino read-through view of the MongoDB supply-chain events source.", "Tier.Tier2"),
    "atc_trino.postgres_sales.monitor.cdc_operations": (
        "Operational view: CDC operation counts per table.", "Tier.Tier3"),
    "atc_trino.postgres_sales.monitor.cdc_recent_events": (
        "Operational view: most recent CDC events.", "Tier.Tier3"),
    "atc_trino.postgres_sales.monitor.debezium_connectors": (
        "Operational view: Debezium connector status.", "Tier.Tier3"),
    "atc_trino.postgres_sales.monitor.kafka_topics": (
        "Operational view: Kafka topic inventory and lag.", "Tier.Tier3"),
    "atc_trino.postgres_sales.monitor.spark_applications": (
        "Operational view: Spark application run history.", "Tier.Tier3"),
}
done = 0
for fqn, (desc, tier) in BIZ.items():
    eid = fqn2id.get(fqn)
    if not eid:
        print("  ! not found:", fqn); continue
    st, t = req("GET", "/v1/tables/" + eid + "?fields=tags,owners")
    ops = [{"op": "add", "path": "/description", "value": desc}]
    tags = [tg for tg in (t.get("tags") or []) if not tg.get("tagFQN", "").startswith("Tier.")]
    tags.append({"tagFQN": tier, "source": "Classification",
                 "labelType": "Manual", "state": "Confirmed"})
    ops.append({"op": "add" if "tags" not in t else "replace", "path": "/tags", "value": tags})
    st2, r2 = patch(eid, ops)
    print("  biz %-55s desc+%s -> %s" % (fqn.split(".")[-1][:50], tier, st2))
    done += 1 if st2 == 200 else 0
print("business enriched:", done)