#!/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", } ICP_CATEGORY_LABELS = { "fintech": "Fintech", "retail": "Retail", "e_commerce": "E-Commerce", "ai": "AI", "igaming": "iGaming", } TRACK_STEP_LABELS = { 0: "N/A", 1: "email", 2: "connected", } DEFAULT_SECTION_LIMIT = 10 LATAM_STOCK_SCOPE_SQL = """ ( source_type = 'active_candidate' AND status = 'active' AND kind = 'external' AND sales_force_region = 'LATAM' AND plan NOT IN ('free_trial', 'new_user', 'closing') ) """ PREVIOUS_UTC_DAY_SIGNUP_SCOPE_SQL = """ ( country_region = 'LATAM' AND created_at >= UTC_DATE() - INTERVAL 1 DAY AND created_at < UTC_DATE() ) """ SUMMARY_TENANT_SCOPE_SQL = f""" ( ( {LATAM_STOCK_SCOPE_SQL} AND processed_at >= UTC_TIMESTAMP() - INTERVAL 24 HOUR ) OR {PREVIOUS_UTC_DAY_SIGNUP_SCOPE_SQL} ) """ 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_one(conn, sql, params=()): conn.ping(reconnect=True) with conn.cursor() as cur: cur.execute(sql, params) return cur.fetchone() or {} 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 fetch_priority_tenants(conn, priority, limit=None): if priority not in {"P0", "P1"}: raise RuntimeError(f"unsupported priority: {priority}") 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, icp_categories, source_type, total_score, priority, track_step, processed_at FROM latam_active_tenants WHERE {SUMMARY_TENANT_SCOPE_SQL} AND priority = %s ORDER BY total_score DESC, tenant_avg_qps DESC, cluster_count DESC, tenant_id ASC LIMIT %s """, (priority, 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", "latam_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.icp_categories, a.total_score, a.priority, 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 sales_force_region = 'LATAM' AND COALESCE(event_time, created_at) >= UTC_TIMESTAMP() - INTERVAL 7 DAY GROUP BY tenant_id ) i LEFT JOIN latam_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 fetch_summary_metrics(conn): tenant_metrics = fetch_one( conn, f""" WITH scoped_tenants AS ( SELECT a.*, CASE WHEN {PREVIOUS_UTC_DAY_SIGNUP_SCOPE_SQL} THEN 1 ELSE 0 END AS is_previous_day_signup, CASE WHEN {SUMMARY_TENANT_SCOPE_SQL} THEN 1 ELSE 0 END AS is_summary_tenant FROM latam_active_tenants a ) SELECT COUNT(CASE WHEN is_previous_day_signup = 1 THEN 1 END) AS signup_total, COUNT(CASE WHEN is_previous_day_signup = 1 AND LOWER(TRIM(COALESCE(entity_ai_category, ''))) IN ( 'company_ai_core', 'company_ai_related', 'company_non_ai', 'institution_ai_related', 'institution_non_ai' ) THEN 1 END) AS signup_non_individual, COUNT(CASE WHEN is_previous_day_signup = 1 AND entity_ai_category IN ( 'company_ai_core', 'company_ai_related', 'institution_ai_related' ) THEN 1 END) AS signup_ai_related, COUNT(CASE WHEN is_summary_tenant = 1 AND priority = 'P0' THEN 1 END) AS priority_p0_total, COUNT(CASE WHEN is_summary_tenant = 1 AND priority = 'P1' THEN 1 END) AS priority_p1_total, COUNT(CASE WHEN is_summary_tenant = 1 AND JSON_LENGTH(icp_categories) > 0 THEN 1 END) AS icp_match_total FROM scoped_tenants """, ) import_table = safe_table_name(env("IMPORT_EVENT_COUNT_TABLE", "latam_tenant_events")) import_metrics = fetch_one( conn, f""" SELECT COUNT(DISTINCT tenant_id) AS recent_import_total, COUNT(*) AS recent_import_event_total FROM {import_table} WHERE event_type IN ('CreateImport', 'ImportData') AND LOWER(COALESCE(event_status, '')) = 'succeeded' AND sales_force_region = 'LATAM' AND COALESCE(event_time, created_at) >= UTC_TIMESTAMP() - INTERVAL 7 DAY """, ) return {**tenant_metrics, **import_metrics} def is_ai_related(row): category = str(row.get("entity_ai_category") or "").strip().lower() return category in AI_CATEGORIES 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 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): raw_qps = row.get("tenant_avg_qps") if raw_qps is None: return "N/A" qps = float_value(raw_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 icp_categories_label(row): value = row.get("icp_categories") if not value: return "-" if isinstance(value, str): try: value = json.loads(value) except json.JSONDecodeError: value = value.split(",") if not isinstance(value, (list, tuple, set)): return "-" labels = [] for category in value: key = str(category or "").strip().lower() label = ICP_CATEGORY_LABELS.get(key) if label and label not in labels: labels.append(label) return "/".join(labels) if labels else "-" 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"icp={icp_categories_label(row)}") 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 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") p0_rows = report.get("p0_tenants", []) p1_rows = report.get("p1_tenants", []) import_rows = report.get("recent_imports", []) return { "msg_type": "interactive", "card": { "config": {"wide_screen_mode": True}, "header": { "template": "blue", "title": feishu_text(f"LATAM 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"- ICP-matched tenants: **{metrics['icp_match_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( f"**P0 Tenants (showing {len(p0_rows)} of {metrics['priority_p0_total']})**\n" + "\n".join(priority_lines(p0_rows, "P0")) ), }, {"tag": "hr"}, { "tag": "div", "text": feishu_md( f"**P1 Tenants (showing {len(p1_rows)} of {metrics['priority_p1_total']})**\n" + "\n".join(priority_lines(p1_rows, "P1")) ), }, {"tag": "hr"}, { "tag": "div", "text": feishu_md( f"**Imports in Last 7 Days " f"(showing {len(import_rows)} of {metrics['recent_import_total']})**\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: metrics = fetch_summary_metrics(conn) p0_rows = fetch_priority_tenants(conn, "P0") p1_rows = fetch_priority_tenants(conn, "P1") import_rows = fetch_recent_import_tenants(conn) finally: conn.close() report = { "p0_tenants": p0_rows, "p1_tenants": p1_rows, "recent_imports": import_rows, } send_feishu_card(build_card(metrics, report)) return { "metrics": metrics, "signup_count": int_value(metrics.get("signup_total")), "priority_count": int_value(metrics.get("priority_p0_total")) + int_value(metrics.get("priority_p1_total")), "recent_import_count": len(import_rows), "elapsed_seconds": round((datetime.now(timezone.utc) - started).total_seconds(), 3), }