add history (last human entry)

This commit is contained in:
Tomáš Mládek 2021-02-10 23:21:05 +01:00
parent 5b9c4827aa
commit 26bda6277c
2 changed files with 30 additions and 6 deletions

View file

@ -4,7 +4,7 @@ import os
import re import re
import sqlite3 import sqlite3
from dataclasses import dataclass from dataclasses import dataclass
from datetime import datetime from datetime import datetime, timedelta
from time import sleep from time import sleep
from typing import List, Optional from typing import List, Optional
@ -34,8 +34,19 @@ def _get_db(filepath: str):
return connection return connection
def _fetch_leases(db, from_ts: datetime):
c = db.cursor()
query = c.execute(
'SELECT ts, mac, hostname, ip FROM spottings WHERE ts > ? ORDER BY ts DESC',
(int(from_ts.timestamp()),)
)
for row in query:
yield Lease(datetime.fromtimestamp(row[0]), row[1], row[2], row[3])
@dataclass @dataclass
class Lease: class Lease:
ts: datetime
mac: str mac: str
hostname: Optional[str] hostname: Optional[str]
ip: str ip: str
@ -78,6 +89,9 @@ def run_forever(address: str, period: int, ssid: str, output: str):
dhcp_leases = api.get_resource('/ip/dhcp-server/lease').call('print') dhcp_leases = api.get_resource('/ip/dhcp-server/lease').call('print')
logging.debug(f"Got {len(dhcp_leases)} DHCP leases.") logging.debug(f"Got {len(dhcp_leases)} DHCP leases.")
now = datetime.now()
timestamp = int(now.timestamp())
registered_leases: List[Lease] = [] registered_leases: List[Lease] = []
for client in filter(lambda c: ssid in c['ssid'], currently_registered): for client in filter(lambda c: ssid in c['ssid'], currently_registered):
try: try:
@ -87,7 +101,8 @@ def run_forever(address: str, period: int, ssid: str, output: str):
except StopIteration: except StopIteration:
continue continue
registered_leases.append( registered_leases.append(
Lease(ip=lease['active-address'], mac=lease['active-mac-address'], hostname=lease.get('host-name')) Lease(ts=now, ip=lease['active-address'], mac=lease['active-mac-address'],
hostname=lease.get('host-name'))
) )
logging.info(f"Found {len(registered_leases)} registered leases.") logging.info(f"Found {len(registered_leases)} registered leases.")
logging.debug(", ".join([str(lease) for lease in registered_leases])) logging.debug(", ".join([str(lease) for lease in registered_leases]))
@ -101,9 +116,6 @@ def run_forever(address: str, period: int, ssid: str, output: str):
else: else:
status = Status(status="empty", text="There are no devices connected?") status = Status(status="empty", text="There are no devices connected?")
now = datetime.now()
timestamp = int(now.timestamp())
logging.debug("Logging into the database...") logging.debug("Logging into the database...")
cur = db.cursor() cur = db.cursor()
cur.executemany("INSERT INTO spottings (ts, mac, hostname, ip) VALUES (?,?,?,?)", [ cur.executemany("INSERT INTO spottings (ts, mac, hostname, ip) VALUES (?,?,?,?)", [
@ -126,12 +138,16 @@ def run_forever(address: str, period: int, ssid: str, output: str):
for lease in registered_leases: for lease in registered_leases:
writer.writerow((lease.ip, lease.mac, lease.hostname or "???")) writer.writerow((lease.ip, lease.mac, lease.hostname or "???"))
elif output_file.endswith(".html"): 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) for ch in config.computer_hostnames)), None)
logging.debug(f"Outputting HTML file into {output_file}...") logging.debug(f"Outputting HTML file into {output_file}...")
with open(output_file, 'w') as file: with open(output_file, 'w') as file:
out_str = jinja_env.get_template("index.html").render( out_str = jinja_env.get_template("index.html").render(
now=now, now=now,
leases=registered_leases, leases=registered_leases,
status=status status=status,
last_human=last_human
) )
file.write(out_str) file.write(out_str)

View file

@ -69,5 +69,13 @@
</tr> </tr>
{% endfor %} {% endfor %}
</table> </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> </body>
</html> </html>