-
Notifications
You must be signed in to change notification settings - Fork 2
/
Copy pathaiolog.py
158 lines (137 loc) · 5.17 KB
/
aiolog.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
145
146
147
148
149
150
151
152
153
154
155
156
157
158
import io
import re
import asyncio
class AioStream(io.StringIO):
def __init__(self, alloc_size):
super().__init__(alloc_size)
self._max_size = alloc_size
self._write = super().write
self._tmp = io.StringIO(100)
self._lw = 0
self._comp = False
def write(self, sdata):
if sdata.endswith("\n"):
if self.tell() + len(sdata) + self._tmp.tell() >= self._max_size:
# clean
self._write(" " * ((self._max_size - self.tell()) - 1))
self._write("\n")
# rotate
self.seek(0)
self._comp = True
self._tmp.seek(0)
self._write(self._tmp.read(self._lw))
self._tmp.seek(0)
self._write(sdata)
else:
self._lw = self._tmp.write(sdata)
def cat(self, grep=""):
index = self.tell()
if self._comp:
self.readline()
if grep:
for line in self:
if (
line
and ("*" in grep or isinstance(grep, list))
and self._grep(grep, line)
):
print(line, end="")
elif isinstance(grep, str):
if grep in line:
print(line, end="")
else:
for line in self:
if line.strip():
print(line, end="")
self.seek(0)
# read and grep for regex
if grep:
for line in self:
if (
line
and ("*" in grep or isinstance(grep, list))
and self._grep(grep, line)
):
print(line, end="")
elif isinstance(grep, str):
if grep in line:
print(line, end="")
if self.tell() >= index:
self.seek(index)
return
else:
for line in self:
if line.strip():
print(line, end="")
if self.tell() >= index:
self.seek(index)
return
self.seek(index)
def _grep(self, patt, line):
if isinstance(patt, list):
pass
else:
patt = [patt]
_pattlst = (
re.compile(_patt.replace(".", r"\.").replace("*", ".*") + "$")
for _patt in patt
)
try:
return any(_pattrn.match(line) for _pattrn in _pattlst)
except Exception:
return None
async def follow(self, grep="", wait=0.05):
init_index = self.tell()
while True:
try:
current_index = self.tell()
if current_index != init_index:
self.seek(init_index)
if grep:
if "*" not in grep:
for line in self:
if line and grep in line:
print(line, end="")
if self.tell() == current_index:
break
if current_index < init_index:
self.seek(0)
init_index = 0
for line in self:
if line and grep in line:
print(line, end="")
if self.tell() == current_index:
break
else:
for line in self:
if line and self._grep(grep, line):
print(line, end="")
if self.tell() == current_index:
break
if current_index < init_index:
self.seek(0)
init_index = 0
for line in self:
if line and self._grep(grep, line):
print(line, end="")
if self.tell() == current_index:
break
else:
for line in self:
if line.strip():
print(line, end="")
if self.tell() == current_index:
break
if current_index < init_index:
self.seek(0)
init_index = 0
for line in self:
if line.strip():
print(line, end="")
if self.tell() == current_index:
break
init_index = current_index
await asyncio.sleep(wait)
except KeyboardInterrupt:
break
streamlog = AioStream(2000)