#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ HTML report for one capture session. Creates: reports/sessions/.html """ from __future__ import annotations import html import json from pathlib import Path import pynmea2 from reports.database import query_all, query_one from reports.sessions import ( session_statistics, trip_time, session_directory_size, coordinates_restored_count, ap_without_coordinates_count, ) from reports.utils import ( format_datetime as format_report_datetime, format_duration, format_size, ) from reports.template import ( page_begin, page_end, page_header, card, ReportContext, ) # ============================================================ # Helpers # ============================================================ def esc(value) -> str: if value is None: return "" return html.escape(str(value)) def format_datetime(value): return format_report_datetime( value ) def format_session_folder(name): try: from datetime import datetime, timezone dt = datetime.strptime( name, "%Y%m%d_%H%M%S" ).replace( tzinfo=timezone.utc ) return format_report_datetime( dt.isoformat() ) except Exception: return name # ============================================================ # Session information # ============================================================ def load_session( conn, session_id: int, ): session = session_statistics( conn, session_id ) if session is None: raise RuntimeError( f"Session {session_id} not found." ) trip_start, trip_end = trip_time( conn, session_id ) if trip_start: session["trip_start"] = trip_start if trip_end: session["trip_end"] = trip_end if trip_start and trip_end: session["duration"] = format_duration( trip_start, trip_end ) session["directory_size"] = format_size( session_directory_size( conn, session_id ) ) return session def load_trip_information( conn, session_id, ): row = query_one( conn, """ SELECT MIN(timestamp), MAX(timestamp), COUNT(*) FROM gps_track WHERE session_id=? """, ( session_id, ), ) if ( row is None or row[0] is None ): return { "trip_start": None, "trip_end": None, "trip_points": 0, } return { "trip_start": row[0], "trip_end": row[1], "trip_points": row[2], } # ============================================================ # GPS track # ============================================================ def load_gps_track( conn, session_id, ): rows = query_all( conn, """ SELECT latitude, longitude FROM gps_track WHERE session_id=? ORDER BY timestamp """, ( session_id, ), ) track = [] for row in rows: latitude = row[0] longitude = row[1] if ( latitude is None or longitude is None ): continue if ( latitude == 0 or longitude == 0 ): continue track.append( { "lat": latitude, "lon": longitude, } ) return track # ============================================================ # Session observations # ============================================================ def load_session_observations( conn, session_id, ): rows = query_all( conn, """ SELECT latitude, longitude, rssi, observed_at, id, bssid, essid, vendor, encryption, channel, frequency, has_handshake, has_pmkid, is_cracked FROM ( SELECT obs.latitude, obs.longitude, obs.rssi, obs.observed_at, ap.id, ap.bssid, ap.essid, ap.vendor, ap.encryption, ap.channel, ap.frequency, CASE WHEN hs.id IS NULL THEN 0 ELSE 1 END AS has_handshake, ap.has_pmkid, CASE WHEN EXISTS ( SELECT 1 FROM credentials AS c JOIN handshakes AS ch ON ch.id = c.handshake_id WHERE c.access_point_id = ap.id AND ch.session_id = obs.session_id ) THEN 1 ELSE 0 END AS is_cracked, ROW_NUMBER() OVER ( PARTITION BY obs.access_point_id ORDER BY obs.rssi DESC, obs.observed_at DESC ) AS rn FROM access_point_observations AS obs JOIN access_points AS ap ON ap.id = obs.access_point_id LEFT JOIN handshakes AS hs ON hs.access_point_id = ap.id AND hs.session_id = obs.session_id WHERE obs.session_id = ? ) WHERE rn = 1 ORDER BY has_handshake DESC, is_cracked DESC, rssi DESC """, ( session_id, ), ) observations = [] for row in rows: observations.append( { "latitude": row[0], "longitude": row[1], "rssi": row[2], "observed_at": row[3], "id": row[4], "bssid": row[5], "essid": row[6], "vendor": row[7], "encryption": row[8], "channel": row[9], "frequency": row[10], "handshake": bool(row[11]), "pmkid": bool(row[12]), "cracked": bool(row[13]), } ) return observations # ============================================================ # Summary # ============================================================ def summary_block( conn, session, observations, ): access_points = len(observations) handshake_ap = sum( 1 for ap in observations if ap["handshake"] ) handshake_records = session.get( "handshake_records", handshake_ap ) pmkid_ap = sum( 1 for ap in observations if ap["pmkid"] ) cracked_ap = sum( 1 for ap in observations if ap["cracked"] ) coordinates_restored = session.get( "coordinates_restored", coordinates_restored_count( conn, session["id"] ) ) ap_without_coordinates = session.get( "ap_without_coordinates", ap_without_coordinates_count( conn, session["id"] ) ) html = f"""
{access_points}
Access Points
{coordinates_restored}
Coordinates Restored
{ap_without_coordinates}
AP without Coordinates
{handshake_ap}
Handshake AP
{handshake_records}
Handshake Records
{pmkid_ap}
PMKID AP
{cracked_ap}
Cracked AP
{session.get("directory_size","-")}
Directory size
""" return card( "📊 Session Summary", html, ) # ============================================================ # Files # ============================================================ def files_block( session, ): capture = Path( session["capture_file"] ) directory = capture.parent if not directory.exists(): html = """

📁 Каталог файлов этой поездки отсутствует.

Возможно, он был удалён пользователем вручную. WiFi GPS Mapper не удаляет каталоги поездок автоматически. Все данные сохранены в базе данных

""" else: html = """ """ files = [ "capture.pcapng", "gps_track.nmea", "access_points.csv", "handshakes.22000", "essid.txt", "parser.log", ] for name in files: filename = directory / name html += f""" """ html += """
File Exists
{name} {"✅" if filename.exists() else "—"}
""" return card( "📂 Session Files", html, ) # ============================================================ # JavaScript # ============================================================ def build_track_js( track, ): if not track: return "const track = [];" return ( "const track = " + json.dumps(track) + ";" ) def build_marker_js( observations, ): markers = [] for ap in observations: if ( ap["latitude"] is None or ap["longitude"] is None ): continue popup = f""" {esc(ap["essid"])}
{esc(ap["bssid"])}
Vendor: {esc(ap["vendor"])}
Encryption: {esc(ap["encryption"])}
Channel: {esc(ap["channel"])}    Frequency: {esc(ap["frequency"])}
RSSI: {esc(ap["rssi"])} dBm
Handshake: {"✅" if ap["handshake"] else "—"}    PMKID: {"✅" if ap["pmkid"] else "—"}    Password: {"✅" if ap["cracked"] else "—"} """ markers.append( { "lat": ap["latitude"], "lon": ap["longitude"], "popup": popup, "handshake": ap["handshake"], "pmkid": ap["pmkid"], "cracked": ap["cracked"], } ) return ( "const markers = " + json.dumps( markers, ensure_ascii=False, ) + ";" ) # ============================================================ # Map # ============================================================ def map_block( track, observations, context=None, ): track_data = [ [ float(p["lat"]), float(p["lon"]) ] for p in track if ( p.get("lat") is not None and p.get("lon") is not None and p.get("lat") != 0 and p.get("lon") != 0 ) ] marker_js = build_marker_js( observations ) maplibre_css_url = context.asset_url( "maplibre/maplibre-gl.css" ) maplibre_js_url = context.asset_url( "maplibre/maplibre-gl.js" ) html = f"""
""" title = """ 🗺 Capture Map """ return card( title, html, ) # ============================================================ # Access Points # ============================================================ def access_points_block( observations, ): html = """ """ handshake_points = [ ap for ap in observations if ap["handshake"] ] for ap in handshake_points: html += f""" """ html += """
BSSID ESSID Vendor Encryption Channel RSSI HS PMKID PASS
{esc(ap["bssid"])} {esc(ap["essid"])} {esc(ap["vendor"])} {esc(ap["encryption"])} {esc(ap["channel"])} {esc(ap["rssi"])} 🤝 {"🟠" if ap["pmkid"] else ""} {"🔑" if ap["cracked"] else ""}
""" return card( f"📡 Access Points Handshake ({len(handshake_points)})", html, ) # ============================================================ # Credentials # ============================================================ def credentials_block( conn, session_id, ): rows = query_all( conn, """ SELECT ap.bssid, ap.essid, c.password, c.source, c.created_at, c.verified FROM credentials c JOIN access_points ap ON ap.id=c.access_point_id JOIN handshakes hs ON hs.id=c.handshake_id WHERE hs.session_id=? GROUP BY c.id ORDER BY c.created_at DESC """, ( session_id, ), ) if not rows: return "" html = """ """ for row in rows: ( bssid, essid, password, source, created, verified, ) = row html += f""" """ html += """
BSSID ESSID Password Source Verified
{esc(bssid)} {esc(essid)} {esc(password)} {esc(source)} {"✅" if verified else ""}
""" return card( f"🔑 Credentials ({len(rows)})", html, ) # ============================================================ # Handshakes # ============================================================ def handshakes_block( conn, session_id, ): rows = query_all( conn, """ SELECT ap.bssid, ap.essid, h.type, h.message_pair, h.captured_at FROM handshakes h JOIN access_points ap ON ap.id=h.access_point_id JOIN access_point_observations obs ON obs.access_point_id=ap.id WHERE obs.session_id=? GROUP BY h.id ORDER BY h.captured_at DESC """, ( session_id, ), ) if not rows: return "" html = """ """ for row in rows: ( bssid, essid, hs_type, pair, captured, ) = row html += f""" """ html += """
BSSID ESSID Type Message Pair Captured
{esc(bssid)} {esc(essid)} {esc(hs_type)} {esc(pair)} {esc(format_datetime(captured))}
""" return card( f"🤝 Handshakes ({len(rows)})", html, ) # ============================================================ # Session page # ============================================================ def render_session_page( conn, session_id, context=None, ): if context is None: context = ReportContext() context = context.session_page() session = load_session( conn, session_id, ) trip = load_trip_information( conn, session_id, ) session.update( trip ) track = load_gps_track( conn, session_id, ) observations = load_session_observations( conn, session_id, ) html = page_begin( f"Session {format_session_folder(Path(session['capture_file']).parent.name)}", active="capture_sessions", context=context, ) html += page_header( f"🚗 Session {format_datetime(session.get('trip_start'))}", f""" Trip start: {esc(format_datetime(session.get("trip_start")))}
Trip end: {esc(format_datetime(session.get("trip_end")))}
Duration: {esc(session.get("duration", "-"))}
Database imported: {esc(format_datetime(session["start_time"]))}
Session folder: {esc(Path(session["capture_file"]).parent.name)} """ ) html += summary_block( conn, session, observations, ) html += files_block( session, ) html += map_block( track, observations, context, ) html += access_points_block( observations, ) # Handshakes block removed html += credentials_block( conn, session_id, ) html += page_end() return html def save_session_page( conn, session_id, context=None, ): html = render_session_page( conn, session_id, context, ) output_dir = ( Path(__file__).parent / "output" / "sessions" ) output_dir.mkdir( parents=True, exist_ok=True, ) filename = ( output_dir / f"{session_id}.html" ) filename.write_text( html, encoding="utf-8", ) return filename # ============================================================ # Build all pages # ============================================================ def build_all_session_pages( conn, context=None, ): rows = query_all( conn, """ SELECT id FROM capture_sessions ORDER BY start_time """, ) generated = [] for ( session_id, ) in rows: generated.append( save_session_page( conn, session_id, context, ) ) return generated