forked from Boris-code/feapder
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathbitarray.py
More file actions
154 lines (122 loc) · 4.41 KB
/
bitarray.py
File metadata and controls
154 lines (122 loc) · 4.41 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
# -*- coding: utf-8 -*-
"""
Created on 2018/12/14 1:05 PM
---------
@summary:
---------
@author: Boris
@email: boris_liu@foxmail.com
"""
from __future__ import absolute_import
from feapder.db.redisdb import RedisDB
class BitArray:
def setall(self, value):
pass
def __repr__(self):
raise ImportError("this method mush be implement")
def set(self, offsets, values):
"""
设置字符串数字某一位的值， 返回之前的值
@param offsets: 支持列表或单个值
@param values: 支持列表或单个值
@return: list / 单个值
"""
raise ImportError("this method mush be implement")
def get(self, offsets):
"""
取字符串数字某一位的值
@param offsets: 支持列表或单个值
@return: list / 单个值
"""
raise ImportError("this method mush be implement")
def count(self, value=True):
raise ImportError("this method mush be implement")
class MemoryBitArray(BitArray):
def __init__(self, num_bits):
try:
import bitarray
except Exception as e:
raise Exception(
'需要安装feapder完整版\ncommand: pip install "feapder[all]"\n若安装出错，参考：https://feapder.com/#/question/%E5%AE%89%E8%A3%85%E9%97%AE%E9%A2%98'
)
self.num_bits = num_bits
self.bitarray = bitarray.bitarray(num_bits, endian="little")
self.setall(0)
def __repr__(self):
return "MemoryBitArray: {}".format(self.num_bits)
def setall(self, value):
self.bitarray.setall(value)
def set(self, offsets, values):
"""
设置字符串数字某一位的值， 返回之前的值
@param offsets: 支持列表或单个值
@param values: 支持列表或单个值
@return: list / 单个值
"""
old_values = []
if isinstance(offsets, list):
if not isinstance(values, list):
values = [values] * len(offsets)
else:
assert len(offsets) == len(values), "offsets值要与values值一一对应"
for offset, value in zip(offsets, values):
old_values.append(int(self.bitarray[offset]))
self.bitarray[offset] = value
else:
old_values = int(self.bitarray[offsets])
self.bitarray[offsets] = values
return old_values
def get(self, offsets):
"""
取字符串数字某一位的值
@param offsets: 支持列表或单个值
@return: list / 单个值
"""
if isinstance(offsets, list):
return [self.bitarray[offset] for offset in offsets]
else:
return self.bitarray[offsets]
def count(self, value=True):
return self.bitarray.count(value)
class RedisBitArray(BitArray):
"""
仿bitarray 基于redis
"""
redis_db = None
def __init__(self, name, redis_url=None):
self.name = name
self.count_cached_name = name + "_count_cached"
if not self.__class__.redis_db:
self.__class__.redis_db = RedisDB(url=redis_url)
def __repr__(self):
return "RedisBitArray: {}".format(self.name)
def set(self, offsets, values):
"""
设置字符串数字某一位的值， 返回之前的值
@param offsets: 支持列表或单个值
@param values: 支持列表或单个值
@return: list / 单个值
"""
# 对offsets进行分片，最大100000个
results = []
batch_size = 170000
for i in range(0, len(offsets), batch_size):
results.extend(
self.redis_db.setbit(
self.name,
offsets[i : i + batch_size],
values[i : i + batch_size] if isinstance(values, list) else values,
)
)
return results
def get(self, offsets):
return self.redis_db.getbit(self.name, offsets)
def count(self, value=True):
# 先查redis的缓存，若没有 在统计数量
count = self.redis_db.strget(self.count_cached_name)
if count:
return int(count)
else:
count = self.redis_db.bitcount(self.name) # 被设置为 1 的比特位的数量
self.redis_db.strset(self.count_cached_name, count, ex=1800) # 半小时过期
return count