#!/usr/bin/env python3
import hashlib
import json
import os
import ssl
import time
import urllib.error
import urllib.parse
import urllib.request
from datetime import date, datetime, timezone
from decimal import Decimal

import pymysql


HUBSPOT_COMPANY_PROPERTIES = [
    "name",
    "domain",
    "website",
    "hs_additional_domains",
    "account_id__c",
    "portal_id__c",
    "account_allias__c",
    "hs_lastmodifieddate",
]

MULTI_LABEL_SUFFIXES = {
    "ac.in", "ac.jp", "ac.kr", "ac.nz", "ac.th", "ac.uk",
    "co.id", "co.in", "co.jp", "co.kr", "co.nz", "co.th", "co.uk",
    "com.au", "com.cn", "com.hk", "com.my", "com.ph", "com.sg", "com.tw",
    "edu.au", "edu.cn", "edu.hk", "edu.sg",
    "gov.au", "gov.cn", "gov.hk", "gov.sg",
    "net.au", "net.cn", "org.au", "org.cn", "org.uk",
}


def env(name, default=None, required=False):
    value = os.environ.get(name, default)
    if required and (value is None or str(value).strip() == ""):
        raise RuntimeError(f"{name} is required")
    return value


def json_default(value):
    if isinstance(value, datetime):
        if value.tzinfo is not None:
            value = value.astimezone(timezone.utc).replace(tzinfo=None)
        return value.strftime("%Y-%m-%d %H:%M:%S")
    if isinstance(value, date):
        return value.isoformat()
    if isinstance(value, Decimal):
        return str(value)
    raise TypeError(f"{type(value).__name__} is not JSON serializable")


def log_event(event_name, **fields):
    print(
        json.dumps(
            {"event": event_name, **fields},
            default=json_default,
            separators=(",", ":"),
            sort_keys=True,
        )
    )


def target_db_connect():
    database_url = env("TARGET_DATABASE_URL", required=True)
    parsed = urllib.parse.urlparse(database_url)
    database = parsed.path.lstrip("/")
    if not parsed.hostname or not database:
        raise RuntimeError("TARGET_DATABASE_URL must include host and database")

    query = urllib.parse.parse_qs(parsed.query)
    ssl_config = None
    if str((query.get("ssl_verify_cert") or [""])[0]).lower() not in {"0", "false"}:
        ssl_config = ssl.create_default_context()
        ca_path = env("TARGET_TIDB_SSL_CA_PATH") or env("TIDB_SSL_CA_PATH")
        if ca_path:
            ssl_config.load_verify_locations(cafile=ca_path)

    return pymysql.connect(
        host=parsed.hostname,
        port=parsed.port or 4000,
        user=urllib.parse.unquote(parsed.username or ""),
        password=urllib.parse.unquote(parsed.password or ""),
        database=urllib.parse.unquote(database),
        charset="utf8mb4",
        cursorclass=pymysql.cursors.DictCursor,
        autocommit=False,
        connect_timeout=10,
        read_timeout=60,
        write_timeout=60,
        ssl=ssl_config,
    )


def parse_hubspot_datetime(value):
    value = str(value or "").strip()
    if not value:
        return None
    if value.endswith("Z"):
        value = value[:-1] + "+00:00"
    try:
        parsed = datetime.fromisoformat(value)
    except ValueError:
        return None
    if parsed.tzinfo is not None:
        parsed = parsed.astimezone(timezone.utc).replace(tzinfo=None)
    return parsed


def normalized_host(value):
    value = str(value or "").strip().lower()
    if not value or value == "null":
        return ""
    if "://" not in value:
        value = "//" + value
    host = (urllib.parse.urlparse(value).hostname or "").strip(". ").lower()
    return host[4:] if host.startswith("www.") else host


def registrable_domain(value):
    host = normalized_host(value)
    parts = [part for part in host.split(".") if part]
    if len(parts) < 2:
        return host
    if ".".join(parts[-2:]) in MULTI_LABEL_SUFFIXES and len(parts) >= 3:
        return ".".join(parts[-3:])
    return ".".join(parts[-2:])


def additional_domain_values(value):
    normalized = str(value or "").strip().replace("\n", ";").replace(",", ";")
    return [item.strip() for item in normalized.split(";") if item.strip()]


def hubspot_request(method, path, body=None):
    token = env("HUBSPOT_PRIVATE_APP_TOKEN", required=True)
    timeout = int(env("HUBSPOT_REQUEST_TIMEOUT_SECONDS", "30"))
    retries = int(env("HUBSPOT_RETRY_COUNT", "3"))
    data = json.dumps(body).encode("utf-8") if body is not None else None
    headers = {"Authorization": f"Bearer {token}", "Accept": "application/json"}
    if data is not None:
        headers["Content-Type"] = "application/json"

    for attempt in range(retries + 1):
        request = urllib.request.Request(
            f"https://api.hubapi.com{path}",
            data=data,
            headers=headers,
            method=method,
        )
        try:
            with urllib.request.urlopen(request, timeout=timeout) as response:
                return json.loads(response.read().decode("utf-8"))
        except urllib.error.HTTPError as exc:
            retryable = exc.code == 429 or 500 <= exc.code < 600
            if not retryable or attempt >= retries:
                detail = exc.read().decode("utf-8", errors="replace")[:500]
                raise RuntimeError(f"HubSpot HTTP {exc.code}: {detail}") from exc
            retry_after = exc.headers.get("Retry-After")
            delay = float(retry_after) if retry_after else min(2**attempt, 8)
        except (urllib.error.URLError, TimeoutError) as exc:
            if attempt >= retries:
                raise RuntimeError(f"HubSpot request failed: {exc}") from exc
            delay = min(2**attempt, 8)
        time.sleep(delay)
    raise RuntimeError("HubSpot request retry loop exhausted")


def fetch_hubspot_list(list_id):
    response = hubspot_request("GET", f"/crm/v3/lists/{urllib.parse.quote(list_id)}")
    return response.get("list") or response


def fetch_hubspot_memberships(list_id, run_id):
    results = []
    after = None
    page = 0
    reported_total = None
    while True:
        query = {"limit": 250}
        if after:
            query["after"] = after
        response = hubspot_request(
            "GET",
            f"/crm/v3/lists/{urllib.parse.quote(list_id)}/memberships?"
            + urllib.parse.urlencode(query),
        )
        page += 1
        page_results = response.get("results") or []
        results.extend(page_results)
        reported_total = response.get("total", reported_total)
        log_event(
            "hubspot_membership_page",
            run_id=run_id,
            page=page,
            page_count=len(page_results),
            accumulated_count=len(results),
            reported_total=reported_total,
        )
        after = ((response.get("paging") or {}).get("next") or {}).get("after")
        if not after:
            return results, reported_total


def fetch_hubspot_companies(company_ids, run_id):
    companies = []
    for offset in range(0, len(company_ids), 100):
        batch_ids = company_ids[offset : offset + 100]
        response = hubspot_request(
            "POST",
            "/crm/v3/objects/companies/batch/read",
            {
                "properties": HUBSPOT_COMPANY_PROPERTIES,
                "inputs": [{"id": company_id} for company_id in batch_ids],
            },
        )
        batch_results = response.get("results") or []
        companies.extend(batch_results)
        log_event(
            "hubspot_company_batch",
            run_id=run_id,
            batch_number=offset // 100 + 1,
            requested_count=len(batch_ids),
            returned_count=len(batch_results),
            accumulated_count=len(companies),
        )
    return companies


def normalized_hubspot_company(company, membership_timestamp, sync_id, synced_at):
    properties = company.get("properties") or {}
    additional_domains = additional_domain_values(properties.get("hs_additional_domains"))
    additional_roots = sorted(
        {registrable_domain(value) for value in additional_domains if registrable_domain(value)}
    )
    return {
        "hubspot_company_id": str(company.get("id") or "").strip(),
        "membership_timestamp": parse_hubspot_datetime(membership_timestamp),
        "portal_id": str(properties.get("portal_id__c") or "").strip() or None,
        "account_id": str(properties.get("account_id__c") or "").strip() or None,
        "name": properties.get("name"),
        "account_alias": properties.get("account_allias__c"),
        "domain": normalized_host(properties.get("domain")) or None,
        "website": properties.get("website"),
        "additional_domains": json.dumps(additional_domains, separators=(",", ":")),
        "domain_root": registrable_domain(properties.get("domain")) or None,
        "website_domain_root": registrable_domain(properties.get("website")) or None,
        "additional_domain_roots": json.dumps(additional_roots, separators=(",", ":")),
        "hubspot_modified_at": parse_hubspot_datetime(properties.get("hs_lastmodifieddate")),
        "sync_id": sync_id,
        "synced_at": synced_at,
    }


def build_hubspot_snapshot(list_id, run_id, now):
    list_info = fetch_hubspot_list(list_id)
    if list_info.get("processingStatus") != "COMPLETE":
        raise RuntimeError(
            f"HubSpot list processing status is {list_info.get('processingStatus')!r}"
        )

    memberships, reported_total = fetch_hubspot_memberships(list_id, run_id)
    membership_map = {
        str(item.get("recordId") or "").strip(): item.get("membershipTimestamp")
        for item in memberships
        if str(item.get("recordId") or "").strip()
    }
    company_ids = sorted(membership_map)
    expected_total = int(
        reported_total if reported_total is not None else list_info.get("size") or 0
    )
    if len(company_ids) != expected_total:
        raise RuntimeError(
            f"HubSpot membership count mismatch unique={len(company_ids)} expected={expected_total}"
        )

    companies = fetch_hubspot_companies(company_ids, run_id)
    company_map = {str(item.get("id") or "").strip(): item for item in companies}
    missing_ids = sorted(set(company_ids) - set(company_map))
    if missing_ids:
        raise RuntimeError(f"HubSpot Company batch response missed {len(missing_ids)} records")

    fingerprint = [
        {"id": company_id, "properties": company_map[company_id].get("properties") or {}}
        for company_id in company_ids
    ]
    snapshot_hash = hashlib.sha256(
        json.dumps(fingerprint, sort_keys=True, separators=(",", ":")).encode("utf-8")
    ).hexdigest()
    sync_id = f"{run_id}-{snapshot_hash[:16]}"
    synced_at = now.astimezone(timezone.utc).replace(tzinfo=None)
    rows = [
        normalized_hubspot_company(
            company_map[company_id], membership_map[company_id], sync_id, synced_at
        )
        for company_id in company_ids
    ]
    return rows, {
        "sync_id": sync_id,
        "synced_at": synced_at,
        "list_total": len(rows),
        "snapshot_hash": snapshot_hash,
    }


def persist_hubspot_snapshot(conn, list_id, rows, metadata):
    with conn.cursor() as cursor:
        cursor.executemany(
            """
            INSERT INTO apac_hubspot_target_accounts (
              hubspot_company_id, membership_timestamp, portal_id, account_id,
              name, account_alias, domain, website, additional_domains,
              domain_root, website_domain_root, additional_domain_roots,
              hubspot_modified_at, sync_id, synced_at
            ) VALUES (
              %(hubspot_company_id)s, %(membership_timestamp)s, %(portal_id)s, %(account_id)s,
              %(name)s, %(account_alias)s, %(domain)s, %(website)s, %(additional_domains)s,
              %(domain_root)s, %(website_domain_root)s, %(additional_domain_roots)s,
              %(hubspot_modified_at)s, %(sync_id)s, %(synced_at)s
            )
            ON DUPLICATE KEY UPDATE
              membership_timestamp = VALUES(membership_timestamp),
              portal_id = VALUES(portal_id), account_id = VALUES(account_id),
              name = VALUES(name), account_alias = VALUES(account_alias),
              domain = VALUES(domain), website = VALUES(website),
              additional_domains = VALUES(additional_domains),
              domain_root = VALUES(domain_root),
              website_domain_root = VALUES(website_domain_root),
              additional_domain_roots = VALUES(additional_domain_roots),
              hubspot_modified_at = VALUES(hubspot_modified_at),
              sync_id = VALUES(sync_id), synced_at = VALUES(synced_at)
            """,
            rows,
        )
        cursor.execute(
            "DELETE FROM apac_hubspot_target_accounts WHERE sync_id <> %s",
            (metadata["sync_id"],),
        )
        cursor.execute(
            """
            INSERT INTO apac_hubspot_target_sync_state (
              list_id, last_successful_sync_id, last_successful_sync_at,
              last_successful_list_total, last_successful_snapshot_hash,
              last_attempt_at, last_attempt_status, last_error
            ) VALUES (%s, %s, %s, %s, %s, UTC_TIMESTAMP(), 'success', NULL)
            ON DUPLICATE KEY UPDATE
              last_successful_sync_id = VALUES(last_successful_sync_id),
              last_successful_sync_at = VALUES(last_successful_sync_at),
              last_successful_list_total = VALUES(last_successful_list_total),
              last_successful_snapshot_hash = VALUES(last_successful_snapshot_hash),
              last_attempt_at = UTC_TIMESTAMP(), last_attempt_status = 'success',
              last_error = NULL
            """,
            (
                list_id,
                metadata["sync_id"],
                metadata["synced_at"],
                metadata["list_total"],
                metadata["snapshot_hash"],
            ),
        )
    conn.commit()


def record_hubspot_sync_failure(conn, list_id, error_message):
    with conn.cursor() as cursor:
        cursor.execute(
            """
            INSERT INTO apac_hubspot_target_sync_state (
              list_id, last_attempt_at, last_attempt_status, last_error
            ) VALUES (%s, UTC_TIMESTAMP(), 'failed', %s)
            ON DUPLICATE KEY UPDATE
              last_attempt_at = UTC_TIMESTAMP(), last_attempt_status = 'failed',
              last_error = VALUES(last_error)
            """,
            (list_id, str(error_message)[:65535]),
        )
    conn.commit()


def read_hubspot_sync_state(conn, list_id):
    with conn.cursor() as cursor:
        cursor.execute(
            """
            SELECT last_successful_sync_id, last_successful_sync_at,
                   last_successful_list_total, last_attempt_status
            FROM apac_hubspot_target_sync_state
            WHERE list_id = %s
            """,
            (list_id,),
        )
        return cursor.fetchone() or {}


def snapshot_message_metadata(state, fresh, now):
    synced_at = state.get("last_successful_sync_at")
    if synced_at and synced_at.tzinfo is None:
        synced_at = synced_at.replace(tzinfo=timezone.utc)
    status = "fresh" if fresh else "stale" if state.get("last_successful_sync_id") else "unavailable"
    return {
        "hubspot_sync_id": state.get("last_successful_sync_id"),
        "hubspot_snapshot_synced_at": (
            synced_at.isoformat().replace("+00:00", "Z") if synced_at else None
        ),
        "hubspot_snapshot_status": status,
        "hubspot_snapshot_age_seconds": (
            max(0, int((now - synced_at).total_seconds())) if synced_at else None
        ),
        "hubspot_list_total": state.get("last_successful_list_total"),
    }


def refresh_hubspot_snapshot(run_id, now):
    list_id = str(env("HUBSPOT_TARGET_LIST_ID", "19544")).strip()
    started = time.time()
    conn = None
    fresh = False
    try:
        conn = target_db_connect()
        log_event("hubspot_sync_started", run_id=run_id, list_id=list_id)
        rows, metadata = build_hubspot_snapshot(list_id, run_id, now)
        persist_hubspot_snapshot(conn, list_id, rows, metadata)
        fresh = True
        log_event(
            "hubspot_sync_completed",
            run_id=run_id,
            list_id=list_id,
            sync_id=metadata["sync_id"],
            list_total=metadata["list_total"],
            snapshot_hash=metadata["snapshot_hash"],
            elapsed_seconds=round(time.time() - started, 3),
        )
    except Exception as exc:
        if conn is not None:
            conn.rollback()
            try:
                record_hubspot_sync_failure(conn, list_id, str(exc))
            except Exception as state_exc:
                conn.rollback()
                log_event(
                    "hubspot_sync_failure_state_write_failed",
                    run_id=run_id,
                    error_type=type(state_exc).__name__,
                    error=str(state_exc)[:500],
                )
        log_event(
            "hubspot_sync_failed",
            run_id=run_id,
            list_id=list_id,
            error_type=type(exc).__name__,
            error=str(exc)[:500],
            elapsed_seconds=round(time.time() - started, 3),
        )

    state = {}
    if conn is not None:
        try:
            state = read_hubspot_sync_state(conn, list_id)
        except Exception as exc:
            log_event(
                "hubspot_sync_state_read_failed",
                run_id=run_id,
                error_type=type(exc).__name__,
                error=str(exc)[:500],
            )
        finally:
            conn.close()
    result = snapshot_message_metadata(state, fresh, now)
    log_event(
        "hubspot_snapshot_selected",
        run_id=run_id,
        list_id=list_id,
        sync_id=result["hubspot_sync_id"],
        snapshot_status=result["hubspot_snapshot_status"],
        snapshot_age_seconds=result["hubspot_snapshot_age_seconds"],
        list_total=result["hubspot_list_total"],
    )
    return result