构建端到端数据管道

import requests
import json
import uuid
import random
import time
import os
from datetime import datetime, timezone, timedelta

# 配置 - 从之前的步骤中导出
STREAM_ENDPOINT = os.environ["STREAM_ENDPOINT"]# 来自您创建的流
API_TOKEN = os.environ["WRANGLER_R2_SQL_AUTH_TOKEN"] #前面创建的同一个
EVENTS_TO_SEND = 1000 # 随时调整它

def generate_transaction():
    """生成一些偶尔有欺诈的随机交易"""

    # User IDs
    high_risk_users = [1001, 1002, 1003, 1004, 1005]
    normal_users = list(range(1006, 2000))

    user_id = random.choice(high_risk_users + normal_users)
    is_high_risk_user = user_id in high_risk_users

    # 生成金额
    if random.random() < 0.05:
        amount = round(random.uniform(5000, 50000), 2)
    elif random.random() < 0.03:
        amount = round(random.uniform(0.01, 1.00), 2)
    else:
        amount = round(random.uniform(10, 500), 2)

    # 地点
    normal_locations = ["NEW_YORK", "LOS_ANGELES", "CHICAGO", "MIAMI", "SEATTLE", "SAN FRANCISCO"]
    high_risk_locations = ["UNKNOWN_LOCATION", "VPN_EXIT", "MARS", "BAT_CAVE"]

    if is_high_risk_user and random.random() < 0.3:
        location = random.choice(high_risk_locations)
    else:
        location = random.choice(normal_locations)

    # 商户类别
    normal_merchants = ["GROCERY", "GAS_STATION", "RESTAURANT", "RETAIL"]
    high_risk_merchants = ["GAMBLING", "CRYPTO", "MONEY_TRANSFER", "GIFT_CARDS"]

    if random.random() < 0.1:  # 10% 高风险商户
        merchant_category = random.choice(high_risk_merchants)
    else:
        merchant_category = random.choice(normal_merchants)

    # 一系列检查，用于按一定幅度增加欺诈评分
    fraud_score = 0
    if amount > 2000: fraud_score += 0.4
    if amount < 1: fraud_score += 0.3
    if location in high_risk_locations: fraud_score += 0.5
    if merchant_category in high_risk_merchants: fraud_score += 0.3
    if is_high_risk_user: fraud_score += 0.2

    # 比较欺诈评分
    is_fraud = random.random() < min(fraud_score * 0.3, 0.8)

    # 生成时间戳（有些欺诈发生在不寻常的时间）
    base_time = datetime.now(timezone.utc)
    if is_fraud and random.random() < 0.4:  # 40% 的欺诈在晚上
        hour = random.randint(0, 5)  # 深夜/清晨
        transaction_time = base_time.replace(hour=hour)
    else:
        transaction_time = base_time - timedelta(
            hours=random.randint(0, 168)  # 上周
        )

    return {
        "transaction_id": str(uuid.uuid4()),
        "user_id": user_id,
        "amount": amount,
        "transaction_timestamp": transaction_time.isoformat(),
        "location": location,
        "merchant_category": merchant_category,
        "is_fraud": True if is_fraud else False
    }

def send_batch_to_stream(events, batch_size=100):
    """分批将事件发送到 Cloudflare Stream"""

    headers = {
        "Authorization": f"Bearer {API_TOKEN}",
        "Content-Type": "application/json"
    }

    total_sent = 0
    fraud_count = 0

    for i in range(0, len(events), batch_size):
        batch = events[i:i + batch_size]
        fraud_in_batch = sum(1 for event in batch if event["is_fraud"] == True)

        try:
            response = requests.post(STREAM_ENDPOINT, headers=headers, json=batch)

            if response.status_code in [200, 201]:
                total_sent += len(batch)
                fraud_count += fraud_in_batch
                print(f"Sent batch of {len(batch)} events (Total: {total_sent})")
            else:
                print(f"Failed to send batch: {response.status_code} - {response.text}")

        except Exception as e:
            print(f"Error sending batch: {e}")

        time.sleep(0.1)

    return total_sent, fraud_count

def main():
    print("Generating fraud detection data...")

    # 生成事件
    events = []
    for i in range(EVENTS_TO_SEND):
        events.append(generate_transaction())
        if (i + 1) % 100 == 0:
            print(f"Generated {i + 1} events...")

    fraud_events = sum(1 for event in events if event["is_fraud"] == True)
    print(f"📊 Generated {len(events)} total events ({fraud_events} fraud, {fraud_events/len(events)*100:.1f}%)")

    # 发送到流
    print("Sending data to Pipeline stream...")
    sent, fraud_sent = send_batch_to_stream(events)

    print(f"\nComplete!")
    print(f"   Events sent: {sent:,}")
    print(f"   Fraud events: {fraud_sent:,} ({fraud_sent/sent*100:.1f}%)")
    print(f"   Data is now flowing through your pipeline!")

if __name__ == "__main__":
    main()