from ..snowflake.base import SnowflakeJob
from src.utils.snowflake.constants import QueryType, Role, Warehouse
from dagster import OpExecutionContext
from typing import Optional
import pandas as pd
import os
import stripe

# Email domain block list
STRIPE_RADAR_LIST_ID = "rsl_1KtA5uE1icI7lnWdJDzrdXY7"


class StripeBlockedEmailDomainSync(SnowflakeJob):
    def __init__(self):
        super().__init__(
            name="stripe_blocked_email_domain_sync",
            description="Sync domain blocks to Stripe Radar lists",
            query_file="../bots/queries/blocked_email_domain_sync.sql",
            query_type=QueryType.SELECT,
            warehouse=Warehouse.SMALL,
            role=Role.ACCOUNTADMIN,
            group_name="finops",
            # monitored=True,
            owners=["team:core-pod"],
            metadata={
                "slack": "#tech-anti-bots",
            },
            tags={"team": "core-pod", "category": "finops", "tech-alerts": "true"},
            deps=["bot_status_changes"],
        )

    def post_execute(self, result: Optional[pd.DataFrame], context: OpExecutionContext) -> None:
        """Sync bot domain blocks to Stripe Radar lists"""
        if result is None or len(result) == 0:
            context.log.info("No bot domains to sync")
            return

        # Convert result to DataFrame if it's not already
        if not isinstance(result, pd.DataFrame):
            result = pd.DataFrame(result)

        # Get Stripe secret key from environment
        stripe_secret_key = os.getenv("STRIPE_API_KEY")
        if not stripe_secret_key:
            raise ValueError("STRIPE_API_KEY environment variable not set")

        # Store context and secret key for async functions
        self.context = context
        self.stripe_secret_key = stripe_secret_key
        stripe.api_key = stripe_secret_key

        context.log.info(f"Preview of bot domains to sync:\n{result.head(10).to_markdown()}")

        # Summary metrics
        total_processed = 0
        success_count = 0
        failure_count = 0

        # Process batches
        for i in range(0, len(result)):
            domain = result.iloc[i]["ENTITY_ID"]
            # Process batch concurrently
            try:
                context.log.info(f"Adding {domain} to Stripe Radar list: {STRIPE_RADAR_LIST_ID}")
                stripe.radar.ValueListItem.create(
                    value_list=STRIPE_RADAR_LIST_ID,
                    value=domain,
                )
                success_count += 1
            except stripe.InvalidRequestError as e:
                if "already exists" in str(e):
                    context.log.info(f"Domain {domain} already exists in Stripe Radar list")
                else:
                    failure_count += 1

            total_processed += 1

        # Log summary
        context.log.info(
            f"Sync complete. "
            f"Total processed: {total_processed}, "
            f"Success: {success_count}, "
            f"Failures: {failure_count}"
        )


job = StripeBlockedEmailDomainSync()