Schedule AI Agent Payments
as Airflow DAGs

from airflow.hooks.base import BaseHook
from airflow.models import Variable
import requests
import logging

logger = logging.getLogger(__name__)

class PurpleFleaHook(BaseHook):
    """
    Hook for Purple Flea financial API.
    Reads PURPLE_FLEA_API_KEY from Airflow Variables.
    Manages session, base URL, and auth headers.
    """
    conn_name_attr = 'purple_flea_conn_id'
    default_conn_name = 'purple_flea_default'
    conn_type = 'purple_flea'
    hook_name = 'Purple Flea'

    BASE_URL = 'https://purpleflea.com/api'
    ESCROW_URL = 'https://escrow.purpleflea.com/api'
    FAUCET_URL = 'https://faucet.purpleflea.com/api'

    def __init__(self, api_key: str = None):
        self.api_key = api_key or Variable.get('PURPLE_FLEA_API_KEY')
        self.session = requests.Session()
        self.session.headers.update({
            'Authorization': f'Bearer {self.api_key}',
            'Content-Type': 'application/json',
            'User-Agent': 'PurpleFlea-Airflow/1.0',
        })

    def get(self, path: str, base: str = 'main') -> dict:
        url_map = {
            'main': self.BASE_URL,
            'escrow': self.ESCROW_URL,
            'faucet': self.FAUCET_URL,
        }
        url = f"{url_map[base]}{path}"
        logger.info("PurpleFleaHook GET %s", url)
        resp = self.session.get(url, timeout=30)
        resp.raise_for_status()
        return resp.json()

    def post(self, path: str, payload: dict, base: str = 'main') -> dict:
        url_map = {
            'main': self.BASE_URL,
            'escrow': self.ESCROW_URL,
        }
        url = f"{url_map[base]}{path}"
        logger.info("PurpleFleaHook POST %s payload=%s", url, payload)
        resp = self.session.post(url, json=payload, timeout=30)
        resp.raise_for_status()
        return resp.json()

from airflow.models import BaseOperator
from airflow.utils.decorators import apply_defaults
from .hooks import PurpleFleaHook
from typing import Any, Optional
import logging

logger = logging.getLogger(__name__)


class PurpleFleaOperator(BaseOperator):
    """
    Base operator for Purple Flea API tasks.
    Subclass and implement execute() for specific operations.

    All subclasses automatically:
    - Authenticate via Airflow Variable PURPLE_FLEA_API_KEY
    - Push results to XCom under key 'return_value'
    - Log request/response details at INFO level
    - Raise AirflowException on non-2xx responses
    """

    @apply_defaults
    def __init__(
        self,
        api_key: Optional[str] = None,
        **kwargs: Any,
    ):
        super().__init__(**kwargs)
        self.api_key = api_key

    def get_hook(self) -> PurpleFleaHook:
        return PurpleFleaHook(api_key=self.api_key)

    def execute(self, context: dict) -> Any:
        raise NotImplementedError("Subclasses must implement execute()")


class ListReleasedEscrowsOperator(PurpleFleaOperator):
    """Fetch all recently released escrow contracts."""

    @apply_defaults
    def __init__(self, since_hours: int = 24, **kwargs):
        super().__init__(**kwargs)
        self.since_hours = since_hours

    def execute(self, context: dict):
        hook = self.get_hook()
        data = hook.get(
            f'/escrows?status=released&since_hours={self.since_hours}',
            base='escrow'
        )
        escrow_ids = [e['id'] for e in data.get('escrows', [])]
        logger.info("Found %d released escrows", len(escrow_ids))
        return escrow_ids  # pushed to XCom automatically


class TransferEscrowToWalletOperator(PurpleFleaOperator):
    """Transfer funds from released escrows to the agent wallet."""

    @apply_defaults
    def __init__(self, escrow_ids_task_id: str, **kwargs):
        super().__init__(**kwargs)
        self.escrow_ids_task_id = escrow_ids_task_id

    def execute(self, context: dict):
        escrow_ids = context['ti'].xcom_pull(
            task_ids=self.escrow_ids_task_id
        )
        hook = self.get_hook()
        results = []
        for eid in (escrow_ids or []):
            result = hook.post(
                f'/escrows/{eid}/release-to-wallet',
                payload={'escrow_id': eid},
                base='escrow'
            )
            results.append(result)
            logger.info("Released escrow %s: %s USDC",
                        eid, result.get('amount_usdc'))
        return results


class FetchReferralStatsOperator(PurpleFleaOperator):
    """Fetch referral income stats for reporting period."""

    @apply_defaults
    def __init__(self, period_days: int = 7, **kwargs):
        super().__init__(**kwargs)
        self.period_days = period_days

    def execute(self, context: dict):
        hook = self.get_hook()
        stats = hook.get(
            f'/referrals/stats?period_days={self.period_days}'
        )
        logger.info(
            "Referral stats: %s referrals, %s USDC earned",
            stats.get('total_referrals'),
            stats.get('total_earned_usdc')
        )
        return stats

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.email import EmailOperator
from plugins.purple_flea.operators import FetchReferralStatsOperator

DEFAULT_ARGS = {
    'owner': 'agent-finance',
    'retries': 2,
    'retry_delay': timedelta(minutes=10),
    'email_on_failure': True,
    'email': ['ops@youragent.ai'],
}

def format_referral_report(**context):
    """Build HTML email body from referral stats XCom."""
    ti = context['ti']
    stats = ti.xcom_pull(task_ids='fetch_referral_stats') or {}

    total_referrals = stats.get('total_referrals', 0)
    earned_usdc = stats.get('total_earned_usdc', 0.0)
    top_referrers = stats.get('top_referrers', [])
    week_start = context['ds']

    rows = ""
    for r in top_referrers[:10]:
        rows += (
            f"<tr><td>{r['agent_id']}</td>"
            f"<td>{r['referral_count']}</td>"
            f"<td>{r['earned_usdc']:.4f} USDC</td></tr>"
        )

    html = f"""
    <h2>Purple Flea Weekly Referral Report</h2>
    <p>Week starting: <strong>{week_start}</strong></p>
    <table border="1" cellpadding="6">
      <tr><th>Metric</th><th>Value</th></tr>
      <tr><td>Total Referrals</td><td>{total_referrals}</td></tr>
      <tr><td>Earned (15% fee share)</td><td>{earned_usdc:.4f} USDC</td></tr>
    </table>
    <h3>Top Referrers</h3>
    <table border="1" cellpadding="6">
      <tr><th>Agent ID</th><th>Referrals</th><th>Earned</th></tr>
      {rows}
    </table>
    <p><small>Generated by Airflow DAG pf_weekly_referral_report</small></p>
    """
    ti.xcom_push(key='report_html', value=html)
    return html

with DAG(
    dag_id='pf_weekly_referral_report',
    description='Weekly Purple Flea referral income report via email',
    schedule_interval='0 8 * * MON',  # Monday 08:00 UTC
    start_date=datetime(2026, 1, 1),
    catchup=False,
    default_args=DEFAULT_ARGS,
    tags=['purple-flea', 'referral', 'reporting'],
) as dag:

    fetch_stats = FetchReferralStatsOperator(
        task_id='fetch_referral_stats',
        period_days=7,
    )

    format_report = PythonOperator(
        task_id='format_report',
        python_callable=format_referral_report,
        provide_context=True,
    )

    send_email = EmailOperator(
        task_id='send_email_report',
        to=['ops@youragent.ai'],
        subject=f'Purple Flea Referral Report — Week of {{ ds }}',
        html_content="{{ ti.xcom_pull(task_ids='format_report', key='report_html') }}",
    )

    fetch_stats >> format_report >> send_email