python使用布隆過濾器的案例分析

from pybloom_live import ScalableBloomFilter,BloomFilter

# 可自動(dòng)伸縮的布隆過濾器
bloom = ScalableBloomFilter(initial_capacity=100,error_rate=0.001)

# 添加內(nèi)容
bloom.add('daqi')
print('daqi'in bloom)

# 定長的布隆過濾器
bloom1 = BloomFilter(capacity=10000)
bloom1.add('daqi')
print('daqi'in bloom1)

import random
import mmh4
from bitarray import bitarray
import os.path
import re


# bitarray長度
BIT_SIZE = 50000

class BloomFilter():

  def __init__(self):
    bit_array = bitarray(BIT_SIZE)
    bit_array.setall(0)
    self.bit_array = bit_array
    self.bit_size = self.length()


  def get_points(self, url):
    """
    生成需要插入的位置
    :param url:
    :return:節(jié)點(diǎn)的列表
    """
    point_list = []
    for i in range(7):
      point = mmh4.hash(url,30+i) % self.bit_size
      point_list.append(point)
    return point_list

  def add(self, url):
    """
    添加url到bitarray中
    :param url:
    :return:
    """
    res = self.bitarray_expand()
    points = self.get_points(url)
    try:
      for point in points:
        self.bit_array[point] = 1
      return '注冊完成！'
    except Exception as e:
      return e

  def contains(self,url):
    """
    驗(yàn)證url是否存在
    :param url:
    :return:True or False
    """
    points = self.get_points(url)
    # 在bitarray中查找對(duì)應(yīng)的點(diǎn)，如果有一個(gè)點(diǎn)值為0就說明該url不存在
    for p in points:
      if self.bit_array[p] == 0:
        return False
    return True


  def count(self):
    """
    獲取bitarrray中使用的節(jié)點(diǎn)數(shù)
    :return: bitarray長度
    """
    return self.bit_array.count()


  def length(self):
    """
    獲取bitarray的長度
    :return:bitarray的長度
    """
    return len(self.bit_array)


  def bitarray_expand(self):
    """
    擴(kuò)充bitarray長度
    :return:bitarray的長度或使用率，布隆過濾器的bitarray的使用最好不要超過50%,這樣誤判率低一些
    """
    isusespace = round(int(self.count()) / int(self.length()),4)
    if 0.50 < isusespace:
      # 新建bitarray
      expand_bitarray = bitarray(BIT_SIZE)
      expand_bitarray.setall(0)
      # 增加新建的bitarray
      self.bit_array = self.bit_array + expand_bitarray
      self.bit_size = self.length()
      return self.bit_size
    else:
      return f'長度尚可,{round(isusespace * 100,2)}%'



def get_captcha():
  """
  生成用于測試的隨機(jī)碼
  :return:
  """
  seed = '0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ'
  captcha = ""
  for i in range(10):
    captcha += random.choice(seed)
  print(captcha)
  return captcha


if __name__ == '__main__':
  bloom = BloomFilter()
  for i in range(100000):
    bloom.add(f'www.{get_captcha()}.com')
    print(bloom.length())
    print(bloom.count())
  print(bloom.count())