ap_scanner/exporter.py (4845B)
1 """CSV and JSON export for scan results."""
2
3 import csv
4 import json
5 from datetime import datetime
6 from pathlib import Path
7
8 from ap_scanner.models import AccessPoint
9
10
11 def generate_filename(ext: str = "csv") -> str:
12 """Generate a default filename with timestamp."""
13 ts = datetime.now().strftime("%Y%m%d_%H%M%S")
14 return f"ap_scan_{ts}.{ext}"
15
16
17 def save_csv(aps: dict[str, AccessPoint], filepath: str | None = None) -> str:
18 """Save AP scan results to CSV file in wifi-inspector compatible format.
19
20 Returns the filepath that was written.
21 """
22 if filepath is None:
23 filepath = generate_filename()
24
25 path = Path(filepath)
26 path.parent.mkdir(parents=True, exist_ok=True)
27
28 fieldnames = [
29 "SSID", "MAC Address", "RSSI",
30 "Channel", "Band", "Width", "802.11",
31 "WEP", "WPA", "WPA2", "WPA3", "WPS",
32 "Vendor", "Beacon Count",
33 "First seen", "Last seen",
34 "Active Scan IP", "Active Scan Gateway",
35 "Active Scan Clients", "Active Scan Ports",
36 ]
37
38 sorted_aps = sorted(
39 aps.values(),
40 key=lambda ap: (ap.rssi if ap.rssi is not None else -999),
41 reverse=True,
42 )
43
44 with open(path, "w", newline="", encoding="utf-8") as f:
45 writer = csv.DictWriter(f, fieldnames=fieldnames, delimiter=";",
46 quoting=csv.QUOTE_ALL)
47 writer.writeheader()
48 for ap in sorted_aps:
49 asr = ap.active_scan_result
50 writer.writerow({
51 "SSID": ap.display_ssid,
52 "MAC Address": ap.bssid,
53 "RSSI": str(ap.rssi) if ap.rssi is not None else "",
54 "Channel": ap.channel_display,
55 "Band": ap.band,
56 "Width": str(ap.width),
57 "802.11": ap.wifi_standard,
58 "WEP": ap.wep,
59 "WPA": ap.wpa,
60 "WPA2": ap.wpa2,
61 "WPA3": ap.wpa3,
62 "WPS": ap.wps,
63 "Vendor": ap.vendor,
64 "Beacon Count": str(ap.beacon_count),
65 "First seen": ap.first_seen.strftime("%Y-%m-%d %H:%M:%S"),
66 "Last seen": ap.last_seen.strftime("%Y-%m-%d %H:%M:%S"),
67 "Active Scan IP": asr.assigned_ip if asr else "",
68 "Active Scan Gateway": asr.gateway_ip if asr else "",
69 "Active Scan Clients": str(asr.client_count) if asr else "",
70 "Active Scan Ports": ",".join(str(p) for p in asr.gateway_open_ports) if asr else "",
71 })
72
73 return str(path)
74
75
76 def save_json(aps: dict[str, AccessPoint], filepath: str | None = None) -> str:
77 """Save AP scan results to JSON file.
78
79 Returns the filepath that was written.
80 """
81 if filepath is None:
82 filepath = generate_filename("json")
83
84 path = Path(filepath)
85 path.parent.mkdir(parents=True, exist_ok=True)
86
87 sorted_aps = sorted(
88 aps.values(),
89 key=lambda ap: (ap.rssi if ap.rssi is not None else -999),
90 reverse=True,
91 )
92
93 records = []
94 for ap in sorted_aps:
95 record = {
96 "ssid": ap.display_ssid,
97 "bssid": ap.bssid,
98 "rssi": ap.rssi,
99 "channel": ap.channel,
100 "channel_display": ap.channel_display,
101 "band": ap.band,
102 "width": ap.width,
103 "wifi_standard": ap.wifi_standard,
104 "encryption": ap.encryption,
105 "wep": ap.wep,
106 "wpa": ap.wpa,
107 "wpa2": ap.wpa2,
108 "wpa3": ap.wpa3,
109 "wps": ap.wps,
110 "vendor": ap.vendor,
111 "beacon_count": ap.beacon_count,
112 "first_seen": ap.first_seen.strftime("%Y-%m-%d %H:%M:%S"),
113 "last_seen": ap.last_seen.strftime("%Y-%m-%d %H:%M:%S"),
114 }
115 if ap.active_scan_result:
116 asr = ap.active_scan_result
117 record["active_scan"] = {
118 "scan_time": asr.scan_time.isoformat(),
119 "connected": asr.connected,
120 "assigned_ip": asr.assigned_ip,
121 "subnet_mask": asr.subnet_mask,
122 "gateway_ip": asr.gateway_ip,
123 "gateway_mac": asr.gateway_mac,
124 "dns_servers": asr.dns_servers,
125 "dhcp_server": asr.dhcp_server,
126 "gateway_open_ports": asr.gateway_open_ports,
127 "clients": [
128 {"ip": c.ip, "mac": c.mac, "vendor": c.vendor}
129 for c in asr.clients
130 ],
131 "duration_secs": asr.duration_secs,
132 "error": asr.error,
133 }
134 records.append(record)
135
136 with open(path, "w", encoding="utf-8") as f:
137 json.dump({"scan_time": datetime.now().isoformat(), "access_points": records},
138 f, ensure_ascii=False, indent=2)
139
140 return str(path)