#!/usr/bin/env python3
import json
import os
import re
import ssl
import urllib.parse
import urllib.request
from datetime import datetime, timezone
from zoneinfo import ZoneInfo

import pymysql


AI_CATEGORIES = {
    "company_ai_core",
    "company_ai_related",
    "institution_ai_related",
    "AI-related company/institution",
    "AI-used but not core AI business",
}

TRACK_STEP_LABELS = {
    0: "N/A",
    1: "email",
    2: "connected",
}

DEFAULT_SECTION_LIMIT = 500

SUMMARY_TENANT_SCOPE_SQL = """
(
  (
    source_type = 'active_candidate'
    AND processed_at >= UTC_TIMESTAMP() - INTERVAL 24 HOUR
  )
  OR (
    source_type = 'new_signup'
    AND created_at >= UTC_DATE() - INTERVAL 1 DAY
    AND created_at < UTC_DATE()
  )
)
"""

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 mysql_connect():
    database_url = env("TARGET_DATABASE_URL") or env("DATABASE_URL")
    if database_url:
        return mysql_connect_url(database_url)

    return pymysql.connect(
        host=env("TARGET_DB_HOST", required=True),
        port=int(env("TARGET_DB_PORT", "3306")),
        user=env("TARGET_DB_USER", required=True),
        password=env("TARGET_DB_PASSWORD", required=True),
        database=env("TARGET_DB_NAME", required=True),
        charset="utf8mb4",
        cursorclass=pymysql.cursors.DictCursor,
        autocommit=True,
        connect_timeout=10,
        read_timeout=60,
        write_timeout=60,
    )


def mysql_connect_url(database_url):
    parsed = urllib.parse.urlparse(database_url)
    if not parsed.hostname:
        raise RuntimeError("TARGET_DATABASE_URL must include a host")
    database = parsed.path.lstrip("/")
    if not database:
        raise RuntimeError("TARGET_DATABASE_URL must include a database name")

    ssl_ctx = ssl.create_default_context()
    ssl_ctx.minimum_version = ssl.TLSVersion.TLSv1_2
    ca_path = env("TARGET_TIDB_SSL_CA_PATH") or env("TIDB_SSL_CA_PATH")
    if ca_path:
        ssl_ctx.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=True,
        connect_timeout=10,
        read_timeout=60,
        write_timeout=60,
        ssl=ssl_ctx,
    )


def fetch_rows(conn, sql, params):
    conn.ping(reconnect=True)
    with conn.cursor() as cur:
        cur.execute(sql, params)
        return cur.fetchall()


def fetch_monitored_tenants(conn):
    return fetch_rows(
        conn,
        """
        SELECT
          tenant_id,
          name,
          email,
          email_domain,
          country,
          plan,
          has_credit_card,
          billing_config_id,
          cluster_count,
          starter_cluster_count,
          essential_cluster_count,
          premium_cluster_count,
          dedicated_cluster_count,
          tenant_avg_qps,
          entity_ai_category,
          business_description,
          official_website,
          processed_at
        FROM emea_active_tenants
        WHERE processed_at >= UTC_TIMESTAMP() - INTERVAL 24 HOUR
        ORDER BY tenant_avg_qps DESC, cluster_count DESC, tenant_id ASC
        LIMIT 200
        """,
        (),
    )


def fetch_new_signups(conn):
    return fetch_rows(
        conn,
        """
        SELECT
          tenant_id,
          name,
          email,
          email_domain,
          country,
          plan,
          total_score,
          priority,
          key_lead,
          track_step,
          source_type,
          entity_ai_category,
          business_description,
          official_website,
          created_at,
          processed_at
        FROM emea_active_tenants
        WHERE source_type = 'new_signup'
          AND created_at >= UTC_DATE() - INTERVAL 1 DAY
          AND created_at < UTC_DATE()
        ORDER BY created_at DESC, tenant_id ASC
        LIMIT 200
        """,
        (),
    )


def int_env(name, default):
    try:
        return int(str(env(name, str(default))).strip())
    except (TypeError, ValueError):
        return default


def safe_table_name(name):
    table = str(name or "").strip()
    if not re.fullmatch(r"[A-Za-z0-9_]+", table):
        raise RuntimeError(f"invalid table name: {table}")
    return table


def quote_identifier(name):
    identifier = str(name or "").strip()
    if not re.fullmatch(r"[A-Za-z0-9_]+", identifier):
        raise RuntimeError(f"invalid identifier: {identifier}")
    return f"`{identifier}`"


def table_columns(conn, table):
    rows = fetch_rows(conn, f"SHOW COLUMNS FROM {quote_identifier(table)}", ())
    return {str(row.get("Field") or "") for row in rows}


def first_existing_column(columns, candidates, required_name):
    for candidate in candidates:
        if candidate in columns:
            return candidate
    raise RuntimeError(f"{required_name} column not found; available columns: {', '.join(sorted(columns))}")


def fetch_priority_tenants(conn, limit=None):
    return fetch_rows(
        conn,
        f"""
        SELECT
          tenant_id,
          name,
          email,
          email_domain,
          country,
          plan,
          has_credit_card,
          cluster_count,
          starter_cluster_count,
          essential_cluster_count,
          premium_cluster_count,
          dedicated_cluster_count,
          tenant_avg_qps,
          entity_ai_category,
          source_type,
          total_score,
          priority,
          key_lead,
          track_step,
          processed_at
        FROM emea_active_tenants
        WHERE {SUMMARY_TENANT_SCOPE_SQL}
          AND priority IN ('P0', 'P1')
        ORDER BY
          FIELD(priority, 'P0', 'P1'),
          total_score DESC,
          tenant_avg_qps DESC,
          cluster_count DESC,
          tenant_id ASC
        LIMIT %s
        """,
        (limit or int_env("SUMMARY_SECTION_LIMIT", DEFAULT_SECTION_LIMIT),),
    )


def fetch_key_leads(conn, limit=None):
    return fetch_rows(
        conn,
        f"""
        SELECT
          tenant_id,
          name,
          email,
          email_domain,
          country,
          plan,
          has_credit_card,
          cluster_count,
          starter_cluster_count,
          essential_cluster_count,
          premium_cluster_count,
          dedicated_cluster_count,
          tenant_avg_qps,
          entity_ai_category,
          source_type,
          total_score,
          priority,
          key_lead,
          track_step,
          processed_at
        FROM emea_active_tenants
        WHERE {SUMMARY_TENANT_SCOPE_SQL}
          AND key_lead = 1
        ORDER BY
          FIELD(priority, 'P0', 'P1', 'P2'),
          total_score DESC,
          tenant_avg_qps DESC,
          cluster_count DESC,
          tenant_id ASC
        LIMIT %s
        """,
        (limit or int_env("SUMMARY_SECTION_LIMIT", DEFAULT_SECTION_LIMIT),),
    )


def fetch_recent_import_tenants(conn, limit=None):
    import_table = safe_table_name(env("IMPORT_EVENT_COUNT_TABLE", "emea_tenant_events"))
    sql = f"""
        SELECT
          i.tenant_id,
          COALESCE(a.name, i.tenant_name) AS name,
          a.created_at,
          a.email,
          a.email_domain,
          COALESCE(a.country, i.country) AS country,
          COALESCE(a.plan, i.tenant_plan) AS plan,
          a.has_credit_card,
          a.cluster_count,
          a.starter_cluster_count,
          a.essential_cluster_count,
          a.premium_cluster_count,
          a.dedicated_cluster_count,
          a.tenant_avg_qps,
          a.entity_ai_category,
          a.total_score,
          a.priority,
          a.key_lead,
          a.track_step,
          a.source_type,
          i.import_event_count,
          i.last_import_time
        FROM (
          SELECT
            tenant_id,
            MAX(tenant_name) AS tenant_name,
            MAX(tenant_plan) AS tenant_plan,
            MAX(country) AS country,
            COUNT(*) AS import_event_count,
            MAX(COALESCE(event_time, created_at)) AS last_import_time
          FROM {import_table}
          WHERE event_type IN ('CreateImport', 'ImportData')
            AND LOWER(COALESCE(event_status, '')) = 'succeeded'
            AND COALESCE(event_time, created_at) >= UTC_TIMESTAMP() - INTERVAL 7 DAY
          GROUP BY tenant_id
        ) i
        LEFT JOIN emea_active_tenants a ON a.tenant_id = CAST(i.tenant_id AS CHAR)
        ORDER BY
          i.import_event_count DESC,
          i.last_import_time DESC,
          i.tenant_id ASC
        LIMIT %s
        """
    return fetch_rows(
        conn,
        sql,
        (limit or int_env("SUMMARY_SECTION_LIMIT", DEFAULT_SECTION_LIMIT),),
    )


def is_individual(row):
    return str(row.get("entity_ai_category") or "").strip().lower() == "individual"


def is_ai_related(row):
    category = str(row.get("entity_ai_category") or "").strip()
    normalized = category.lower()
    return category in AI_CATEGORIES or "ai" in normalized


def active_score(row):
    score = 0
    if row.get("has_credit_card"):
        score += 3
    try:
        score += min(int(row.get("cluster_count") or 0), 5)
    except (TypeError, ValueError):
        pass
    try:
        qps = float(row.get("tenant_avg_qps") or 0)
        if qps >= 100:
            score += 5
        elif qps >= 10:
            score += 3
        elif qps > 0:
            score += 1
    except (TypeError, ValueError):
        pass
    if is_ai_related(row):
        score += 3
    if not is_individual(row):
        score += 1
    return score


def int_value(value):
    try:
        return int(value or 0)
    except (TypeError, ValueError):
        return 0


def float_value(value):
    try:
        return float(value or 0)
    except (TypeError, ValueError):
        return 0.0


def stable_attention_sort_key(row):
    qps = float_value(row.get("tenant_avg_qps"))
    cluster_count = int_value(row.get("cluster_count"))
    return (
        -active_score(row),
        -qps,
        -cluster_count,
        -int(bool(row.get("has_credit_card"))),
        -int(is_ai_related(row)),
        str(row.get("tenant_id") or ""),
    )


def attention_reason(row):
    signals = [f"score={active_score(row)}"]
    if row.get("has_credit_card"):
        signals.append("has credit card")
    cluster_count = int_value(row.get("cluster_count"))
    if cluster_count > 0:
        signals.append(f"{cluster_count} clusters")
    qps = float_value(row.get("tenant_avg_qps"))
    if qps > 0:
        signals.append(f"{qps:.3f} avg QPS")
    if is_ai_related(row):
        signals.append(f"AI-related: {row.get('entity_ai_category')}")
    elif not is_individual(row):
        signals.append(f"non-individual: {row.get('entity_ai_category')}")
    return ", ".join(signals)


def suggested_attention_action(row):
    if row.get("has_credit_card") and float_value(row.get("tenant_avg_qps")) >= 10:
        return "Prioritize sales/CS follow-up; validate use case and expansion potential."
    if float_value(row.get("tenant_avg_qps")) >= 10:
        return "Review recent usage and consider outreach for workload qualification."
    if int_value(row.get("cluster_count")) >= 5:
        return "Review cluster footprint and confirm whether the tenant needs onboarding or account support."
    if is_ai_related(row):
        return "Check company context and consider targeted AI workload follow-up."
    return "Review in DevOps and consider sales/CS follow-up."


def deterministic_attention_items(rows, limit=8):
    ranked = sorted(rows, key=stable_attention_sort_key)
    return [
        {
            "tenant_id": row.get("tenant_id"),
            "name": row.get("name"),
            "reason": attention_reason(row),
            "suggested_action": suggested_attention_action(row),
        }
        for row in ranked[:limit]
    ]


def is_engaged_tenant(row):
    if row.get("has_credit_card"):
        return True
    try:
        if int(row.get("cluster_count") or 0) > 0:
            return True
    except (TypeError, ValueError):
        pass
    try:
        return float(row.get("tenant_avg_qps") or 0) > 0
    except (TypeError, ValueError):
        return False


def positive_int(value):
    try:
        return int(value or 0) > 0
    except (TypeError, ValueError):
        return False


def positive_float(value):
    try:
        return float(value or 0) > 0
    except (TypeError, ValueError):
        return False


def compact_monitored_rows(rows):
    ranked = sorted(rows, key=stable_attention_sort_key)
    return [
        {
            "tenant_id": row.get("tenant_id"),
            "name": row.get("name"),
            "email_domain": row.get("email_domain"),
            "country": row.get("country"),
            "plan": row.get("plan"),
            "has_credit_card": bool(row.get("has_credit_card")),
            "cluster_count": row.get("cluster_count"),
            "starter_cluster_count": row.get("starter_cluster_count"),
            "essential_cluster_count": row.get("essential_cluster_count"),
            "premium_cluster_count": row.get("premium_cluster_count"),
            "dedicated_cluster_count": row.get("dedicated_cluster_count"),
            "tenant_avg_qps": float(row.get("tenant_avg_qps") or 0),
            "entity_ai_category": row.get("entity_ai_category"),
            "business_description": row.get("business_description"),
            "official_website": row.get("official_website"),
            "score": active_score(row),
        }
        for row in ranked[:30]
    ]


def compact_signup_rows(rows):
    interesting = [row for row in rows if not is_individual(row) or is_ai_related(row)]
    return [
        {
            "tenant_id": row.get("tenant_id"),
            "name": row.get("name"),
            "email_domain": row.get("email_domain"),
            "country": row.get("country"),
            "plan": row.get("plan"),
            "entity_ai_category": row.get("entity_ai_category"),
            "business_description": row.get("business_description"),
            "official_website": row.get("official_website"),
        }
        for row in interesting[:30]
    ]


def build_metrics(monitored_rows, signup_rows):
    return {
        "monitored_total": len(monitored_rows),
        "engaged_total": sum(1 for row in monitored_rows if is_engaged_tenant(row)),
        "monitored_ai_related": sum(1 for row in monitored_rows if is_ai_related(row)),
        "monitored_with_credit_card": sum(1 for row in monitored_rows if row.get("has_credit_card")),
        "monitored_with_clusters": sum(1 for row in monitored_rows if positive_int(row.get("cluster_count"))),
        "monitored_with_qps": sum(1 for row in monitored_rows if positive_float(row.get("tenant_avg_qps"))),
        "signup_total": len(signup_rows),
        "signup_non_individual": sum(1 for row in signup_rows if not is_individual(row)),
        "signup_ai_related": sum(1 for row in signup_rows if is_ai_related(row)),
    }


def build_daily_report_metrics(priority_rows, key_lead_rows, import_rows, signup_rows=None):
    signup_rows = signup_rows or []
    return {
        "priority_total": len(priority_rows),
        "priority_p0_total": sum(1 for row in priority_rows if row.get("priority") == "P0"),
        "priority_p1_total": sum(1 for row in priority_rows if row.get("priority") == "P1"),
        "key_lead_total": len(key_lead_rows),
        "recent_import_total": len(import_rows),
        "recent_import_event_total": sum(int_value(row.get("import_event_count")) for row in import_rows),
        "signup_total": len(signup_rows),
        "signup_non_individual": sum(1 for row in signup_rows if not is_individual(row)),
        "signup_ai_related": sum(1 for row in signup_rows if is_ai_related(row)),
    }


def llm_summary(metrics, monitored_rows, signup_rows):
    selected_attention = deterministic_attention_items(monitored_rows)
    api_key = env("OPENAI_API_KEY")
    if not api_key:
        return fallback_summary(
            metrics,
            monitored_rows,
            signup_rows,
            selected_attention,
            "OPENAI_API_KEY is not configured.",
        )

    body = {
        "model": env("OPENAI_SUMMARY_MODEL", env("OPENAI_WEB_SEARCH_MODEL", "gpt-4.1-mini")),
        "temperature": 0,
        "input": (
            "You are summarizing an EMEA tenant monitoring report for sales/CS follow-up. "
            "Rows from emea_active_tenants are monitored tenants, not necessarily active tenants. "
            "Infer engagement/activity from has_credit_card, cluster_count, and tenant_avg_qps. "
            "Use the provided structured data only. Return valid JSON with keys: "
            "headline, tenants_worth_attention, signup_summary, risks. "
            "tenants_worth_attention must contain exactly the same tenants, in the same order, as "
            "selected_monitored_tenants. You may rewrite reason and suggested_action for clarity, but do not add, "
            "remove, reorder, or replace tenant_id values. "
            "signup_summary must be a concise paragraph. When signup_summary mentions a specific new signup "
            "by name, immediately append its tenant id as a markdown link in this exact format: "
            "Name ([tenant_id](https://ops.tidbcloud.com/orgs/tenant_id)). "
            "Only link tenant ids that appear in new_signup_candidates. risks must be an array of short strings.\n\n"
            + json.dumps(
                {
                    "metrics": metrics,
                    "selected_monitored_tenants": selected_attention,
                    "monitored_tenant_candidates": monitored_rows,
                    "new_signup_candidates": signup_rows,
                },
                ensure_ascii=False,
                default=str,
            )
        ),
    }
    req = urllib.request.Request(
        "https://api.openai.com/v1/responses",
        data=json.dumps(body).encode("utf-8"),
        headers={
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json",
        },
        method="POST",
    )
    try:
        with urllib.request.urlopen(req, timeout=int(env("LLM_TIMEOUT_SECONDS", "60"))) as resp:
            raw = resp.read().decode("utf-8")
        response = json.loads(raw)
        text = response.get("output_text") or extract_response_text(response)
        summary = json.loads(clean_json(text))
        summary["tenants_worth_attention"] = normalize_attention_items(
            summary.get("tenants_worth_attention"),
            selected_attention,
        )
        return summary
    except Exception as exc:
        return fallback_summary(
            metrics,
            monitored_rows,
            signup_rows,
            selected_attention,
            f"LLM summary failed: {exc}",
        )


def normalize_attention_items(llm_items, selected_attention):
    by_id = {
        str(item.get("tenant_id")): item
        for item in (llm_items or [])
        if item.get("tenant_id")
    }
    normalized = []
    for selected in selected_attention:
        tenant_id = str(selected.get("tenant_id") or "")
        llm_item = by_id.get(tenant_id, {})
        normalized.append(
            {
                "tenant_id": selected.get("tenant_id"),
                "name": selected.get("name"),
                "reason": llm_item.get("reason") or selected.get("reason"),
                "suggested_action": llm_item.get("suggested_action") or selected.get("suggested_action"),
            }
        )
    return normalized


def extract_response_text(response):
    chunks = []
    for item in response.get("output", []) or []:
        for content in item.get("content", []) or []:
            if content.get("type") in {"output_text", "text"}:
                chunks.append(content.get("text") or "")
    return "\n".join(chunks).strip()


def clean_json(text):
    text = str(text or "").strip()
    if text.startswith("```json"):
        text = text[len("```json") :]
    if text.startswith("```"):
        text = text[len("```") :]
    if text.endswith("```"):
        text = text[: -len("```")]
    return text.strip()


def fallback_summary(metrics, monitored_rows, signup_rows, selected_attention, reason):
    signup_mentions = linked_signup_mentions(signup_rows[:3])
    signup_detail = f" Notable signups: {signup_mentions}." if signup_mentions else ""
    return {
        "headline": "EMEA tenant monitor summary generated with fallback rules.",
        "tenants_worth_attention": selected_attention,
        "signup_summary": (
            f"{metrics['signup_total']} new signups in the previous UTC day; "
            f"{metrics['signup_non_individual']} non-individual; "
            f"{metrics['signup_ai_related']} AI-related."
            f"{signup_detail}"
        ),
        "risks": [reason],
    }


def linked_signup_mentions(rows):
    mentions = []
    for row in rows:
        tenant_id = row.get("tenant_id")
        name = row.get("name") or tenant_id
        if tenant_id:
            mentions.append(f"{name} ([{tenant_id}](https://ops.tidbcloud.com/orgs/{tenant_id}))")
        elif name:
            mentions.append(str(name))
    return ", ".join(mentions)


def tenant_ref(row):
    tenant_id = row.get("tenant_id")
    tenant_label = row.get("name") or tenant_id
    if tenant_id:
        return f"{tenant_label} ([{tenant_id}](https://ops.tidbcloud.com/orgs/{tenant_id}))"
    return str(tenant_label or "Unknown tenant")


def track_step_label(row):
    track_step = int_value(row.get("track_step"))
    return TRACK_STEP_LABELS.get(track_step, str(track_step))


def qps_label(row):
    qps = float_value(row.get("tenant_avg_qps"))
    if qps <= 0:
        return "0"
    return f"{qps:.3f}"


def credit_card_label(row):
    return "yes" if row.get("has_credit_card") else "no"


def cluster_count_label(row):
    return (
        f"{int_value(row.get('cluster_count'))}"
        f"({int_value(row.get('starter_cluster_count'))}/"
        f"{int_value(row.get('essential_cluster_count'))}/"
        f"{int_value(row.get('premium_cluster_count'))}/"
        f"{int_value(row.get('dedicated_cluster_count'))})"
    )


def base_tenant_detail(row, include_priority=False, include_usage=True, include_created_at=False):
    details = []
    if include_priority:
        details.append(f"priority={row.get('priority') or '-'}")
    if row.get("total_score") is not None:
        details.append(f"score={row.get('total_score')}")
    if include_priority is False and row.get("priority") is not None and not include_usage:
        details.append(f"priority={row.get('priority') or '-'}")
    details.append(f"track_step={track_step_label(row)}")
    if include_created_at:
        details.append(f"tenant_created_at={row.get('created_at') or '-'}")
    if include_usage:
        details.append(f"clusters={cluster_count_label(row)}")
        details.append(f"qps={qps_label(row)}")
        details.append(f"credit_card={credit_card_label(row)}")
    details.append(f"plan={row.get('plan') or '-'}")
    details.append(f"category={row.get('entity_ai_category') or '-'}")
    details.append(f"email={row.get('email') or '-'}")
    return ", ".join(details)


def priority_lines(rows, priority):
    filtered = [row for row in rows if row.get("priority") == priority]
    if not filtered:
        return [f"- No {priority} tenants identified."]
    return [
        f"- **{tenant_ref(row)}**: {base_tenant_detail(row)}"
        for row in filtered
    ]


def key_lead_lines(rows):
    if not rows:
        return ["- No key_lead=1 tenants identified."]
    return [
        f"- **{tenant_ref(row)}**: {base_tenant_detail(row, include_priority=True)}"
        for row in rows
    ]


def signup_lines(rows):
    interesting = [row for row in rows if not is_individual(row) or is_ai_related(row)]
    selected = interesting[:10]
    if not selected:
        return ["- No non-individual or AI-related new signups in the previous UTC day."]
    return [
        f"- **{tenant_ref(row)}**: {base_tenant_detail(row)}"
        for row in selected
    ]


def import_lines(rows):
    if not rows:
        return ["- No tenants with successful imports in the last 7 days."]
    lines = []
    for row in rows:
        last_import_time = row.get("last_import_time")
        details = [
            f"import_event_count={int_value(row.get('import_event_count'))}",
            f"last_import_time={last_import_time}" if last_import_time else "last_import_time=-",
            base_tenant_detail(row, include_usage=False, include_created_at=True),
        ]
        lines.append(f"- **{tenant_ref(row)}**: " + ", ".join(details))
    return lines


def feishu_text(text):
    return {"tag": "plain_text", "content": text}


def feishu_md(text):
    return {"tag": "lark_md", "content": text}


def build_card(metrics, report):
    shanghai_today = datetime.now(ZoneInfo("Asia/Shanghai")).strftime("%Y-%m-%d")
    priority_rows = report.get("priority_tenants", [])
    key_lead_rows = report.get("key_leads", [])
    import_rows = report.get("recent_imports", [])
    signup_rows = report.get("new_signups", [])

    return {
        "msg_type": "interactive",
        "card": {
            "config": {"wide_screen_mode": True},
            "header": {
                "template": "blue",
                "title": feishu_text(f"EMEA Tenants Monitor Daily Summary - {shanghai_today}"),
            },
            "elements": [
                {
                    "tag": "div",
                    "text": feishu_md(
                        f"**Daily Signals**\n"
                        f"- New signups in previous UTC day: **{metrics['signup_total']}**\n"
                        f"- New signup non-individual: **{metrics['signup_non_individual']}**\n"
                        f"- New signup AI-related: **{metrics['signup_ai_related']}**\n"
                        f"- P0 tenants: **{metrics['priority_p0_total']}**\n"
                        f"- P1 tenants: **{metrics['priority_p1_total']}**\n"
                        f"- key_lead=1 tenants: **{metrics['key_lead_total']}**\n"
                        f"- Tenants with imports in last 7 days: **{metrics['recent_import_total']}**\n"
                        f"- Import events in last 7 days: **{metrics['recent_import_event_total']}**"
                    ),
                },
                {"tag": "hr"},
                {
                    "tag": "div",
                    "text": feishu_md(
                        "**P0 Tenants**\n"
                        + "\n".join(priority_lines(priority_rows, "P0"))
                    ),
                },
                {"tag": "hr"},
                {
                    "tag": "div",
                    "text": feishu_md(
                        "**P1 Tenants**\n"
                        + "\n".join(priority_lines(priority_rows, "P1"))
                    ),
                },
                {"tag": "hr"},
                {
                    "tag": "div",
                    "text": feishu_md("**key_lead=1**\n" + "\n".join(key_lead_lines(key_lead_rows))),
                },
                {"tag": "hr"},
                {
                    "tag": "div",
                    "text": feishu_md(
                        "**Imports in Last 7 Days**\n" + "\n".join(import_lines(import_rows))
                    ),
                },
            ],
        },
    }


def send_feishu_card(card):
    webhook_url = env("FEISHU_WEBHOOK_URL", required=True)
    req = urllib.request.Request(
        webhook_url,
        data=json.dumps(card).encode("utf-8"),
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    with urllib.request.urlopen(req, timeout=20) as resp:
        body = resp.read().decode("utf-8")
    result = json.loads(body)
    if result.get("code") not in (0, None):
        raise RuntimeError(f"Feishu webhook failed: {body}")
    return result


def lambda_handler(event, context):
    started = datetime.now(timezone.utc)
    conn = mysql_connect()
    try:
        signup_rows = fetch_new_signups(conn)
        priority_rows = fetch_priority_tenants(conn)
        key_lead_rows = fetch_key_leads(conn)
        import_rows = fetch_recent_import_tenants(conn)
    finally:
        conn.close()

    metrics = build_daily_report_metrics(priority_rows, key_lead_rows, import_rows, signup_rows)
    report = {
        "new_signups": signup_rows,
        "priority_tenants": priority_rows,
        "key_leads": key_lead_rows,
        "recent_imports": import_rows,
    }
    send_feishu_card(build_card(metrics, report))
    return {
        "metrics": metrics,
        "signup_count": len(signup_rows),
        "priority_count": len(priority_rows),
        "key_lead_count": len(key_lead_rows),
        "recent_import_count": len(import_rows),
        "elapsed_seconds": round((datetime.now(timezone.utc) - started).total_seconds(), 3),
    }