Reads the source CSV file into a DataFrame (or list of dicts if pandas unavailable).
(self)
| 1409 | self.all_known_hosts_count = 0 |
| 1410 | |
| 1411 | def read_csv(self): |
| 1412 | """ |
| 1413 | Reads the source CSV file into a DataFrame (or list of dicts if pandas unavailable). |
| 1414 | """ |
| 1415 | if pd is None: |
| 1416 | # Fallback: use csv module |
| 1417 | self._rows = [] |
| 1418 | try: |
| 1419 | if not os.path.exists(self.source_csv_path) or os.path.getsize(self.source_csv_path) == 0: |
| 1420 | return |
| 1421 | with open(self.source_csv_path, 'r') as f: |
| 1422 | reader = csv.DictReader(f) |
| 1423 | self._rows = list(reader) |
| 1424 | self.logger.debug(f"Read {len(self._rows)} rows from {self.source_csv_path} (csv fallback)") |
| 1425 | except Exception as e: |
| 1426 | self.logger.error(f"Error reading CSV (fallback): {e}") |
| 1427 | return |
| 1428 | try: |
| 1429 | if not os.path.exists(self.source_csv_path): |
| 1430 | self.logger.warning(f"Source CSV file does not exist: {self.source_csv_path}") |
| 1431 | # Create an empty DataFrame with expected columns |
| 1432 | self.df = pd.DataFrame(columns=['MAC Address', 'IPs', 'Hostnames', 'Ports', 'Alive']) |
| 1433 | return |
| 1434 | |
| 1435 | # Check if file is empty |
| 1436 | if os.path.getsize(self.source_csv_path) == 0: |
| 1437 | self.logger.warning(f"Source CSV file is empty: {self.source_csv_path}") |
| 1438 | self.df = pd.DataFrame(columns=['MAC Address', 'IPs', 'Hostnames', 'Ports', 'Alive']) |
| 1439 | return |
| 1440 | |
| 1441 | # Try to read the CSV, catching specific pandas errors |
| 1442 | try: |
| 1443 | self.df = pd.read_csv(self.source_csv_path) |
| 1444 | except pd.errors.EmptyDataError: |
| 1445 | self.logger.warning(f"Source CSV file has no data to parse: {self.source_csv_path}") |
| 1446 | self.df = pd.DataFrame(columns=['MAC Address', 'IPs', 'Hostnames', 'Ports', 'Alive']) |
| 1447 | return |
| 1448 | except Exception as read_error: |
| 1449 | # Catch any other CSV reading errors (e.g., "No columns to parse from file") |
| 1450 | self.logger.warning(f"Could not parse CSV file: {read_error}") |
| 1451 | self.df = pd.DataFrame(columns=['MAC Address', 'IPs', 'Hostnames', 'Ports', 'Alive']) |
| 1452 | return |
| 1453 | |
| 1454 | # Check if DataFrame is empty or missing required columns |
| 1455 | if self.df.empty: |
| 1456 | self.logger.warning(f"Source CSV file has no data: {self.source_csv_path}") |
| 1457 | self.df = pd.DataFrame(columns=['MAC Address', 'IPs', 'Hostnames', 'Ports', 'Alive']) |
| 1458 | return |
| 1459 | |
| 1460 | # Ensure required columns exist |
| 1461 | required_columns = ['MAC Address', 'IPs', 'Hostnames', 'Ports', 'Alive'] |
| 1462 | missing_columns = [col for col in required_columns if col not in self.df.columns] |
| 1463 | if missing_columns: |
| 1464 | self.logger.warning(f"Missing columns in CSV: {missing_columns}") |
| 1465 | for col in missing_columns: |
| 1466 | self.df[col] = '' if col != 'Alive' else '0' |
| 1467 | |
| 1468 | self.logger.debug(f"Successfully read {len(self.df)} rows from {self.source_csv_path}") |
no test coverage detected