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
|
import pytz
from .mysql import mysql_now, MySQLDatabase, datetime_fmt
from ..api.types import (
BotType,
SoundSensorLocation
)
from typing import Optional
from datetime import datetime
from html import escape
class OpenwrtLogRecord:
id: int
log_time: datetime
received_time: datetime
text: str
def __init__(self, id, text, log_time, received_time):
self.id = id
self.text = text
self.log_time = log_time
self.received_time = received_time
def __repr__(self):
return f"<b>{self.log_time.strftime('%H:%M:%S')}</b> {escape(self.text)}"
class BotsDatabase(MySQLDatabase):
def add_request(self,
bot: BotType,
user_id: int,
message: str):
with self.cursor() as cursor:
cursor.execute("INSERT INTO requests_log (user_id, message, bot, time) VALUES (%s, %s, %s, %s)",
(user_id, message, bot.name.lower(), mysql_now()))
self.commit()
def add_openwrt_logs(self,
lines: list[tuple[datetime, str]]):
now = datetime.now()
with self.cursor() as cursor:
for line in lines:
time, text = line
cursor.execute("INSERT INTO openwrt (log_time, received_time, text) VALUES (%s, %s, %s)",
(time.strftime(datetime_fmt), now.strftime(datetime_fmt), text))
self.commit()
def add_sound_hits(self,
hits: list[tuple[SoundSensorLocation, int]],
time: datetime):
with self.cursor() as cursor:
for loc, count in hits:
cursor.execute("INSERT INTO sound_hits (location, `time`, hits) VALUES (%s, %s, %s)",
(loc.name.lower(), time.strftime(datetime_fmt), count))
self.commit()
def get_sound_hits(self,
location: SoundSensorLocation,
after: Optional[datetime] = None,
last: Optional[int] = None) -> list[dict]:
with self.cursor(dictionary=True) as cursor:
sql = "SELECT `time`, hits FROM sound_hits WHERE location=%s"
args = [location.name.lower()]
if after:
sql += ' AND `time` >= %s ORDER BY time DESC'
args.append(after)
elif last:
sql += ' ORDER BY time DESC LIMIT 0, %s'
args.append(last)
else:
raise ValueError('no `after`, no `last`, what do you expect?')
cursor.execute(sql, tuple(args))
data = []
for row in cursor.fetchall():
data.append({
'time': row['time'],
'hits': row['hits']
})
return data
def get_openwrt_logs(self,
filter_text: str,
min_id: int,
limit: int = None) -> list[OpenwrtLogRecord]:
tz = pytz.timezone('Europe/Moscow')
with self.cursor(dictionary=True) as cursor:
sql = "SELECT * FROM openwrt WHERE text LIKE %s AND id > %s"
if limit is not None:
sql += f" LIMIT {limit}"
cursor.execute(sql, (f'%{filter_text}%', min_id))
data = []
for row in cursor.fetchall():
data.append(OpenwrtLogRecord(
id=int(row['id']),
text=row['text'],
log_time=row['log_time'].astimezone(tz),
received_time=row['received_time'].astimezone(tz)
))
return data
|
