FORMAT_CTR = {
    'banner': 0.05,          # 5% - 배너 광고는 낮음
    'interstitial': 0.12,    # 12% - 전면 광고는 중간
    'rewarded_video': 0.25   # 25% - 리워드 광고는 높음
}

COUNTRY_ECPM = {
    'US': (0.02, 0.10),   # 미국: $0.02~$0.10
    'KR': (0.01, 0.05),   # 한국: $0.01~$0.05
    'JP': (0.015, 0.08),  # 일본: $0.015~$0.08
}

from confluent_kafka import Producer
import json
import random
from datetime import datetime
from faker import Faker

fake = Faker()

# Kafka Producer 설정
conf = {'bootstrap.servers': 'localhost:9092'}
producer = Producer(conf)

def generate_ad_event():
    """광고 이벤트 하나 생성"""
    
    # 광고 포맷 선택
    ad_format = random.choice(['banner', 'interstitial', 'rewarded_video'])
    
    # 해당 포맷의 CTR 기반으로 클릭 여부 결정
    ctr = FORMAT_CTR[ad_format]
    is_click = random.random() < ctr
    
    # 클릭이면 수익 발생
    revenue = 0
    if is_click:
        country = random.choice(['US', 'KR', 'JP', ...])
        ecpm_range = COUNTRY_ECPM[country]
        revenue = round(random.uniform(*ecpm_range), 4)
    
    return {
        'event_id': f"E{int(time.time()*1000)}{random.randint(100,999)}",
        'timestamp': datetime.now().isoformat(),
        'user_id': f"U{fake.random_int(min=10000, max=99999)}",
        'campaign_id': random.choice(['C001', 'C002', ...]),
        'ad_format': ad_format,
        'country': country,
        'device': random.choice(['iOS', 'Android']),
        'event_type': 'click' if is_click else 'impression',
        'revenue': revenue
    }

is_click = random.random() < ctr

producer.produce(
    'ad-events',
    key=event['event_id'].encode('utf-8'),
    value=json.dumps(event).encode('utf-8')
)
producer.poll(0)  # 비동기 처리

if event_count % 1000 == 0:
    producer.flush()  # 버퍼 강제 비우기

from confluent_kafka import Producer
import json
import random
import time
from datetime import datetime
from faker import Faker

fake = Faker()

conf = {'bootstrap.servers': 'localhost:9092'}
producer = Producer(conf)

CAMPAIGNS = ['C001', 'C002', 'C003', 'C004', 'C005']
AD_FORMATS = ['banner', 'interstitial', 'rewarded_video']
COUNTRIES = ['US', 'KR', 'JP', 'UK', 'DE', 'FR', 'CA']
DEVICES = ['iOS', 'Android']

COUNTRY_ECPM = {
    'US': (0.02, 0.10),
    'KR': (0.01, 0.05),
    'JP': (0.015, 0.08),
    'UK': (0.015, 0.07),
    'DE': (0.01, 0.06),
    'FR': (0.01, 0.05),
    'CA': (0.015, 0.07)
}

FORMAT_CTR = {
    'banner': 0.05,
    'interstitial': 0.12,
    'rewarded_video': 0.25
}

def delivery_report(err, msg):
    if err is not None:
        print(f'❌ Message delivery failed: {err}')

def generate_ad_event():
    timestamp = datetime.now().isoformat()
    user_id = f"U{fake.random_int(min=10000, max=99999)}"
    campaign_id = random.choice(CAMPAIGNS)
    ad_format = random.choice(AD_FORMATS)
    country = random.choice(COUNTRIES)
    device = random.choice(DEVICES)
    
    ctr = FORMAT_CTR[ad_format]
    is_click = random.random() < ctr
    event_type = 'click' if is_click else 'impression'
    
    revenue = 0
    if is_click:
        ecpm_range = COUNTRY_ECPM[country]
        revenue = round(random.uniform(*ecpm_range), 4)
    
    return {
        'event_id': f"E{int(time.time()*1000)}{random.randint(100,999)}",
        'timestamp': timestamp,
        'user_id': user_id,
        'campaign_id': campaign_id,
        'ad_format': ad_format,
        'country': country,
        'device': device,
        'event_type': event_type,
        'revenue': revenue
    }

def main():
    print("🚀 Starting Ad Log Generator...")
    print("📊 Generating ~100 events/sec")
    print("📍 Kafka Topic: ad-events")
    print("=" * 50)
    
    event_count = 0
    start_time = time.time()
    
    try:
        while True:
            event = generate_ad_event()
            
            producer.produce(
                'ad-events',
                key=event['event_id'].encode('utf-8'),
                value=json.dumps(event).encode('utf-8'),
                callback=delivery_report
            )
            producer.poll(0)
            
            event_count += 1
            
            if event_count % 1000 == 0:
                producer.flush()
                elapsed = time.time() - start_time
                rate = event_count / elapsed
                print(f"✅ Sent {event_count:,} events | Rate: {rate:.1f} events/sec")
            
            time.sleep(0.01)
            
    except KeyboardInterrupt:
        print("\n\n🛑 Stopping generator...")
        producer.flush()
        elapsed = time.time() - start_time
        print(f"📊 Final Stats:")
        print(f"   - Total Events: {event_count:,}")
        print(f"   - Duration: {elapsed:.1f}s")
        print(f"   - Avg Rate: {event_count/elapsed:.1f} events/sec")

if __name__ == "__main__":
    main()