blob: 07d9953b0c63e733c3ff458387485c4de94727fd (
plain)
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
|
#!/usr/bin/env python3
import __py_include
import re
from html import escape
from typing import Optional
from homekit.config import AppConfigUnit, config
from homekit.modem import ModemsConfig, E3372
from homekit.database import MySQLHomeDatabase
from homekit.telegram import send_message
db: Optional[MySQLHomeDatabase] = None
class VkSmsCheckerConfig(AppConfigUnit):
NAME = 'vk_sms_checker'
@classmethod
def schema(cls) -> Optional[dict]:
return {
'modem': {'type': 'string', 'required': True}
}
@staticmethod
def custom_validator(data):
if data['modem'] not in ModemsConfig():
raise ValueError('invalid modem')
def get_last_time() -> int:
cur = db.cursor()
cur.execute("SELECT last_message_time FROM vk_sms LIMIT 1")
return int(cur.fetchone()[0])
def set_last_time(timestamp: int) -> None:
cur = db.cursor()
cur.execute("UPDATE vk_sms SET last_message_time=%s", (timestamp,))
db.commit()
def check_sms():
modem = ModemsConfig()[config.app_config['modem']]
cl = E3372(modem['ip'], legacy_token_auth=modem['legacy_auth'])
messages = cl.sms_list()
messages.reverse()
last_time = get_last_time()
new_last_time = None
results = []
if not messages:
return
for m in messages:
if m['UnixTime'] <= last_time:
continue
new_last_time = m['UnixTime']
if re.match(r'^vk', m['Phone'], flags=re.IGNORECASE) or re.match(r'vk', m['Content'], flags=re.IGNORECASE):
results.append(m)
if results:
for m in results:
text = '<b>'+escape(m['Phone'])+'</b> ('+m['Date']+')'
text += "\n"+escape(m['Content'])
send_message(text=text, chat='vk_sms_checker')
if new_last_time:
set_last_time(new_last_time)
if __name__ == '__main__':
db = MySQLHomeDatabase()
config.load_app(VkSmsCheckerConfig)
check_sms()
|
