#!/usr/bin/env python3 import json import os import ssl import time import urllib.error import urllib.parse import urllib.request from datetime import datetime, timezone from decimal import Decimal, InvalidOperation from email.utils import parseaddr import pymysql import boto3 from cluster_metrics import DEVOPS_BASE_URL, get_json, get_tenant_cluster_stats TENANT_COLUMNS = [ "id", "created_at", "updated_at", "name", "email", "country", "time_zone", "status", "kind", "plan", "alias", "country_region", "sales_owner_email", "custom_id", "segmentation", "sales_force_region", "sales_force_kind", "tam_name", ] DEFAULT_TENANT_SELECT_SQL = """ SELECT id, created_at, updated_at, name, email, country, time_zone, status, kind, plan, alias, country_region, sales_owner_email, custom_id, segmentation, sales_force_region, sales_force_kind, tam_name FROM tenants WHERE status = 'active' AND kind = 'external' AND sales_force_region = 'EMEA' AND plan NOT IN ('free_trial', 'new_user') AND country IN ( -- 核心及西欧 'GB', 'DE', 'FR', 'IT', 'NL', 'BE', 'CH', 'AT', 'IE', 'LU', 'MC', 'AD', -- 北欧 'SE', 'DK', 'FI', 'NO', 'IS', 'AX', 'FO', -- 南欧 'ES', 'PT', 'GR', 'MT', 'CY', 'GI', 'VA', 'SM', -- 中东欧 'PL', 'CZ', 'HU', 'RO', 'SK', 'SI', 'HR', 'BG', 'EE', 'LV', 'LT', 'RS', 'BA', 'ME', 'MK', 'AL', 'TR', -- 特定保留 'IL', 'IN' ); """ DEFAULT_NEW_SIGNUP_SELECT_SQL = """ SELECT id, created_at, updated_at, name, email, country, time_zone, status, kind, plan, alias, country_region, sales_owner_email, custom_id, segmentation, sales_force_region, sales_force_kind, tam_name FROM tenants WHERE created_at >= UTC_DATE() - INTERVAL 1 DAY AND created_at < UTC_DATE() AND country_region = 'EMEA' AND country IN ( -- 核心及西欧 'GB', 'DE', 'FR', 'IT', 'NL', 'BE', 'CH', 'AT', 'IE', 'LU', 'MC', 'AD', -- 北欧 'SE', 'DK', 'FI', 'NO', 'IS', 'AX', 'FO', -- 南欧 'ES', 'PT', 'GR', 'MT', 'CY', 'GI', 'VA', 'SM', -- 中东欧 'PL', 'CZ', 'HU', 'RO', 'SK', 'SI', 'HR', 'BG', 'EE', 'LV', 'LT', 'RS', 'BA', 'ME', 'MK', 'AL', 'TR', -- 特定保留 'IL', 'IN' ); """ FREE_EMAIL_DOMAINS = { "gmail.com", "googlemail.com", "hotmail.com", "outlook.com", "live.com", "msn.com", "icloud.com", "me.com", "mac.com", "yahoo.com", "ymail.com", "aol.com", "proton.me", "protonmail.com", "qq.com", "163.com", "126.com", } _last_billing_request_at = 0.0 _sqs_client = None 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 sqs_client(): global _sqs_client if _sqs_client is None: _sqs_client = boto3.client("sqs") return _sqs_client def mysql_connect(prefix): database_url = env(f"{prefix}_DATABASE_URL") if not database_url and prefix == "TARGET": database_url = env("DATABASE_URL") if database_url: return mysql_connect_url(database_url, prefix) return pymysql.connect( host=env(f"{prefix}_DB_HOST", required=True), port=int(env(f"{prefix}_DB_PORT", "3306")), user=env(f"{prefix}_DB_USER", required=True), password=env(f"{prefix}_DB_PASSWORD", required=True), database=env(f"{prefix}_DB_NAME", required=True), charset="utf8mb4", cursorclass=pymysql.cursors.DictCursor, autocommit=False, connect_timeout=10, read_timeout=60, write_timeout=60, ) def mysql_connect_url(database_url, prefix): parsed = urllib.parse.urlparse(database_url) if not parsed.hostname: raise RuntimeError(f"{prefix}_DATABASE_URL must include a host") database = parsed.path.lstrip("/") if not database: raise RuntimeError(f"{prefix}_DATABASE_URL must include a database name") query = urllib.parse.parse_qs(parsed.query) ssl_ctx = ssl.create_default_context() ssl_ctx.minimum_version = ssl.TLSVersion.TLSv1_2 ca_path = env(f"{prefix}_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=False, connect_timeout=int(first_query_value(query, "connect_timeout", "10")), read_timeout=int(first_query_value(query, "read_timeout", "60")), write_timeout=int(first_query_value(query, "write_timeout", "60")), ssl=ssl_ctx, ) def first_query_value(query, key, default): values = query.get(key) if not values: return default return values[0] or default def ensure_connection(conn): conn.ping(reconnect=True) def list_tenants(conn): sql = env("TENANT_SELECT_SQL", DEFAULT_TENANT_SELECT_SQL).strip() ensure_connection(conn) with conn.cursor() as cur: cur.execute(sql) return cur.fetchall() def list_new_signup_tenants(conn): sql = env("NEW_SIGNUP_SELECT_SQL", DEFAULT_NEW_SIGNUP_SELECT_SQL).strip() ensure_connection(conn) with conn.cursor() as cur: cur.execute(sql) return cur.fetchall() def encode_path(value): return urllib.parse.quote(str(value), safe="") def devops_get(path, devops_token, operator_email=None): headers = { "X-API-TOKEN": devops_token, "X-BILLING-REQUEST-SOURCE": "ops", } if operator_email: headers["X-ACCOUNT-USER-EMAIL"] = operator_email base_url = env("DEVOPS_BASE_URL", DEVOPS_BASE_URL).rstrip("/") return billing_get_json(f"{base_url}{path}", headers) def billing_get_json(url, headers): max_attempts = int(env("BILLING_RETRY_COUNT", "3")) + 1 for attempt in range(1, max_attempts + 1): throttle_billing_request() try: return get_json(url, headers) except urllib.error.HTTPError as exc: if exc.code != 429 or attempt >= max_attempts: raise time.sleep(billing_retry_delay(exc, attempt)) def throttle_billing_request(): global _last_billing_request_at interval = float(env("BILLING_REQUEST_INTERVAL_SECONDS", "2.2")) if interval <= 0: return now = time.monotonic() wait_seconds = _last_billing_request_at + interval - now if wait_seconds > 0: time.sleep(wait_seconds) _last_billing_request_at = time.monotonic() def billing_retry_delay(exc, attempt): retry_after = exc.headers.get("Retry-After") if exc.headers else None if retry_after: try: return min(float(retry_after), 60.0) except ValueError: pass return min(5 * attempt, 60) def is_non_expired_card(card, now): if not isinstance(card, dict): return False try: exp_year = int(str(card.get("exp_year") or "").strip()) exp_month = int(str(card.get("exp_month") or "").strip()) except ValueError: return False if exp_month < 1 or exp_month > 12: return False return exp_year > now.year or (exp_year == now.year and exp_month >= now.month) def find_valid_credit_card(tenant_id, devops_token, operator_email=None, now=None, billing_config_id=None): now = now or datetime.now(timezone.utc) tenant = encode_path(tenant_id) billing_config_id = normalized_billing_config_id(billing_config_id) if billing_config_id: print('billing_config_id exists\n') card = find_valid_credit_card_in_billing_config( tenant, tenant_id, billing_config_id, devops_token, operator_email, now, ) if card: return card return { "has_credit_card": False, "billing_config_id": billing_config_id, "payment_method_id": None, } configs = devops_get( f"/api/v1/devops/billing-proxy/api/v1/billing/tenants/{tenant}/billing_configs", devops_token, operator_email, ) print('request devops billing\n') if not isinstance(configs, list): raise RuntimeError(f"unexpected billing configs response for tenant_id={tenant_id}") first_active_billing_config_id = None for config in configs: if not isinstance(config, dict): continue if str(config.get("status") or "").strip().lower() != "active": continue billing_config_id = config.get("billing_config_id") if not billing_config_id: continue if first_active_billing_config_id is None: first_active_billing_config_id = int(billing_config_id) card = find_valid_credit_card_in_billing_config( tenant, tenant_id, int(billing_config_id), devops_token, operator_email, now, ) if card: return card return { "has_credit_card": False, "billing_config_id": first_active_billing_config_id, "payment_method_id": None, } def find_valid_credit_card_in_billing_config( tenant, tenant_id, billing_config_id, devops_token, operator_email, now, ): methods = devops_get( "/api/v1/devops/billing-proxy/api/v1/billing/" f"tenants/{tenant}/billing_config_id/{encode_path(billing_config_id)}" "/payments/stripe/payment_methods", devops_token, operator_email, ) if not isinstance(methods, list): raise RuntimeError( f"unexpected payment methods response for tenant_id={tenant_id}, " f"billing_config_id={billing_config_id}" ) for method in methods: if not isinstance(method, dict) or not str(method.get("id") or "").strip(): continue card = method.get("card") if not isinstance(card, dict): continue if str(card.get("funding") or "").strip().lower() != "credit": continue if is_non_expired_card(card, now): return { "has_credit_card": True, "billing_config_id": int(billing_config_id), "payment_method_id": str(method.get("id") or "").strip(), } return None def fetch_credit_summary(tenant_id, billing_config_id, devops_token, operator_email=None, now=None): billing_config_id = normalized_billing_config_id(billing_config_id) if not billing_config_id: return { "active_total_credits": Decimal("0"), "latest_credit_grant_days_ago": None, } now = now or datetime.now(timezone.utc) tenant = encode_path(tenant_id) credits = devops_get( "/api/v1/devops/billing-proxy/api/v1/billing/" f"tenants/{tenant}/billing_config_id/{encode_path(billing_config_id)}" "/credits", devops_token, operator_email, ) if not isinstance(credits, list): raise RuntimeError( f"unexpected credits response for tenant_id={tenant_id}, " f"billing_config_id={billing_config_id}" ) active_credits = [ credit for credit in credits if isinstance(credit, dict) and not str(credit.get("closed_reason") or "").strip() ] total = Decimal("0") latest_start = None for credit in active_credits: total += decimal_value(credit.get("total_credits")) start_at = parse_iso_datetime(credit.get("start_date")) if start_at is not None and (latest_start is None or start_at > latest_start): latest_start = start_at latest_days = None if latest_start is not None: latest_days = max((now - latest_start).days, 0) return { "active_total_credits": total, "latest_credit_grant_days_ago": latest_days, } def decimal_value(value): try: return Decimal(str(value or "0").strip()) except (InvalidOperation, ValueError): return Decimal("0") def parse_iso_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 None: return parsed.replace(tzinfo=timezone.utc) return parsed.astimezone(timezone.utc) def normalized_billing_config_id(value): if value is None: return None value = str(value).strip() if not value or value == "0": return None try: return int(value) except ValueError: return None def email_domain(email): address = parseaddr(str(email or "").strip().lower())[1] if "@" not in address: return "" domain = address.rsplit("@", 1)[1].strip(". ") return domain def enqueue_scoring_request(tenant_id, source_type, event_type, now=None): queue_url = env("SCORING_QUEUE_URL", "") if not queue_url: return False now = now or datetime.now(timezone.utc) message = { "tenant_id": str(tenant_id), "source_type": source_type, "event_type": event_type, "processed_at": now.isoformat().replace("+00:00", "Z"), } try: sqs_client().send_message(QueueUrl=queue_url, MessageBody=json.dumps(message)) return True except Exception as exc: print(f"failed to enqueue scoring request for tenant_id={tenant_id}: {exc}") return False def get_cached_domain_intelligence(conn, domain): ensure_connection(conn) with conn.cursor() as cur: cur.execute( """ SELECT email_domain, entity_ai_category, business_description, official_website FROM domain_intelligence WHERE email_domain = %s AND (expires_at IS NULL OR expires_at > UTC_TIMESTAMP()) """, (domain,), ) return cur.fetchone() def get_existing_active_tenant_billing_config_id(conn, tenant_id): ensure_connection(conn) with conn.cursor() as cur: cur.execute( """ SELECT billing_config_id FROM emea_active_tenants WHERE tenant_id = %s """, (str(tenant_id),), ) row = cur.fetchone() if not row: return None return normalized_billing_config_id(row.get("billing_config_id")) def get_existing_active_tenant_enrichment(conn, tenant_id): ensure_connection(conn) with conn.cursor() as cur: cur.execute( """ SELECT has_credit_card, billing_config_id, active_total_credits, latest_credit_grant_days_ago, cluster_count, starter_cluster_count, essential_cluster_count, premium_cluster_count, dedicated_cluster_count, tenant_avg_qps, email_domain, entity_ai_category, business_description, official_website FROM emea_active_tenants WHERE tenant_id = %s """, (str(tenant_id),), ) return cur.fetchone() or {} def cached_or_existing_domain_info(conn, tenant): tenant_id = str(tenant.get("id")) existing = get_existing_active_tenant_enrichment(conn, tenant_id) domain = email_domain(tenant.get("email")) or existing.get("email_domain") or "" cached = get_cached_domain_intelligence(conn, domain) if domain else None if cached: return { "email_domain": domain, "entity_ai_category": cached.get("entity_ai_category") or "unknown", "business_description": cached.get("business_description") or "", "official_website": cached.get("official_website") or "", } return { "email_domain": domain, "entity_ai_category": existing.get("entity_ai_category") or "unknown", "business_description": existing.get("business_description") or "", "official_website": existing.get("official_website") or "", } def active_failure_enrichment(conn, tenant): existing = get_existing_active_tenant_enrichment(conn, str(tenant.get("id"))) domain_info = cached_or_existing_domain_info(conn, tenant) return { "has_credit_card": existing.get("has_credit_card"), "billing_config_id": existing.get("billing_config_id"), "active_total_credits": existing.get("active_total_credits"), "latest_credit_grant_days_ago": existing.get("latest_credit_grant_days_ago"), "cluster_count": existing.get("cluster_count"), "starter_cluster_count": existing.get("starter_cluster_count") or 0, "essential_cluster_count": existing.get("essential_cluster_count") or 0, "premium_cluster_count": existing.get("premium_cluster_count") or 0, "dedicated_cluster_count": existing.get("dedicated_cluster_count") or 0, "tenant_avg_qps": existing.get("tenant_avg_qps"), "email_domain": domain_info["email_domain"], "entity_ai_category": domain_info["entity_ai_category"], "business_description": domain_info["business_description"], "official_website": domain_info["official_website"], } def upsert_domain_intelligence(conn, domain, payload, source): ensure_connection(conn) with conn.cursor() as cur: cur.execute( """ INSERT INTO domain_intelligence ( email_domain, entity_ai_category, business_description, official_website, source, raw_response, expires_at, created_at, updated_at ) VALUES ( %(email_domain)s, %(entity_ai_category)s, %(business_description)s, %(official_website)s, %(source)s, %(raw_response)s, DATE_ADD(UTC_TIMESTAMP(), INTERVAL %(ttl_days)s DAY), UTC_TIMESTAMP(), UTC_TIMESTAMP() ) ON DUPLICATE KEY UPDATE entity_ai_category = VALUES(entity_ai_category), business_description = VALUES(business_description), official_website = VALUES(official_website), source = VALUES(source), raw_response = VALUES(raw_response), expires_at = VALUES(expires_at), updated_at = UTC_TIMESTAMP() """, { "email_domain": domain, "entity_ai_category": payload["entity_ai_category"], "business_description": payload["business_description"], "official_website": payload["official_website"], "source": source, "raw_response": payload.get("raw_response"), "ttl_days": int(env("DOMAIN_INTELLIGENCE_TTL_DAYS", "30")), }, ) def classify_domain_with_web_search(domain): openai_api_key = env("OPENAI_API_KEY") if not openai_api_key: return { "entity_ai_category": "unknown", "business_description": "OPENAI_API_KEY is not configured; domain was not classified.", "official_website": "", "raw_response": "", } prompt = f""" Use web search to identify the organization that uses the exact email domain "{domain}". Classify the domain into exactly one entity_ai_category: - individual - company_ai_core - company_ai_related - company_non_ai - institution_ai_related - institution_non_ai - unknown Rules: 1. Treat public personal email providers as individual. 2. official_website must be the organization's official website when verified. 3. business_description must be one concise sentence. 4. Return only valid JSON with keys: entity_ai_category, business_description, official_website. """ body = { "model": env("OPENAI_WEB_SEARCH_MODEL", "gpt-5.4-mini"), "tools": [{"type": "web_search_preview"}], "input": prompt, } req = urllib.request.Request( "https://api.openai.com/v1/responses", data=json.dumps(body).encode("utf-8"), headers={ "Authorization": f"Bearer {openai_api_key}", "Content-Type": "application/json", }, method="POST", ) with urllib.request.urlopen(req, timeout=int(env("LLM_TIMEOUT_SECONDS", "60"))) as resp: raw = resp.read().decode("utf-8") try: response = json.loads(raw) except json.JSONDecodeError as exc: return unknown_domain_payload( f"OpenAI response was not valid JSON: {exc}", raw, ) output_text = response.get("output_text") if not output_text: 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 "") output_text = "\n".join(chunks).strip() if not output_text: return unknown_domain_payload("OpenAI response did not include output_text.", raw) try: payload = json.loads(clean_json(output_text)) except json.JSONDecodeError as exc: return unknown_domain_payload( f"OpenAI output_text was not valid JSON: {exc}", raw, ) return { "entity_ai_category": str(payload.get("entity_ai_category") or "unknown").strip(), "business_description": str(payload.get("business_description") or "").strip(), "official_website": str(payload.get("official_website") or "").strip(), "raw_response": raw, } def unknown_domain_payload(reason, raw_response=""): return { "entity_ai_category": "unknown", "business_description": reason, "official_website": "", "raw_response": raw_response, } 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 get_domain_intelligence(conn, domain): if not domain: return { "entity_ai_category": "unknown", "business_description": "Missing or invalid email domain.", "official_website": "", } if domain in FREE_EMAIL_DOMAINS: payload = { "entity_ai_category": "individual", "business_description": "Personal/free email domain; no company business inferred.", "official_website": "", "raw_response": "", } upsert_domain_intelligence(conn, domain, payload, "free_email_domain_list") return payload cached = get_cached_domain_intelligence(conn, domain) if cached: return cached payload = classify_domain_with_web_search(domain) print('use llm\n') upsert_domain_intelligence(conn, domain, payload, "openai_web_search") return payload def upsert_active_tenant(conn, tenant, enrichment, error_message=None): values = {} for column in TENANT_COLUMNS: values[column] = tenant.get(column) values.update(enrichment) values["tenant_id"] = str(tenant.get("id")) values["error_message"] = error_message values.setdefault("source_type", "active_candidate") values.setdefault("total_score", None) values.setdefault("priority", None) for column in ( "starter_cluster_count", "essential_cluster_count", "premium_cluster_count", "dedicated_cluster_count", ): values.setdefault(column, 0) ensure_connection(conn) with conn.cursor() as cur: cur.execute( """ INSERT INTO emea_active_tenants ( tenant_id, created_at, updated_at, name, email, country, time_zone, status, kind, plan, alias, country_region, sales_owner_email, custom_id, segmentation, sales_force_region, sales_force_kind, tam_name, has_credit_card, billing_config_id, active_total_credits, latest_credit_grant_days_ago, cluster_count, starter_cluster_count, essential_cluster_count, premium_cluster_count, dedicated_cluster_count, tenant_avg_qps, email_domain, entity_ai_category, business_description, official_website, source_type, total_score, priority, processed_at, error_message ) VALUES ( %(tenant_id)s, %(created_at)s, %(updated_at)s, %(name)s, %(email)s, %(country)s, %(time_zone)s, %(status)s, %(kind)s, %(plan)s, %(alias)s, %(country_region)s, %(sales_owner_email)s, %(custom_id)s, %(segmentation)s, %(sales_force_region)s, %(sales_force_kind)s, %(tam_name)s, %(has_credit_card)s, %(billing_config_id)s, %(active_total_credits)s, %(latest_credit_grant_days_ago)s, %(cluster_count)s, %(starter_cluster_count)s, %(essential_cluster_count)s, %(premium_cluster_count)s, %(dedicated_cluster_count)s, %(tenant_avg_qps)s, %(email_domain)s, %(entity_ai_category)s, %(business_description)s, %(official_website)s, %(source_type)s, %(total_score)s, %(priority)s, UTC_TIMESTAMP(), %(error_message)s ) ON DUPLICATE KEY UPDATE created_at = VALUES(created_at), updated_at = VALUES(updated_at), name = VALUES(name), email = VALUES(email), country = VALUES(country), time_zone = VALUES(time_zone), status = VALUES(status), kind = VALUES(kind), plan = VALUES(plan), alias = VALUES(alias), country_region = VALUES(country_region), sales_owner_email = VALUES(sales_owner_email), custom_id = VALUES(custom_id), segmentation = VALUES(segmentation), sales_force_region = VALUES(sales_force_region), sales_force_kind = VALUES(sales_force_kind), tam_name = VALUES(tam_name), has_credit_card = VALUES(has_credit_card), billing_config_id = VALUES(billing_config_id), active_total_credits = VALUES(active_total_credits), latest_credit_grant_days_ago = VALUES(latest_credit_grant_days_ago), cluster_count = VALUES(cluster_count), starter_cluster_count = VALUES(starter_cluster_count), essential_cluster_count = VALUES(essential_cluster_count), premium_cluster_count = VALUES(premium_cluster_count), dedicated_cluster_count = VALUES(dedicated_cluster_count), tenant_avg_qps = VALUES(tenant_avg_qps), email_domain = VALUES(email_domain), entity_ai_category = VALUES(entity_ai_category), business_description = VALUES(business_description), official_website = VALUES(official_website), source_type = VALUES(source_type), total_score = COALESCE(VALUES(total_score), total_score), priority = COALESCE(VALUES(priority), priority), processed_at = UTC_TIMESTAMP(), error_message = VALUES(error_message) """, values, ) def upsert_new_signup_into_active_tenants(conn, tenant, enrichment, error_message=None): values = {} for column in TENANT_COLUMNS: values[column] = tenant.get(column) values.update(enrichment) values["tenant_id"] = str(tenant.get("id")) values["error_message"] = error_message values["source_type"] = "new_signup" for column in ( "has_credit_card", "billing_config_id", "active_total_credits", "latest_credit_grant_days_ago", "cluster_count", "starter_cluster_count", "essential_cluster_count", "premium_cluster_count", "dedicated_cluster_count", "tenant_avg_qps", "total_score", "priority", ): if column.endswith("_cluster_count"): values.setdefault(column, 0) else: values.setdefault(column, None) ensure_connection(conn) with conn.cursor() as cur: cur.execute( """ INSERT INTO emea_active_tenants ( tenant_id, created_at, updated_at, name, email, country, time_zone, status, kind, plan, alias, country_region, sales_owner_email, custom_id, segmentation, sales_force_region, sales_force_kind, tam_name, has_credit_card, billing_config_id, active_total_credits, latest_credit_grant_days_ago, cluster_count, starter_cluster_count, essential_cluster_count, premium_cluster_count, dedicated_cluster_count, tenant_avg_qps, email_domain, entity_ai_category, business_description, official_website, source_type, total_score, priority, processed_at, error_message ) VALUES ( %(tenant_id)s, %(created_at)s, %(updated_at)s, %(name)s, %(email)s, %(country)s, %(time_zone)s, %(status)s, %(kind)s, %(plan)s, %(alias)s, %(country_region)s, %(sales_owner_email)s, %(custom_id)s, %(segmentation)s, %(sales_force_region)s, %(sales_force_kind)s, %(tam_name)s, %(has_credit_card)s, %(billing_config_id)s, %(active_total_credits)s, %(latest_credit_grant_days_ago)s, %(cluster_count)s, %(starter_cluster_count)s, %(essential_cluster_count)s, %(premium_cluster_count)s, %(dedicated_cluster_count)s, %(tenant_avg_qps)s, %(email_domain)s, %(entity_ai_category)s, %(business_description)s, %(official_website)s, %(source_type)s, %(total_score)s, %(priority)s, UTC_TIMESTAMP(), %(error_message)s ) ON DUPLICATE KEY UPDATE created_at = VALUES(created_at), updated_at = VALUES(updated_at), name = VALUES(name), email = VALUES(email), country = VALUES(country), time_zone = VALUES(time_zone), status = VALUES(status), kind = VALUES(kind), plan = VALUES(plan), alias = VALUES(alias), country_region = VALUES(country_region), sales_owner_email = VALUES(sales_owner_email), custom_id = VALUES(custom_id), segmentation = VALUES(segmentation), sales_force_region = VALUES(sales_force_region), sales_force_kind = VALUES(sales_force_kind), tam_name = VALUES(tam_name), email_domain = VALUES(email_domain), entity_ai_category = VALUES(entity_ai_category), business_description = VALUES(business_description), official_website = VALUES(official_website), total_score = VALUES(total_score), priority = VALUES(priority), processed_at = UTC_TIMESTAMP(), error_message = VALUES(error_message) """, values, ) def upsert_new_signup(conn, tenant, enrichment, error_message=None): values = {} for column in TENANT_COLUMNS: values[column] = tenant.get(column) values.update(enrichment) values["tenant_id"] = str(tenant.get("id")) values["error_message"] = error_message ensure_connection(conn) with conn.cursor() as cur: cur.execute( """ INSERT INTO emea_new_signup ( tenant_id, created_at, updated_at, name, email, country, time_zone, status, kind, plan, alias, country_region, sales_owner_email, custom_id, segmentation, sales_force_region, sales_force_kind, tam_name, email_domain, entity_ai_category, business_description, official_website, processed_at, error_message ) VALUES ( %(tenant_id)s, %(created_at)s, %(updated_at)s, %(name)s, %(email)s, %(country)s, %(time_zone)s, %(status)s, %(kind)s, %(plan)s, %(alias)s, %(country_region)s, %(sales_owner_email)s, %(custom_id)s, %(segmentation)s, %(sales_force_region)s, %(sales_force_kind)s, %(tam_name)s, %(email_domain)s, %(entity_ai_category)s, %(business_description)s, %(official_website)s, UTC_TIMESTAMP(), %(error_message)s ) ON DUPLICATE KEY UPDATE created_at = VALUES(created_at), updated_at = VALUES(updated_at), name = VALUES(name), email = VALUES(email), country = VALUES(country), time_zone = VALUES(time_zone), status = VALUES(status), kind = VALUES(kind), plan = VALUES(plan), alias = VALUES(alias), country_region = VALUES(country_region), sales_owner_email = VALUES(sales_owner_email), custom_id = VALUES(custom_id), segmentation = VALUES(segmentation), sales_force_region = VALUES(sales_force_region), sales_force_kind = VALUES(sales_force_kind), tam_name = VALUES(tam_name), email_domain = VALUES(email_domain), entity_ai_category = VALUES(entity_ai_category), business_description = VALUES(business_description), official_website = VALUES(official_website), processed_at = UTC_TIMESTAMP(), error_message = VALUES(error_message) """, values, ) def process_tenant(target_conn, tenant, tokens, now): tenant_id = str(tenant.get("id")) domain = email_domain(tenant.get("email")) try: domain_info = get_domain_intelligence(target_conn, domain) except Exception as exc: raise RuntimeError(f"domain intelligence failed for domain={domain or ''}: {exc}") from exc try: cluster_stats = get_tenant_cluster_stats( tenant_id, tokens["devops_token"], tokens["clinic_token"], end=int(now.timestamp()), ) print('get cluster_stats\n', cluster_stats["cluster_count"]) except Exception as exc: raise RuntimeError(f"cluster stats failed: {exc}") from exc existing = get_existing_active_tenant_enrichment(target_conn, tenant_id) try: billing_config_id = normalized_billing_config_id(existing.get("billing_config_id")) billing = find_valid_credit_card( tenant_id, tokens["devops_token"], operator_email=tokens.get("operator_email"), now=now, billing_config_id=billing_config_id, ) print('get billing\n',tenant, billing) credit_summary = fetch_credit_summary( tenant_id, billing.get("billing_config_id"), tokens["devops_token"], operator_email=tokens.get("operator_email"), now=now, ) print('get credits\n', credit_summary) billing_error = None except Exception as exc: billing = { "has_credit_card": existing.get("has_credit_card"), "billing_config_id": existing.get("billing_config_id"), } credit_summary = { "active_total_credits": existing.get("active_total_credits") or Decimal("0"), "latest_credit_grant_days_ago": existing.get("latest_credit_grant_days_ago"), } billing_error = f"billing failed: {exc}" enrichment = { "has_credit_card": billing["has_credit_card"], "billing_config_id": billing["billing_config_id"], "active_total_credits": str(credit_summary["active_total_credits"]), "latest_credit_grant_days_ago": credit_summary["latest_credit_grant_days_ago"], "cluster_count": cluster_stats["cluster_count"], "starter_cluster_count": cluster_stats["starter_cluster_count"], "essential_cluster_count": cluster_stats["essential_cluster_count"], "premium_cluster_count": cluster_stats["premium_cluster_count"], "dedicated_cluster_count": cluster_stats["dedicated_cluster_count"], "tenant_avg_qps": cluster_stats["tenant_avg_qps"], "email_domain": domain, "entity_ai_category": domain_info["entity_ai_category"], "business_description": domain_info["business_description"], "official_website": domain_info["official_website"], } errors = cluster_stats.get("errors") or [] if billing_error: errors.append(billing_error) error_message = "; ".join(errors) if errors else None upsert_active_tenant(target_conn, tenant, enrichment, error_message) return "processed_with_errors" if error_message else "processed" def process_new_signup_tenant(target_conn, tenant, now): domain = email_domain(tenant.get("email")) try: domain_info = get_domain_intelligence(target_conn, domain) except Exception as exc: raise RuntimeError(f"domain intelligence failed for domain={domain or ''}: {exc}") from exc enrichment = { "email_domain": domain, "entity_ai_category": domain_info["entity_ai_category"], "business_description": domain_info["business_description"], "official_website": domain_info["official_website"], } upsert_new_signup_into_active_tenants(target_conn, tenant, enrichment) upsert_new_signup(target_conn, tenant, enrichment) def lambda_handler(event, context): now = datetime.now(timezone.utc) tokens = { "devops_token": env("DEVOPS_API_TOKEN", required=True), "clinic_token": env("CLINIC_API_KEY", required=True), "operator_email": env("OPERATOR_EMAIL"), } started = time.time() active_processed = 0 active_failed = 0 new_signup_processed = 0 new_signup_failed = 0 devops_conn = mysql_connect("DEVOPS") tenants = [] new_signup_tenants = [] try: tenants = list_tenants(devops_conn) new_signup_tenants = list_new_signup_tenants(devops_conn) finally: devops_conn.close() target_conn = mysql_connect("TARGET") try: i = 1 for tenant in tenants: print(i) i = i + 1 tenant_id = str(tenant.get("id")) try: process_tenant(target_conn, tenant, tokens, now) target_conn.commit() enqueue_scoring_request(tenant_id, "active_candidate", "active_processed", now=now) active_processed += 1 except Exception as exc: target_conn.rollback() active_failed += 1 try: upsert_active_tenant( target_conn, tenant, { "has_credit_card": None, "billing_config_id": None, "active_total_credits": None, "latest_credit_grant_days_ago": None, "cluster_count": None, "tenant_avg_qps": None, "email_domain": email_domain(tenant.get("email")), "entity_ai_category": "unknown", "business_description": "", "official_website": "", }, error_message=str(exc), ) target_conn.commit() enqueue_scoring_request( tenant_id, "active_candidate", "active_failed", now=now, ) except Exception: target_conn.rollback() print(f"active tenant_id={tenant_id} failed: {exc}") if context is not None and context.get_remaining_time_in_millis() < 30000: print("stopping early because Lambda remaining time is below 30 seconds") return { "active_tenant_count": len(tenants), "active_processed": active_processed, "active_failed": active_failed, "new_signup_tenant_count": 0, "new_signup_processed": new_signup_processed, "new_signup_failed": new_signup_failed, "elapsed_seconds": round(time.time() - started, 3), } for tenant in new_signup_tenants: tenant_id = str(tenant.get("id")) try: process_new_signup_tenant(target_conn, tenant, now) target_conn.commit() enqueue_scoring_request(tenant_id, "new_signup", "new_signup_processed", now=now) new_signup_processed += 1 except Exception as exc: target_conn.rollback() new_signup_failed += 1 error_enrichment = { "email_domain": email_domain(tenant.get("email")), "entity_ai_category": "unknown", "business_description": "", "official_website": "", "error_message": str(exc), } try: upsert_new_signup( target_conn, tenant, error_enrichment, error_message=str(exc), ) upsert_new_signup_into_active_tenants( target_conn, tenant, error_enrichment, error_message=str(exc), ) target_conn.commit() enqueue_scoring_request( tenant_id, "new_signup", "new_signup_failed", now=now, ) except Exception: target_conn.rollback() print(f"new signup tenant_id={tenant_id} failed: {exc}") if context is not None and context.get_remaining_time_in_millis() < 30000: print("stopping early because Lambda remaining time is below 30 seconds") break return { "active_tenant_count": len(tenants), "active_processed": active_processed, "active_failed": active_failed, "new_signup_tenant_count": len(new_signup_tenants), "new_signup_processed": new_signup_processed, "new_signup_failed": new_signup_failed, "elapsed_seconds": round(time.time() - started, 3), } finally: target_conn.close()