forked from Amanbhandula/SmokeDetector
-
Notifications
You must be signed in to change notification settings - Fork 0
/
blacklists.py
144 lines (111 loc) · 4.5 KB
/
blacklists.py
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
from typing import Union
import regex
from globalvars import GlobalVars
from helpers import log
def load_blacklists():
GlobalVars.bad_keywords = Blacklist(Blacklist.KEYWORDS).parse()
GlobalVars.blacklisted_websites = Blacklist(Blacklist.WEBSITES).parse()
GlobalVars.blacklisted_usernames = Blacklist(Blacklist.USERNAMES).parse()
GlobalVars.blacklisted_numbers = Blacklist(Blacklist.NUMBERS).parse()
GlobalVars.watched_keywords = Blacklist(Blacklist.WATCHED_KEYWORDS).parse()
GlobalVars.watched_numbers = Blacklist(Blacklist.WATCHED_NUMBERS).parse()
class BlacklistParser:
def __init__(self, filename):
self._filename = filename
def parse(self):
return None
def add(self, item):
pass
def remove(self, item):
pass
def exists(self, item):
pass
class BasicListParser(BlacklistParser):
def parse(self):
with open(self._filename, 'r', encoding='utf-8') as f:
return [line.rstrip() for line in f if len(line.rstrip()) > 0 and line[0] != '#']
def add(self, item: str):
with open(self._filename, 'a+', encoding='utf-8') as f:
last_char = f.read()[-1:]
if last_char not in ['', '\n']:
item = '\n' + item
f.write(item + '\n')
def remove(self, item: str):
with open(self._filename, 'r+', encoding='utf-8') as f:
items = f.readlines()
items = [x for x in items if item not in x]
f.seek(0)
f.truncate()
f.writelines(items)
def exists(self, item: str):
item = item.lower()
with open(self._filename, 'r', encoding='utf-8') as f:
lines = f.readlines()
for i, x in enumerate(lines, start=1):
if item == x.lower().rstrip('\n'):
return True, i
return False, -1
class TSVDictParser(BlacklistParser):
def parse(self):
dct = {}
with open(self._filename, 'r', encoding='utf-8') as f:
for lineno, line in enumerate(f, 1):
if regex.compile(r'^\s*(?:#|$)').match(line):
continue
try:
when, by_whom, what = line.rstrip().split('\t')
except ValueError as err:
log('error', '{0}:{1}:{2}'.format(self._filename, lineno, err))
continue
if what[0] != "#":
dct[what] = {'when': when, 'by': by_whom}
return dct
def add(self, item: Union[str, dict]):
with open(self._filename, 'a+', encoding='utf-8') as f:
if isinstance(item, dict):
item = '{}\t{}\t{}'.format(item[0], item[1], item[2])
last_char = f.read()[-1:]
if last_char not in ['', '\n']:
item = '\n' + item
f.write(item + '\n')
def remove(self, item: Union[str, dict]):
if isinstance(item, dict):
item = item[2]
with open(self._filename, 'r+', encoding='utf-8') as f:
items = f.readlines()
items = [x for x in items if ('\t' not in x) or
(len(x.split('\t')) == 3 and x.split('\t')[2].strip() != item)]
f.seek(0)
f.truncate()
f.writelines(items)
def exists(self, item: Union[str, dict]):
if isinstance(item, dict):
item = item[2]
item = item.split('\t')[-1]
with open(self._filename, 'r', encoding='utf-8') as f:
lines = f.readlines()
for i, x in enumerate(lines, start=1):
if '\t' not in x:
continue
splat = x.split('\t')
if len(splat) == 3 and splat[2].strip() == item:
return True, i
return False, -1
class Blacklist:
KEYWORDS = ('bad_keywords.txt', BasicListParser)
WEBSITES = ('blacklisted_websites.txt', BasicListParser)
USERNAMES = ('blacklisted_usernames.txt', BasicListParser)
NUMBERS = ('blacklisted_numbers.txt', BasicListParser)
WATCHED_KEYWORDS = ('watched_keywords.txt', TSVDictParser)
WATCHED_NUMBERS = ('watched_numbers.txt', TSVDictParser)
def __init__(self, type):
self._filename = type[0]
self._parser = type[1](self._filename)
def parse(self):
return self._parser.parse()
def add(self, item):
return self._parser.add(item)
def remove(self, item):
return self._parser.remove(item)
def exists(self, item):
return self._parser.exists(item)