improve history, misc. fixes

This commit is contained in:
Tomáš Mládek 2021-02-11 01:09:40 +01:00
parent c8b3a8d98b
commit daedbadaff
2 changed files with 52 additions and 31 deletions

View file

@ -5,6 +5,8 @@ import re
import sqlite3
from dataclasses import dataclass
from datetime import datetime, timedelta
from itertools import groupby
from operator import attrgetter
from time import sleep
from typing import List, Optional
@ -14,6 +16,26 @@ from jinja2 import Environment, select_autoescape, FileSystemLoader
from config import Config
config = Config
@dataclass
class Lease:
ts: datetime
mac: str
hostname: Optional[str]
ip: str
@property
def display(self):
return self.hostname or self.mac
@dataclass
class Status:
open: bool
text: str
def _get_db(filepath: str):
logging.debug(f"Opening database: {filepath}")
@ -44,18 +66,11 @@ def _fetch_leases(db, from_ts: datetime):
yield Lease(datetime.fromtimestamp(row[0]), row[1], row[2], row[3])
@dataclass
class Lease:
ts: datetime
mac: str
hostname: Optional[str]
ip: str
@dataclass
class Status:
status: str
text: str
def _is_human(lease: Lease):
if lease.hostname:
return not any(re.match(ch, lease.hostname) for ch in config.computer_hostnames)
else:
return True
@click.command()
@ -65,7 +80,6 @@ class Status:
@click.option('-o', '--output', multiple=True, help="Output file.")
def run_forever(address: str, period: int, ssid: str, output: str):
logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - [%(levelname)s] %(message)s')
config = Config
db = _get_db("clients.sqlite3")
connection = routeros_api.RouterOsApiPool(
@ -104,17 +118,18 @@ def run_forever(address: str, period: int, ssid: str, output: str):
Lease(ts=now, ip=lease['active-address'], mac=lease['active-mac-address'],
hostname=lease.get('host-name'))
)
registered_leases.sort(key=lambda l: (l.hostname or "").lower())
registered_leases.sort(key=lambda l: not bool(l.hostname))
logging.info(f"Found {len(registered_leases)} registered leases.")
logging.debug(", ".join([str(lease) for lease in registered_leases]))
if len(registered_leases) > 0:
if len([lease for lease in registered_leases
if not any(re.match(ch, lease.hostname or "") for ch in config.computer_hostnames)]) > 0:
status = Status(status="populated", text="There seem to be people!")
if len([lease for lease in registered_leases if _is_human(lease)]) > 0:
status = Status(open=True, text="There seem to be people!")
else:
status = Status(status="empty", text="There are only computers.")
status = Status(open=False, text="There are only computers.")
else:
status = Status(status="empty", text="There are no devices connected?")
status = Status(open=False, text="There are no devices connected?")
logging.debug("Logging into the database...")
cur = db.cursor()
@ -138,9 +153,12 @@ def run_forever(address: str, period: int, ssid: str, output: str):
for lease in registered_leases:
writer.writerow((lease.ip, lease.mac, lease.hostname or "???"))
elif output_file.endswith(".html"):
last_human = next((lease for lease in _fetch_leases(db, now - timedelta(hours=24))
if not any(re.match(ch, lease.hostname or "") for ch in config.computer_hostnames)),
None)
last_change = None
for ts, leases in groupby(_fetch_leases(db, now - timedelta(days=7)), key=attrgetter('ts')):
humans_present = [lease for lease in leases if _is_human(lease)]
if (len(humans_present) > 0) != status.open:
last_change = {'ts': ts, 'leases': humans_present}
break
logging.debug(f"Outputting HTML file into {output_file}...")
with open(output_file, 'w') as file:
@ -148,7 +166,7 @@ def run_forever(address: str, period: int, ssid: str, output: str):
now=now,
leases=registered_leases,
status=status,
last_human=last_human
last_change=last_change
)
file.write(out_str)

View file

@ -50,9 +50,20 @@
<h1>/|\ Anabasis Clients</h1>
<h2>STATUS</h2>
<div class="status-container">
<div class="status status-{{status.status}}">{{status.status.upper()}}</div>
<div class="status status-{{'occupied' if status.open else 'empty'}}">
{{'OCCUPIED' if status.open else 'EMPTY'}}
</div>
<div class="status-explanation">{{status.text}}</div>
</div>
<div>
Since:
{% if last_change %}
{{last_change['ts'].strftime("%c")}}
{% if not status.open %}({{last_change.leases | map(attribute='display') | join(', ')}}){% endif %}
{% else %}
forever?
{% endif %}
</div>
<h2>Current clients</h2>
<div class="datetime">{{now.strftime("%c")}}</div>
<table>
@ -69,13 +80,5 @@
</tr>
{% endfor %}
</table>
<h2>History</h2>
<p>Last human entry:
{% if last_human %}
{{last_human.ts.strftime("%c")}} ({{last_human.hostname or last_human.mac}})
{% else %}
???
{% endif %}
</p>
</body>
</html>