def add_url(self, url: str) -> bool:
    # 1. 使用k个哈希函数计算url的k个位置
    positions = self._get_hash_positions(url)
    
    # 2. 将位数组中这k个位置设置为1
    for pos in positions:
        self.redis_client.setbit(self.bloom_key, pos, 1)
    
    # 3. 同时将URL添加到精确集合中（用于消除误判）
    self.redis_client.sadd(self.exact_key, url)

def might_contain(self, url: str) -> bool:
    # 1. 使用相同的k个哈希函数计算位置
    positions = self._get_hash_positions(url)
    
    # 2. 检查位数组中这k个位置是否都为1
    for pos in positions:
        if not self.redis_client.getbit(self.bloom_key, pos):
            return False
    return True

from celery_task.s3.redis_bloom_filter import S3UrlCacheManager

# 创建缓存管理器
cache_manager = S3UrlCacheManager()

# 添加单个URL
await cache_manager.add_url("s3://bucket/file1.pdf")

# 批量添加URL
urls = ["s3://bucket/file1.pdf", "s3://bucket/file2.txt"]
await cache_manager.add_urls_batch(urls)

# 检查URL是否存在
exists = await cache_manager.might_contain("s3://bucket/file1.pdf")
exact_exists = await cache_manager.contains_exact("s3://bucket/file1.pdf")

# 获取S3文件迭代器
s3_iterator = s3_service.get_files_async_iterator(
    batch_size=1000,
    cache_urls_for_cleanup=True
)

# 定义进度回调函数
def progress_callback(cached_count):
    if cached_count % 50000 == 0:
        print(f"已缓存 {cached_count} 个URL")

# 缓存所有S3 URL
total_cached = await cache_manager.cache_s3_urls_from_iterator(
    s3_iterator, 
    progress_callback
)

# 获取Weaviate中的所有URL
weaviate_urls = ["s3://bucket/file1.pdf", "s3://bucket/file2.txt"]

# 使用布隆过滤器快速识别孤立URL
orphaned_urls = await cache_manager.find_orphaned_urls(weaviate_urls)

# 获取缓存统计信息
stats = await cache_manager.get_stats()
print(f"布隆过滤器存在: {stats.get('bloom_exists', False)}")
print(f"精确URL数量: {stats.get('exact_urls_count', 0)}")
print(f"位数组大小: {stats.get('bit_array_size', 0)}")

# 清空缓存
await cache_manager.clear_cache()

# Redis连接配置
REDIS_HOST = "localhost"
REDIS_PORT = 6379
REDIS_DB = 0
REDIS_PASSWORD = None

# 布隆过滤器配置
BLOOM_FILTER_KEY = "s3_urls_bloom"
EXACT_URLS_KEY = "s3_urls_exact"

# 预期元素数量（影响内存使用）
EXPECTED_ELEMENTS = 1000000

# 误判率（影响内存使用和查询性能）
FALSE_POSITIVE_RATE = 0.01

# 哈希函数数量（影响查询性能）
NUM_HASH_FUNCTIONS = 7

# 获取详细统计信息
stats = await cache_manager.get_stats()
print(f"""
缓存统计:
- 布隆过滤器存在: {stats.get('bloom_exists', False)}
- 精确URL数量: {stats.get('exact_urls_count', 0)}
- 位数组大小: {stats.get('bit_array_size', 0)}
- 哈希函数数量: {stats.get('num_hash_functions', 0)}
- 预期误判率: {stats.get('false_positive_rate', 0)}
""")

import time

# 测试添加性能
start_time = time.time()
for i in range(10000):
    await cache_manager.add_url(f"s3://bucket/file{i}.pdf")
add_time = time.time() - start_time
print(f"添加10000个URL耗时: {add_time:.2f}秒")

# 测试查询性能
start_time = time.time()
for i in range(10000):
    await cache_manager.might_contain(f"s3://bucket/file{i}.pdf")
query_time = time.time() - start_time
print(f"查询10000个URL耗时: {query_time:.2f}秒")

# 检查Redis连接
import redis
r = redis.Redis(host='localhost', port=6379, db=0)
try:
    r.ping()
    print("Redis连接正常")
except:
    print("Redis连接失败")

# 检查Redis内存使用
info = r.info('memory')
print(f"Redis内存使用: {info['used_memory_human']}")
print(f"Redis内存峰值: {info['used_memory_peak_human']}")

# 调整布隆过滤器参数
# 增加位数组大小或减少预期元素数量
EXPECTED_ELEMENTS = 500000  # 减少预期元素数量
FALSE_POSITIVE_RATE = 0.005  # 降低误判率

import logging
logging.basicConfig(level=logging.DEBUG)

# 使用Redis MONITOR命令监控操作
# redis-cli monitor

import cProfile
import pstats

# 性能分析
profiler = cProfile.Profile()
profiler.enable()
# 执行操作
profiler.disable()
stats = pstats.Stats(profiler)
stats.sort_stats('cumulative')
stats.print_stats(10)

"""
Redis-based Bloom Filter for S3 URL caching and efficient orphaned document detection

This module implements a scalable solution for handling millions of S3 URLs using:
1. Redis-based Bloom Filter for fast membership testing
2. Redis Sets for exact URL storage
3. Optimized batch processing for large-scale data
"""

import hashlib
import math
from typing import Iterator, List, Optional, Set

import redis

from app.config.config import settings
from app.utils.logger import celery_logger as logger


class RedisBloomFilter:
    """
    Redis-based Bloom Filter implementation for S3 URL caching

    Features:
    - False positive rate configurable
    - Memory efficient for millions of URLs
    - Automatic bit array sizing
    - Batch operations for better performance
    """

    def __init__(self, redis_client: redis.Redis, key_prefix: str = "s3_bloom",
                 expected_items: int = 1000000, false_positive_rate: float = 0.01):
        """
        Initialize Redis Bloom Filter

        Args:
            redis_client: Redis client instance
            key_prefix: Redis key prefix for bloom filter
            expected_items: Expected number of items (for optimal sizing)
            false_positive_rate: Desired false positive rate (0.01 = 1%)
        """
        self.redis = redis_client
        self.key_prefix = key_prefix
        self.expected_items = expected_items
        self.false_positive_rate = false_positive_rate

        # Calculate optimal parameters
        self.bit_size = self._calculate_bit_size()
        self.hash_count = self._calculate_hash_count()

        # Redis keys
        self.bloom_key = f"{key_prefix}:bloom"
        self.urls_set_key = f"{key_prefix}:urls"
        self.metadata_key = f"{key_prefix}:meta"

        logger.info(f"🎯 Initialized RedisBloomFilter: "
                    f"bit_size={self.bit_size}, hash_count={self.hash_count}, "
                    f"expected_items={expected_items}, fpr={false_positive_rate}")

    def _calculate_bit_size(self) -> int:
        """Calculate optimal bit array size"""
        # m = -n * ln(p) / (ln(2)^2)
        n = self.expected_items
        p = self.false_positive_rate
        m = int(-n * math.log(p) / (math.log(2) ** 2))
        return m

    def _calculate_hash_count(self) -> int:
        """Calculate optimal number of hash functions"""
        # k = (m/n) * ln(2)
        k = int((self.bit_size / self.expected_items) * math.log(2))
        return max(1, k)

    def _hash_functions(self, item: str) -> List[int]:
        """Generate multiple hash values for an item"""
        hashes = []
        # Use different hash algorithms for diversity
        hash1 = int(hashlib.md5(item.encode()).hexdigest(), 16)
        hash2 = int(hashlib.sha1(item.encode()).hexdigest(), 16)

        # Generate k hash values using double hashing
        for i in range(self.hash_count):
            hash_val = (hash1 + i * hash2) % self.bit_size
            hashes.append(hash_val)
        return hashes

    async def add_url(self, url: str) -> bool:
        """
        Add a URL to both bloom filter and exact set

        Args:
            url: S3 URL to add

        Returns:
            True if added successfully
        """
        try:
            # Add to bloom filter
            hash_positions = self._hash_functions(url)

            # Use pipeline for atomic operations
            pipe = self.redis.pipeline()

            # Set bits in bloom filter
            for pos in hash_positions:
                pipe.setbit(self.bloom_key, pos, 1)

            # Add to exact URLs set
            pipe.sadd(self.urls_set_key, url)

            # Execute pipeline
            pipe.execute()

            return True

        except Exception as e:
            logger.error(f"❌ Failed to add URL to bloom filter: {e}")
            return False

    async def add_urls_batch(self, urls: List[str], batch_size: int = 1000) -> int:
        """
        Add multiple URLs in batches for better performance

        Args:
            urls: List of S3 URLs to add
            batch_size: Number of URLs to process in each batch

        Returns:
            Number of URLs successfully added
        """
        added_count = 0

        for i in range(0, len(urls), batch_size):
            batch = urls[i:i + batch_size]

            try:
                # Use pipeline for batch operations
                pipe = self.redis.pipeline()
                for url in batch:
                    # Add to bloom filter
                    hash_positions = self._hash_functions(url)
                    for pos in hash_positions:
                        pipe.setbit(self.bloom_key, pos, 1)

                    # Add to exact URLs set
                    pipe.sadd(self.urls_set_key, url)

                # Execute batch
                pipe.execute()
                added_count += len(batch)

                logger.debug(f"📦 Added batch of {len(batch)} URLs to bloom filter")

            except Exception as e:
                logger.error(f"❌ Failed to add URL batch: {e}")
                continue

        logger.info(f"✅ Added {added_count} URLs to bloom filter")
        return added_count

    async def might_contain(self, url: str) -> bool:
        """
        Check if URL might be in the set (bloom filter check)

        Args:
            url: URL to check

        Returns:
            True if URL might be in set (could be false positive)
            False if URL is definitely not in set
        """
        try:
            hash_positions = self._hash_functions(url)

            # Check all bit positions
            pipe = self.redis.pipeline()
            for pos in hash_positions:
                pipe.getbit(self.bloom_key, pos)

            results = pipe.execute()

            # URL might be in set only if all bits are set
            return all(bit == 1 for bit in results)

        except Exception as e:
            logger.error(f"❌ Bloom filter check failed: {e}")
            return True  # Conservative approach

    async def contains_exact(self, url: str) -> bool:
        """
        Check if URL exists in exact set (no false positives)

        Args:
            url: URL to check

        Returns:
            True if URL exists in exact set
        """
        try:
            return bool(self.redis.sismember(self.urls_set_key, url))
        except Exception as e:
            logger.error(f"❌ Exact set check failed: {e}")
            return False

    async def get_missing_urls(self, urls_to_check: List[str],
                               batch_size: int = 10000) -> Set[str]:
        """
        Efficiently find URLs that are NOT in S3 (missing URLs)

        This is optimized for the orphaned document cleanup use case:
        1. First use bloom filter to quickly eliminate definitely present URLs
        2. Then check exact set for potential matches

        Args:
            urls_to_check: List of URLs to check (from Weaviate)
            batch_size: Batch size for processing

        Returns:
            Set of URLs that are missing from S3
        """
        missing_urls = set()
        total_checked = 0
        bloom_filtered = 0

        logger.info(f"🔍 Checking {len(urls_to_check)} URLs for missing entries")

        for i in range(0, len(urls_to_check), batch_size):
            batch = urls_to_check[i:i + batch_size]

            for url in batch:
                total_checked += 1

                # Step 1: Bloom filter check (fast elimination)
                if not await self.might_contain(url):
                    # Definitely not in S3
                    missing_urls.add(url)
                    bloom_filtered += 1
                else:
                    # Might be in S3, need exact check
                    if not await self.contains_exact(url):
                        missing_urls.add(url)

            if i % (batch_size * 10) == 0:  # Log every 10 batches
                logger.info(f"📊 Processed {total_checked}/{len(urls_to_check)} URLs, "
                            f"bloom filtered: {bloom_filtered}, missing: {len(missing_urls)}")

        logger.info(f"✅ Found {len(missing_urls)} missing URLs out of {total_checked} checked "
                    f"(bloom filter eliminated {bloom_filtered} checks)")

        return missing_urls

    async def clear_cache(self) -> bool:
        """Clear all cached data"""
        try:
            pipe = self.redis.pipeline()
            pipe.delete(self.bloom_key)
            pipe.delete(self.urls_set_key)
            pipe.delete(self.metadata_key)
            pipe.execute()

            logger.info("🧹 Cleared bloom filter cache")
            return True

        except Exception as e:
            logger.error(f"❌ Failed to clear cache: {e}")
            return False

    async def get_cache_stats(self) -> dict:
        """Get cache statistics"""
        try:
            pipe = self.redis.pipeline()
            pipe.memory_usage(self.bloom_key)
            pipe.scard(self.urls_set_key)
            pipe.exists(self.bloom_key)
            results = pipe.execute()
            return {
                "bloom_memory_bytes": results[0] or 0,
                "exact_urls_count": results[1] or 0,
                "bloom_exists": bool(results[2]),
                "estimated_bit_size": self.bit_size,
                "hash_count": self.hash_count,
                "expected_false_positive_rate": self.false_positive_rate
            }

        except Exception as e:
            logger.error(f"❌ Failed to get cache stats: {e}")
            return {}


class S3UrlCacheManager:
    """
    High-level manager for S3 URL caching with bloom filter optimization
    """

    def __init__(self, redis_client: Optional[redis.Redis] = None):
        """Initialize cache manager"""
        if redis_client is None:
            # Use default Redis connection
            redis_client = redis.Redis(
                host=getattr(settings, 'REDIS_HOST', 'localhost'),
                port=getattr(settings, 'REDIS_PORT', 6379),
                db=getattr(settings, 'REDIS_DB', 7),
                decode_responses=True
            )

        self.bloom_filter = RedisBloomFilter(
            redis_client=redis_client,
            key_prefix="s3_orphan_cleanup",
            expected_items=2000000,  # 2M URLs
            false_positive_rate=0.001  # 0.1% false positive rate
        )

    async def cache_s3_urls_from_iterator(self, s3_iterator,
                                          progress_callback=None) -> int:
        """
        Cache S3 URLs from async iterator

        Args:
            s3_iterator: Async iterator yielding batches of file info
            progress_callback: Optional callback for progress updates

        Returns:
            Total number of URLs cached
        """
        total_cached = 0
        batch_urls = []

        logger.info("🔄 Starting S3 URL caching process")

        async for file_batch in s3_iterator:
            # Extract URLs from file batch
            urls = [file_info['url'] for file_info in file_batch]
            batch_urls.extend(urls)

            # Process in larger batches for efficiency
            if len(batch_urls) >= 5000:
                cached_count = await self.bloom_filter.add_urls_batch(batch_urls)
                total_cached += cached_count
                batch_urls = []

                if progress_callback:
                    progress_callback(total_cached)

        # Process remaining URLs
        if batch_urls:
            cached_count = await self.bloom_filter.add_urls_batch(batch_urls)
            total_cached += cached_count

        logger.info(f"✅ Cached {total_cached} S3 URLs in bloom filter")
        return total_cached

    async def find_orphaned_urls(self, weaviate_urls: List[str]) -> Set[str]:
        """
        Find URLs that exist in Weaviate but not in S3 (orphaned)

        Args:
            weaviate_urls: List of URLs from Weaviate

        Returns:
            Set of orphaned URLs
        """
        return await self.bloom_filter.get_missing_urls(weaviate_urls)

    async def get_stats(self) -> dict:
        """Get caching statistics"""
        return await self.bloom_filter.get_cache_stats()

    async def clear_cache(self) -> bool:
        """Clear all cached data"""
        return await self.bloom_filter.clear_cache()