import redis
import time
def test_hyperloglog():
    # 連接 Redis
    r = redis.Redis(host='127.0.0.1', port=6379, db=0, decode_responses=True)
    # 清空測(cè)試 Key
    r.delete('hll:uv:daily')
    # 準(zhǔn)備批量添加
    batch_size = 1000
    total_users = 1000000
    start_time = time.time()
    # 批量添加 100 萬(wàn)用戶
    for i in range(0, total_users, batch_size):
        users = [f"user_{j}" for j in range(i + 1, min(i + batch_size + 1, total_users + 1))]
        r.pfadd('hll:uv:daily', *users)
    # 統(tǒng)計(jì)數(shù)量
    uv_count = r.pfcount('hll:uv:daily')
    # 獲取內(nèi)存占用
    memory_used = r.memory_usage('hll:uv:daily')
    elapsed_time = time.time() - start_time
    print(f"實(shí)際添加用戶數(shù): {total_users}")
    print(f"HyperLogLog 統(tǒng)計(jì)結(jié)果: {uv_count}")
    print(f"誤差率: {abs(uv_count - total_users) / total_users * 100:.4f}%")
    print(f"內(nèi)存占用: {memory_used} bytes ({memory_used / 1024:.2f} KB)")
    print(f"耗時(shí): {elapsed_time:.4f} 秒")
    print(f"吞吐量: {total_users / elapsed_time:.0f} ops/sec")
if __name__ == "__main__":
    test_hyperloglog()

實(shí)際添加用戶數(shù): 1000000
HyperLogLog 統(tǒng)計(jì)結(jié)果: 1008542
誤差率: 0.8542%
內(nèi)存占用: 12288 bytes (12.00 KB)
耗時(shí): 2.3456 秒
吞吐量: 426315 ops/sec

# 日級(jí) UV 統(tǒng)計(jì)
uv:daily:{YYYY-MM-DD}  # 例如: uv:daily:2024-01-15
# 月級(jí) UV 統(tǒng)計(jì)（通過(guò) PFCOUNT 合并日級(jí)數(shù)據(jù)）
uv:monthly:{YYYY-MM}    # 例如: uv:monthly:2024-01
# 全站歷史 UV（通過(guò) PFMERGE 合并月級(jí)數(shù)據(jù)）
uv:lifetime

import redis
from datetime import datetime, timedelta
from typing import List
class UVStatisticsService:
    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client
    def record_visit(self, user_id: int, visit_date: datetime = None):
        """
        記錄用戶訪問(wèn)
        :param user_id: 用戶 ID
        :param visit_date: 訪問(wèn)日期（默認(rèn)當(dāng)天）
        """
        if visit_date is None:
            visit_date = datetime.now()
        date_key = visit_date.strftime("%Y-%m-%d")
        key = f"uv:daily:{date_key}"
        # 添加用戶到當(dāng)日 HyperLogLog
        self.redis.pfadd(key, str(user_id))
        # 設(shè)置過(guò)期時(shí)間（保留 90 天數(shù)據(jù)）
        self.redis.expire(key, 90 * 24 * 60 * 60)
    def get_daily_uv(self, date: datetime = None) -> int:
        """
        獲取指定日期的 UV
        :param date: 查詢?nèi)掌冢J(rèn)當(dāng)天）
        :return: UV 數(shù)量
        """
        if date is None:
            date = datetime.now()
        date_key = date.strftime("%Y-%m-%d")
        key = f"uv:daily:{date_key}"
        return self.redis.pfcount(key)
    def get_date_range_uv(self, start_date: datetime, end_date: datetime) -> int:
        """
        獲取日期范圍內(nèi)的 UV（并集統(tǒng)計(jì)）
        :param start_date: 開(kāi)始日期
        :param end_date: 結(jié)束日期
        :return: 去重后的 UV 數(shù)量
        """
        keys = []
        current_date = start_date
        while current_date <= end_date:
            key = f"uv:daily:{current_date.strftime('%Y-%m-%d')}"
            keys.append(key)
            current_date += timedelta(days=1)
        if not keys:
            return 0
        # 使用 PFCOUNT 計(jì)算并集基數(shù)
        return self.redis.pfcount(*keys)
    def merge_monthly_uv(self, year: int, month: int):
        """
        合并月度 UV 統(tǒng)計(jì)
        :param year: 年份
        :param month: 月份
        """
        # 生成該月所有日期的 Key
        keys = []
        date = datetime(year, month, 1)
        while date.month == month:
            key = f"uv:daily:{date.strftime('%Y-%m-%d')}"
            keys.append(key)
            date += timedelta(days=1)
        # 合并到月度 Key
        monthly_key = f"uv:monthly:{year}-{month:02d}"
        self.redis.pfmerge(monthly_key, *keys)
        # 設(shè)置過(guò)期時(shí)間（保留 2 年）
        self.redis.expire(monthly_key, 2 * 365 * 24 * 60 * 60)

from fastapi import APIRouter, Depends, HTTPException
from datetime import datetime, timedelta
from pydantic import BaseModel
router = APIRouter()
class UVStatsResponse(BaseModel):
    daily_uv: int
    weekly_uv: int
    monthly_uv: int
@router.get("/stats/uv", response_model=UVStatsResponse)
async def get_uv_statistics(
    date: str = None,
    uv_service: UVStatisticsService = Depends(lambda: UVStatisticsService(redis_client))
):
    """
    獲取 UV 統(tǒng)計(jì)數(shù)據(jù)
    :param date: 查詢?nèi)掌冢╕YYY-MM-DD 格式，默認(rèn)當(dāng)天）
    :return: 日、周、月 UV 統(tǒng)計(jì)
    """
    if date:
        try:
            query_date = datetime.strptime(date, "%Y-%m-%d")
        except ValueError:
            raise HTTPException(status_code=400, detail="Invalid date format")
    else:
        query_date = datetime.now()
    # 日 UV
    daily_uv = uv_service.get_daily_uv(query_date)
    # 周 UV（最近 7 天）
    week_start = query_date - timedelta(days=6)
    weekly_uv = uv_service.get_date_range_uv(week_start, query_date)
    # 月 UV（最近 30 天）
    month_start = query_date - timedelta(days=29)
    monthly_uv = uv_service.get_date_range_uv(month_start, query_date)
    return UVStatsResponse(
        daily_uv=daily_uv,
        weekly_uv=weekly_uv,
        monthly_uv=monthly_uv
    )
@router.post("/visit/record")
async def record_user_visit(
    user_id: int,
    uv_service: UVStatisticsService = Depends(lambda: UVStatisticsService(redis_client))
):
    """
    記錄用戶訪問(wèn)
    :param user_id: 用戶 ID
    """
    uv_service.record_visit(user_id)
    return {"status": "success", "message": "Visit recorded"}