# Kafka Producer (Python - confluent-kafka)
from confluent_kafka import Producer
import json
import time
 
conf = {
    'bootstrap.servers': 'broker1:9092,broker2:9092,broker3:9092',
    'acks': 'all',                    # 全ISRレプリカの書き込み確認
    'retries': 5,
    'retry.backoff.ms': 100,
    'enable.idempotence': True,       # べき等プロデューサー（重複防止）
    'compression.type': 'snappy',     # 圧縮でスループット向上
    'linger.ms': 5,                   # バッチ化のための待機時間
    'batch.size': 32768,              # バッチサイズ (32KB)
    'max.in.flight.requests.per.connection': 5,  # べき等有効時は最大5
}
 
producer = Producer(conf)
 
def delivery_report(err, msg):
    if err:
        print(f"配信失敗: {err}")
    else:
        print(f"配信成功: topic={msg.topic()} "
              f"partition={msg.partition()} offset={msg.offset()}")
 
# メッセージ送信
for i in range(1000):
    order = {"order_id": i, "user_id": i % 100, "amount": 1000 + i,
             "timestamp": time.time()}
    producer.produce(
        topic='order-events',
        key=str(order['user_id']).encode('utf-8'),   # 同一ユーザーは同一パーティション
        value=json.dumps(order).encode('utf-8'),
        callback=delivery_report,
    )
    producer.poll(0)  # コールバック処理
 
producer.flush()  # 全メッセージの送信完了を待機

# Kafka Consumer (Python - confluent-kafka)
from confluent_kafka import Consumer, KafkaError
import json
 
conf = {
    'bootstrap.servers': 'broker1:9092',
    'group.id': 'order-processing-group',
    'auto.offset.reset': 'earliest',       # 初回は最古から読む
    'enable.auto.commit': False,           # 手動コミットで確実な処理
    'max.poll.interval.ms': 300000,        # 処理タイムアウト 5分
    'session.timeout.ms': 45000,
    'fetch.min.bytes': 1024,               # 最低1KBでフェッチ（バッチ効率化）
    'fetch.max.wait.ms': 500,              # 最大500ms待機
}
 
consumer = Consumer(conf)
consumer.subscribe(['order-events'])
 
try:
    while True:
        msg = consumer.poll(timeout=1.0)
        if msg is None:
            continue
        if msg.error():
            if msg.error().code() == KafkaError._PARTITION_EOF:
                continue
            raise Exception(f"Consumer error: {msg.error()}")
 
        order = json.loads(msg.value().decode('utf-8'))
        print(f"受信: partition={msg.partition()} "
              f"offset={msg.offset()} order_id={order['order_id']}")
 
        # ビジネスロジック実行
        process_order(order)
 
        # 処理完了後に手動コミット
        consumer.commit(asynchronous=False)
finally:
    consumer.close()

# Kafka を使ったリアルタイム集計の概念的実装
import json
import time
from collections import defaultdict
from dataclasses import dataclass, field
from typing import Callable, Optional
 
@dataclass
class WindowedCounter:
    """タンブリングウィンドウによるリアルタイム集計"""
    window_size_sec: int = 60
    windows: dict = field(default_factory=lambda: defaultdict(lambda: defaultdict(int)))
 
    def add(self, key: str, value: int = 1):
        window_start = int(time.time() / self.window_size_sec) * self.window_size_sec
        self.windows[window_start][key] += value
 
    def get_current_window(self) -> dict:
        window_start = int(time.time() / self.window_size_sec) * self.window_size_sec
        return dict(self.windows[window_start])
 
    def get_window(self, window_start: int) -> dict:
        return dict(self.windows.get(window_start, {}))
 
    def cleanup_old_windows(self, retain_count: int = 10):
        """古いウィンドウを削除してメモリ節約"""
        sorted_windows = sorted(self.windows.keys())
        for w in sorted_windows[:-retain_count]:
            del self.windows[w]
 
 
class StreamProcessor:
    """Kafka Consumer ベースのストリーム処理フレームワーク
 
    機能:
    1. リアルタイムイベント集計（ウィンドウ関数）
    2. イベントフィルタリング / 変換
    3. 出力トピックへの書き込み
    """
 
    def __init__(self, consumer, producer,
                 input_topic: str, output_topic: str):
        self.consumer = consumer
        self.producer = producer
        self.input_topic = input_topic
        self.output_topic = output_topic
        self.counter = WindowedCounter(window_size_sec=60)
        self.filters: list[Callable] = []
        self.transformers: list[Callable] = []
 
    def add_filter(self, predicate: Callable):
        """イベントフィルタを追加"""
        self.filters.append(predicate)
 
    def add_transformer(self, transform: Callable):
        """イベント変換を追加"""
        self.transformers.append(transform)
 
    def process(self):
        """メインの処理ループ"""
        self.consumer.subscribe([self.input_topic])
 
        while True:
            msg = self.consumer.poll(timeout=1.0)
            if msg is None:
                continue
            if msg.error():
                continue
 
            event = json.loads(msg.value().decode('utf-8'))
 
            # フィルタリング
            if not all(f(event) for f in self.filters):
                self.consumer.commit(asynchronous=False)
                continue
 
            # 変換
            for transform in self.transformers:
                event = transform(event)
 
            # 集計
            self.counter.add(event.get("category", "unknown"))
 
            # 出力
            self.producer.produce(
                topic=self.output_topic,
                key=event.get("key", "").encode('utf-8'),
                value=json.dumps(event).encode('utf-8'),
            )
            self.consumer.commit(asynchronous=False)
 
 
# 使用例: 注文イベントから高額注文のみをフィルタリングして集計
processor = StreamProcessor(consumer, producer,
                           "order-events", "high-value-orders")
processor.add_filter(lambda e: e.get("amount", 0) > 10000)
processor.add_transformer(lambda e: {**e, "flagged": True})
# processor.process()

# RabbitMQ Producer (Python - pika)
import pika
import json
import uuid
 
connection = pika.BlockingConnection(
    pika.ConnectionParameters(
        host='localhost',
        credentials=pika.PlainCredentials('app_user', 'secret'),
        heartbeat=600,
        blocked_connection_timeout=300,
    )
)
channel = connection.channel()
 
# Exchange と Queue を宣言
channel.exchange_declare(exchange='order_exchange',
                        exchange_type='topic', durable=True)
channel.queue_declare(queue='order_processing', durable=True, arguments={
    'x-dead-letter-exchange': 'dlx_exchange',        # DLQ 設定
    'x-dead-letter-routing-key': 'order.failed',
    'x-message-ttl': 86400000,                        # TTL: 24時間
    'x-max-length': 100000,                           # キュー最大長
    'x-overflow': 'reject-publish',                   # 溢れ時に拒否
})
channel.queue_bind(exchange='order_exchange', queue='order_processing',
                   routing_key='order.created')
 
# メッセージ送信
message = json.dumps({"order_id": 123, "amount": 5000, "currency": "JPY"})
channel.basic_publish(
    exchange='order_exchange',
    routing_key='order.created',
    body=message,
    properties=pika.BasicProperties(
        delivery_mode=2,                  # メッセージ永続化
        content_type='application/json',
        message_id=str(uuid.uuid4()),
        timestamp=int(time.time()),
        headers={'retry_count': 0},
    ),
)
print(f"送信完了: {message}")
connection.close()

# RabbitMQ Consumer (Python - pika) with retry logic
import pika
import json
import traceback
 
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue='order_processing', durable=True)
channel.basic_qos(prefetch_count=10)    # 同時処理数を制御
 
MAX_RETRIES = 3
 
def callback(ch, method, properties, body):
    order = json.loads(body)
    retry_count = (properties.headers or {}).get('retry_count', 0)
    print(f"処理開始: order_id={order['order_id']} (retry={retry_count})")
 
    try:
        process_order(order)
        ch.basic_ack(delivery_tag=method.delivery_tag)        # 成功 → Ack
        print(f"処理完了: order_id={order['order_id']}")
    except Exception as e:
        print(f"処理失敗: {e}")
        traceback.print_exc()
 
        if retry_count < MAX_RETRIES:
            # リトライ: 同じキューに再投入（retry_count をインクリメント）
            ch.basic_publish(
                exchange='',
                routing_key='order_processing',
                body=body,
                properties=pika.BasicProperties(
                    delivery_mode=2,
                    headers={'retry_count': retry_count + 1},
                ),
            )
            ch.basic_ack(delivery_tag=method.delivery_tag)
            print(f"リトライ予約: order_id={order['order_id']} "
                  f"retry={retry_count + 1}")
        else:
            # リトライ上限超過 → DLQ へ
            ch.basic_nack(delivery_tag=method.delivery_tag,
                          requeue=False)
            print(f"DLQ送信: order_id={order['order_id']} (max retries exceeded)")
 
channel.basic_consume(queue='order_processing', on_message_callback=callback)
print("Consumer 起動完了。メッセージ待機中...")
channel.start_consuming()

# Amazon SQS (Python - boto3)
import boto3
import json
import uuid
import time
 
sqs = boto3.client('sqs', region_name='ap-northeast-1')
QUEUE_URL = 'https://sqs.ap-northeast-1.amazonaws.com/123456789/order-queue.fifo'
 
# --- Producer ---
response = sqs.send_message(
    QueueUrl=QUEUE_URL,
    MessageBody=json.dumps({"order_id": 456, "amount": 8000}),
    MessageAttributes={
        'EventType': {'DataType': 'String', 'StringValue': 'OrderCreated'},
        'Priority': {'DataType': 'Number', 'StringValue': '1'},
    },
    MessageDeduplicationId=str(uuid.uuid4()),  # FIFO: 5分間の重複排除
    MessageGroupId='user-group-42',            # FIFO: 同グループ内で順序保証
)
print(f"送信完了 MessageId: {response['MessageId']}")
 
# --- Consumer (ロングポーリング) ---
while True:
    response = sqs.receive_message(
        QueueUrl=QUEUE_URL,
        MaxNumberOfMessages=10,         # 最大10件一括取得
        WaitTimeSeconds=20,             # ロングポーリング（20秒待機）
        VisibilityTimeout=120,          # 処理中の非表示期間
        MessageAttributeNames=['All'],
    )
    for message in response.get('Messages', []):
        body = json.loads(message['Body'])
        print(f"処理中: order_id={body['order_id']}")
 
        try:
            process_order(body)
            sqs.delete_message(
                QueueUrl=QUEUE_URL,
                ReceiptHandle=message['ReceiptHandle'],
            )
            print(f"処理完了・削除: order_id={body['order_id']}")
        except Exception as e:
            # VisibilityTimeout 後に自動的にキューに戻る
            print(f"処理失敗: {e} → VisibilityTimeout後に再配信")
 
    if not response.get('Messages'):
        time.sleep(1)  # メッセージなし → 短い待機

# べき等 Consumer の実装例
import redis
import hashlib
import json
 
redis_client = redis.Redis(host='localhost', port=6379, db=0)
 
class IdempotentConsumer:
    """同じメッセージを何度受信しても結果が変わらないことを保証する
 
    実装方式:
    1. メッセージIDベース: message_id で重複チェック
    2. ビジネスキーベース: entity_id + version で重複チェック
    3. ハッシュベース: メッセージ内容のハッシュで重複チェック
    """
 
    def __init__(self, redis_client: redis.Redis,
                 dedup_ttl: int = 86400 * 7):
        self.redis = redis_client
        self.dedup_ttl = dedup_ttl  # 重複チェックの保持期間
 
    def process(self, message: dict) -> bool:
        """べき等にメッセージを処理する
 
        Returns:
            True: 処理成功（新規メッセージ）
            False: スキップ（処理済みメッセージ）
        """
        message_id = message.get('message_id')
        if not message_id:
            # message_id がない場合はコンテンツハッシュを使用
            content = json.dumps(message, sort_keys=True)
            message_id = hashlib.sha256(content.encode()).hexdigest()
 
        idempotency_key = f"processed:{message_id}"
 
        # SETNX (Set if Not eXists) で排他制御
        if not self.redis.set(idempotency_key, 'processing',
                              nx=True, ex=self.dedup_ttl):
            print(f"[SKIP] 処理済み or 処理中: {message_id}")
            return False
 
        try:
            # ビジネスロジック実行
            result = self._execute_business_logic(message)
 
            # 処理完了マーク
            self.redis.set(idempotency_key, json.dumps({
                'status': 'completed',
                'result': result,
                'processed_at': time.time(),
            }), ex=self.dedup_ttl)
            return True
 
        except Exception as e:
            # 失敗時はキーを削除してリトライ可能にする
            self.redis.delete(idempotency_key)
            raise
 
    def _execute_business_logic(self, message: dict):
        """ビジネスロジック（サブクラスでオーバーライド）"""
        raise NotImplementedError
 
 
class OrderConsumer(IdempotentConsumer):
    """注文処理の具体的な Consumer"""
 
    def _execute_business_logic(self, message: dict):
        order_id = message['order_id']
        amount = message['amount']
        print(f"[PROCESS] 注文処理: order_id={order_id}, amount={amount}")
        # DB更新、外部API呼び出し等
        return {"order_id": order_id, "status": "confirmed"}
 
 
# 使用例
consumer = OrderConsumer(redis_client)
messages = [
    {"message_id": "msg-001", "order_id": 123, "amount": 5000},
    {"message_id": "msg-001", "order_id": 123, "amount": 5000},  # 重複
    {"message_id": "msg-002", "order_id": 456, "amount": 8000},
]
 
for msg in messages:
    result = consumer.process(msg)
    print(f"  → processed={result}")
# [PROCESS] 注文処理: order_id=123, amount=5000
#   → processed=True
# [SKIP] 処理済み or 処理中: msg-001
#   → processed=False
# [PROCESS] 注文処理: order_id=456, amount=8000
#   → processed=True

# Outbox パターン: DB トランザクションとメッセージ発行の原子性を保証
 
class OutboxPattern:
    """Outbox パターンの実装
 
    問題: DB更新とメッセージ発行の2つの操作を原子的に行えない
    → DB更新成功 + メッセージ発行失敗 = 不整合
 
    解決: DB トランザクション内で outbox テーブルにも書き込み、
    別プロセスが outbox を読んでメッセージブローカーに発行する
 
    DB Transaction:
      1. orders テーブルに INSERT
      2. outbox テーブルに INSERT (同一トランザクション)
 
    Outbox Relay (別プロセス):
      1. outbox テーブルをポーリング
      2. 未送信メッセージをブローカーに発行
      3. 送信済みマークを付与
    """
 
    def __init__(self, db_session, producer):
        self.db = db_session
        self.producer = producer
 
    def create_order(self, order_data: dict):
        """注文作成 + イベント発行（原子的）"""
        with self.db.begin():
            # 1. 注文を保存
            order = Order(**order_data)
            self.db.add(order)
            self.db.flush()  # IDを取得
 
            # 2. Outboxに書き込み（同一トランザクション）
            outbox_event = OutboxEvent(
                aggregate_type='Order',
                aggregate_id=str(order.id),
                event_type='OrderCreated',
                payload=json.dumps({
                    'order_id': order.id,
                    'user_id': order_data['user_id'],
                    'amount': order_data['amount'],
                    'created_at': time.time(),
                }),
                status='PENDING',
            )
            self.db.add(outbox_event)
            # トランザクション完了 → 両方が原子的にコミット
 
    def relay_outbox_events(self, batch_size: int = 100):
        """Outbox テーブルの未送信イベントをブローカーに発行"""
        pending_events = self.db.query(OutboxEvent)\
            .filter_by(status='PENDING')\
            .order_by(OutboxEvent.created_at)\
            .limit(batch_size)\
            .all()
 
        for event in pending_events:
            try:
                self.producer.produce(
                    topic=f"{event.aggregate_type.lower()}-events",
                    key=event.aggregate_id.encode('utf-8'),
                    value=event.payload.encode('utf-8'),
                )
                event.status = 'SENT'
                event.sent_at = time.time()
            except Exception as e:
                print(f"[OUTBOX ERROR] {event.id}: {e}")
 
        self.db.commit()
        self.producer.flush()

# NG: 大きなペイロードをキューに直接格納
 
class BadProducer:
    def send_invoice(self, order_id: int, pdf_data: bytes, images: list[bytes]):
        message = {
            "order_id": order_id,
            "pdf_invoice": base64.b64encode(pdf_data).decode(),  # 10MB
            "images": [base64.b64encode(img).decode() for img in images],  # 数MB
        }
        # 問題: ブローカーのメモリ/ディスクを圧迫
        # 問題: ネットワーク帯域を浪費
        # 問題: Consumer のデシリアライズが遅い
        self.producer.send("invoices", json.dumps(message).encode())
 
 
# OK: Claim-Check パターン（参照のみキューに格納）
 
import boto3
 
class GoodProducer:
    def __init__(self, producer, s3_client):
        self.producer = producer
        self.s3 = s3_client
 
    def send_invoice(self, order_id: int, pdf_data: bytes, images: list[bytes]):
        # Step 1: 大きなデータは S3 に保存
        pdf_key = f"invoices/{order_id}/invoice.pdf"
        self.s3.put_object(Bucket='my-bucket', Key=pdf_key, Body=pdf_data)
 
        image_keys = []
        for i, img in enumerate(images):
            key = f"invoices/{order_id}/image_{i}.jpg"
            self.s3.put_object(Bucket='my-bucket', Key=key, Body=img)
            image_keys.append(key)
 
        # Step 2: 軽量な参照のみキューに格納
        message = {
            "order_id": order_id,
            "invoice_s3_key": pdf_key,
            "image_s3_keys": image_keys,
        }
        self.producer.send("invoices", json.dumps(message).encode())
        # メッセージサイズ: 数百バイト（vs 数十MB）

# NG: Auto-Ack で fire-and-forget
 
def bad_consumer():
    channel.basic_consume(
        queue='payment_queue',
        auto_ack=True,  # 問題: 受信した瞬間にAck
        on_message_callback=process_payment
    )
    # Consumer がクラッシュ → メッセージ消失 → 決済データ欠損
 
 
# OK: 手動 Ack + DLQ + べき等処理
 
def good_consumer():
    channel.basic_qos(prefetch_count=5)  # 同時処理数を制限
    channel.basic_consume(
        queue='payment_queue',
        auto_ack=False,  # 手動Ack
        on_message_callback=safe_process_payment
    )
 
def safe_process_payment(ch, method, properties, body):
    try:
        # べき等処理（重複を安全にスキップ）
        payment = json.loads(body)
        if is_already_processed(payment['payment_id']):
            ch.basic_ack(delivery_tag=method.delivery_tag)
            return
 
        execute_payment(payment)
        mark_as_processed(payment['payment_id'])
        ch.basic_ack(delivery_tag=method.delivery_tag)
    except Exception:
        # 失敗 → DLQ へ（requeue=False で無限ループ防止）
        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)

# NG: Consumer 1台で大量メッセージを処理
 
class UnscalableConsumer:
    """Producer 1000 msg/sec vs Consumer 100 msg/sec
    → キューが無限に溜まり続け、遅延が際限なく増大"""
    pass
 
 
# OK: オートスケーリング + バックプレッシャー
 
class ScalableConsumerManager:
    """キュー深度に基づくオートスケーリング"""
 
    def __init__(self, queue_name: str,
                 target_lag: int = 1000,
                 max_consumers: int = 20,
                 min_consumers: int = 2):
        self.queue_name = queue_name
        self.target_lag = target_lag
        self.max_consumers = max_consumers
        self.min_consumers = min_consumers
        self.current_consumers = min_consumers
 
    def check_and_scale(self, current_lag: int):
        """キュー深度に基づいてConsumer数を調整"""
        if current_lag > self.target_lag * 2:
            # スケールアウト
            desired = min(
                self.current_consumers * 2,
                self.max_consumers
            )
            self._scale_to(desired)
            print(f"[SCALE OUT] {self.current_consumers} → {desired} "
                  f"(lag={current_lag})")
        elif current_lag < self.target_lag * 0.2:
            # スケールイン
            desired = max(
                self.current_consumers // 2,
                self.min_consumers
            )
            self._scale_to(desired)
            print(f"[SCALE IN] {self.current_consumers} → {desired} "
                  f"(lag={current_lag})")
 
    def _scale_to(self, count: int):
        self.current_consumers = count
        # 実際にはKubernetes HPA, ECS Service等で実装

# ヒント: IdempotentConsumer クラスを使用
consumer = OrderConsumer(redis.Redis())
messages = [
    {"message_id": f"msg-{i}", "order_id": i, "amount": 1000 * i}
    for i in range(7)
]
# 重複を追加
messages.extend([
    {"message_id": "msg-0", "order_id": 0, "amount": 0},
    {"message_id": "msg-3", "order_id": 3, "amount": 3000},
    {"message_id": "msg-5", "order_id": 5, "amount": 5000},
])
 
processed = sum(1 for msg in messages if consumer.process(msg))
skipped = len(messages) - processed
print(f"処理: {processed}, スキップ: {skipped}")

import random
 
class RetryableConsumer:
    def __init__(self, max_retries: int = 3, failure_rate: float = 0.3):
        self.max_retries = max_retries
        self.failure_rate = failure_rate
        self.success_count = 0
        self.dlq_count = 0
 
    def process_with_retry(self, message: dict) -> str:
        for attempt in range(self.max_retries + 1):
            if random.random() > self.failure_rate:
                self.success_count += 1
                return "success"
        self.dlq_count += 1
        return "dlq"
 
consumer = RetryableConsumer(max_retries=3, failure_rate=0.3)
random.seed(42)
for i in range(100):
    consumer.process_with_retry({"id": i})
print(f"成功: {consumer.success_count}, DLQ: {consumer.dlq_count}")

# 演習1: 基本実装のテンプレート
class Exercise1:
    """基本的な実装パターンの演習"""
 
    def __init__(self):
        self.data = []
 
    def validate_input(self, value):
        """入力値の検証"""
        if value is None:
            raise ValueError("入力値がNoneです")
        return True
 
    def process(self, value):
        """データ処理のメインロジック"""
        self.validate_input(value)
        self.data.append(value)
        return self.data
 
    def get_results(self):
        """処理結果の取得"""
        return {
            'count': len(self.data),
            'data': self.data
        }
 
# テスト
def test_exercise1():
    ex = Exercise1()
    assert ex.process(1) == [1]
    assert ex.process(2) == [1, 2]
    assert ex.get_results()['count'] == 2
 
    try:
        ex.process(None)
        assert False, "例外が発生するべき"
    except ValueError:
        pass
 
    print("全テスト合格!")
 
test_exercise1()

# 演習2: 応用パターン
from typing import List, Dict, Optional
from datetime import datetime
 
class AdvancedExercise:
    """応用パターンの演習"""
 
    def __init__(self, max_size: int = 100):
        self._items: List[Dict] = []
        self._max_size = max_size
        self._created_at = datetime.now()
 
    def add(self, key: str, value: any) -> bool:
        """アイテムの追加（サイズ制限付き）"""
        if len(self._items) >= self._max_size:
            return False
        self._items.append({
            'key': key,
            'value': value,
            'timestamp': datetime.now().isoformat()
        })
        return True
 
    def find(self, key: str) -> Optional[Dict]:
        """キーによる検索"""
        for item in reversed(self._items):
            if item['key'] == key:
                return item
        return None
 
    def remove(self, key: str) -> bool:
        """キーによる削除"""
        for i, item in enumerate(self._items):
            if item['key'] == key:
                self._items.pop(i)
                return True
        return False
 
    def stats(self) -> Dict:
        """統計情報"""
        return {
            'total_items': len(self._items),
            'max_size': self._max_size,
            'usage_percent': len(self._items) / self._max_size * 100,
            'uptime': str(datetime.now() - self._created_at)
        }
 
# テスト
def test_advanced():
    ex = AdvancedExercise(max_size=3)
    assert ex.add("a", 1) == True
    assert ex.add("b", 2) == True
    assert ex.add("c", 3) == True
    assert ex.add("d", 4) == False  # サイズ制限
    assert ex.find("b")['value'] == 2
    assert ex.remove("b") == True
    assert ex.find("b") is None
    stats = ex.stats()
    assert stats['total_items'] == 2
    print("応用テスト全合格!")
 
test_advanced()

# 演習3: パフォーマンス最適化
import time
from functools import lru_cache
 
# 最適化前（O(n^2)）
def slow_search(data: list, target: int) -> int:
    """非効率な検索"""
    for i in range(len(data)):
        for j in range(i + 1, len(data)):
            if data[i] + data[j] == target:
                return (i, j)
    return (-1, -1)
 
# 最適化後（O(n)）
def fast_search(data: list, target: int) -> tuple:
    """ハッシュマップを使った効率的な検索"""
    seen = {}
    for i, num in enumerate(data):
        complement = target - num
        if complement in seen:
            return (seen[complement], i)
        seen[num] = i
    return (-1, -1)
 
# ベンチマーク
def benchmark():
    import random
    data = list(range(5000))
    random.shuffle(data)
    target = data[100] + data[4000]
 
    start = time.time()
    result1 = slow_search(data, target)
    slow_time = time.time() - start
 
    start = time.time()
    result2 = fast_search(data, target)
    fast_time = time.time() - start
 
    print(f"非効率版: {slow_time:.4f}秒")
    print(f"効率版:   {fast_time:.6f}秒")
    print(f"高速化率: {slow_time/fast_time:.0f}倍")
 
benchmark()

# デバッグ用ユーティリティ
import logging
import traceback
from functools import wraps
 
# ロガーの設定
logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s [%(levelname)s] %(name)s: %(message)s'
)
logger = logging.getLogger(__name__)
 
def debug_decorator(func):
    """関数の入出力をログ出力するデコレータ"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        logger.debug(f"呼び出し: {func.__name__}(args={args}, kwargs={kwargs})")
        try:
            result = func(*args, **kwargs)
            logger.debug(f"戻り値: {func.__name__} -> {result}")
            return result
        except Exception as e:
            logger.error(f"例外発生: {func.__name__}: {e}")
            logger.error(traceback.format_exc())
            raise
    return wrapper
 
@debug_decorator
def process_data(items):
    """データ処理（デバッグ対象）"""
    if not items:
        raise ValueError("空のデータ")
    return [item * 2 for item in items]