Track newly discovered MAC addresses and award points once per device.
(self, mac_addresses)
| 1296 | |
| 1297 | # Subnet scan log – ring buffer of recent scan events for the UI |
| 1298 | self._subnet_scan_log = [] # list of dicts |
| 1299 | self._subnet_scan_log_lock = threading.Lock() |
| 1300 | self._SUBNET_LOG_MAX = 50 # keep last 50 entries |
| 1301 | |
| 1302 | # Seed target/port counters from SQLite so the dashboard shows the |
| 1303 | # last-known numbers from the first /api/status response, instead of |
| 1304 | # 0 until the deferred sync_all_counts() walks every host. |
| 1305 | self._seed_counts_from_db() |
| 1306 | |
| 1307 | def _seed_counts_from_db(self): |
| 1308 | """Populate target/port counters from the active network's SQLite store.""" |
| 1309 | if get_db is None or self._pager_mode: |
| 1310 | return |
| 1311 | try: |
| 1312 | snapshot = get_db(currentdir=self.currentdir).get_count_snapshot() |
| 1313 | except Exception as exc: |
| 1314 | logger.debug(f"Skipping count seed (db not ready): {exc}") |
| 1315 | return |
| 1316 | self.targetnbr = snapshot.get('active_targets', 0) |
| 1317 | self.portnbr = snapshot.get('total_ports', 0) |
| 1318 | self.total_targetnbr = snapshot.get('total_targets', 0) |
| 1319 | self.inactive_targetnbr = snapshot.get('inactive_targets', 0) |
| 1320 | self.networkkbnbr = self.total_targetnbr |
| 1321 | logger.info( |
| 1322 | f"Seeded counters from DB: targets={self.targetnbr} " |
| 1323 | f"(total={self.total_targetnbr}, inactive={self.inactive_targetnbr}), " |
| 1324 | f"ports={self.portnbr}" |
| 1325 | ) |
| 1326 | |
| 1327 | def load_gamification_data(self): |
| 1328 | """Load persistent gamification progress from disk.""" |
| 1329 | os.makedirs(self.datadir, exist_ok=True) |
| 1330 | |
| 1331 | default_data = { |
| 1332 | "version": 1, |
| 1333 | "total_points": 0, |
| 1334 | "level": 1, |
| 1335 | "mac_points": {}, |
no test coverage detected