80 lines
2.4 KiB
Python
Executable File
80 lines
2.4 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
import include_homekit
|
|
import homekit.telegram as telegram
|
|
|
|
from homekit.telegram.config import TelegramChatsConfig
|
|
from homekit.util import validate_mac_address
|
|
from typing import Optional
|
|
from homekit.config import config, AppConfigUnit
|
|
from homekit.database import BotsDatabase, SimpleState
|
|
|
|
|
|
class OpenwrtLogAnalyzerConfig(AppConfigUnit):
|
|
@classmethod
|
|
def schema(cls) -> Optional[dict]:
|
|
return {
|
|
'database_name': {'type': 'string', 'required': True},
|
|
'devices': {
|
|
'type': 'dict',
|
|
'keysrules': {'type': 'string'},
|
|
'valuesrules': {
|
|
'type': 'string',
|
|
'check_with': validate_mac_address
|
|
}
|
|
},
|
|
'limit': {'type': 'integer'},
|
|
'telegram_chat': {'type': 'string'},
|
|
'aps': {
|
|
'type': 'list',
|
|
'schema': {'type': 'integer'}
|
|
}
|
|
}
|
|
|
|
@staticmethod
|
|
def custom_validator(data):
|
|
chats = TelegramChatsConfig()
|
|
if data['telegram_chat'] not in chats:
|
|
return ValueError(f'unknown telegram chat {data["telegram_chat"]}')
|
|
|
|
|
|
def main(mac: str,
|
|
title: str,
|
|
ap: int) -> int:
|
|
db = BotsDatabase()
|
|
|
|
data = db.get_openwrt_logs(filter_text=mac,
|
|
min_id=state['last_id'],
|
|
access_point=ap,
|
|
limit=config['openwrt_log_analyzer']['limit'])
|
|
if not data:
|
|
return 0
|
|
|
|
max_id = 0
|
|
for log in data:
|
|
if log.id > max_id:
|
|
max_id = log.id
|
|
|
|
text = '\n'.join(map(lambda s: str(s), data))
|
|
telegram.send_message(f'<b>{title} (AP #{ap})</b>\n\n' + text, config.app_config['telegram_chat'])
|
|
|
|
return max_id
|
|
|
|
|
|
if __name__ == '__main__':
|
|
config.load_app(OpenwrtLogAnalyzerConfig)
|
|
for ap in config.app_config['aps']:
|
|
dbname = config.app_config['database_name']
|
|
dbname = dbname.replace('.txt', f'-{ap}.txt')
|
|
|
|
state = SimpleState(name=dbname,
|
|
default={'last_id': 0})
|
|
|
|
max_last_id = 0
|
|
for name, mac in config['devices'].items():
|
|
last_id = main(mac, title=name, ap=ap)
|
|
if last_id > max_last_id:
|
|
max_last_id = last_id
|
|
|
|
if max_last_id:
|
|
state['last_id'] = max_last_id
|