95 lines
4.5 KiB
Python
95 lines
4.5 KiB
Python
"""Deterministic, network-free discovery source contracts and adapters."""
|
|
from __future__ import annotations
|
|
from dataclasses import dataclass, field
|
|
from typing import Any, Mapping, Protocol, Sequence
|
|
import csv, io, re
|
|
|
|
SECRET_KEYS = {"password", "passwd", "secret", "token", "api_key", "apikey", "authorization", "private_key", "credential"}
|
|
|
|
def contains_secret(value: Any, path: str = "") -> str | None:
|
|
if isinstance(value, Mapping):
|
|
for key, child in value.items():
|
|
key_text = str(key).lower()
|
|
if key_text in SECRET_KEYS or any(x in key_text for x in ("password", "token", "secret", "api_key", "private_key")):
|
|
return path + str(key)
|
|
found = contains_secret(child, path + str(key) + ".")
|
|
if found: return found
|
|
elif isinstance(value, Sequence) and not isinstance(value, (str, bytes, bytearray)):
|
|
for index, child in enumerate(value):
|
|
found = contains_secret(child, path + str(index) + ".")
|
|
if found: return found
|
|
return None
|
|
|
|
@dataclass(frozen=True)
|
|
class ValidationResult:
|
|
valid: bool
|
|
errors: list[str] = field(default_factory=list)
|
|
|
|
@dataclass(frozen=True)
|
|
class DiscoveryPage:
|
|
records: list[dict[str, str]]
|
|
next_cursor: str | None = None
|
|
metadata: dict[str, Any] = field(default_factory=dict)
|
|
|
|
@dataclass(frozen=True)
|
|
class SourceHealth:
|
|
status: str = "unknown"
|
|
consecutive_failures: int = 0
|
|
circuit_open: bool = False
|
|
last_error: str | None = None
|
|
|
|
class DiscoverySource(Protocol):
|
|
kind: str
|
|
def validate(self, config: Mapping[str, Any]) -> ValidationResult: ...
|
|
def discover(self, config: Mapping[str, Any], cursor: str | None = None) -> DiscoveryPage: ...
|
|
|
|
_FIELDS = ("name", "website", "email", "phone", "description")
|
|
|
|
def normalize_record(row: Mapping[str, Any]) -> dict[str, str]:
|
|
result = {field: str(row.get(field, "")).strip() for field in _FIELDS}
|
|
# Accept common CSV spellings without retaining arbitrary sensitive fields.
|
|
aliases = {"company": "name", "url": "website", "domain": "website"}
|
|
for key, target in aliases.items():
|
|
if not result[target] and row.get(key) is not None: result[target] = str(row[key]).strip()
|
|
return result
|
|
|
|
class ManualSource:
|
|
kind = "manual"
|
|
def validate(self, config: Mapping[str, Any]) -> ValidationResult:
|
|
if not isinstance(config, Mapping): return ValidationResult(False, ["config must be an object"])
|
|
found = contains_secret(config)
|
|
if found: return ValidationResult(False, [f"secret field is not permitted: {found}"])
|
|
rows = config.get("rows")
|
|
if not isinstance(rows, list): return ValidationResult(False, ["rows must be a list"])
|
|
return ValidationResult(True)
|
|
def discover(self, config: Mapping[str, Any], cursor: str | None = None) -> DiscoveryPage:
|
|
validation = self.validate(config)
|
|
if not validation.valid: raise ValueError(validation.errors[0])
|
|
rows = [normalize_record(row) for row in config["rows"] if isinstance(row, Mapping)]
|
|
return DiscoveryPage(rows, None, {"adapter": self.kind})
|
|
|
|
class CsvSource:
|
|
kind = "csv"
|
|
def validate(self, config: Mapping[str, Any]) -> ValidationResult:
|
|
if not isinstance(config, Mapping): return ValidationResult(False, ["config must be an object"])
|
|
found = contains_secret(config)
|
|
if found: return ValidationResult(False, [f"secret field is not permitted: {found}"])
|
|
if not isinstance(config.get("csv"), str): return ValidationResult(False, ["csv must be text"])
|
|
try:
|
|
reader = csv.DictReader(io.StringIO(config["csv"]));
|
|
if not reader.fieldnames: return ValidationResult(False, ["CSV header is required"])
|
|
except csv.Error as exc: return ValidationResult(False, [f"invalid CSV: {exc}"])
|
|
return ValidationResult(True)
|
|
def discover(self, config: Mapping[str, Any], cursor: str | None = None) -> DiscoveryPage:
|
|
validation = self.validate(config)
|
|
if not validation.valid: raise ValueError(validation.errors[0])
|
|
reader = csv.DictReader(io.StringIO(config["csv"].replace("\r\n", "\n")))
|
|
records = [normalize_record({str(k).strip().lower(): v for k, v in row.items()}) for row in reader]
|
|
return DiscoveryPage(records, None, {"adapter": self.kind, "columns": reader.fieldnames or []})
|
|
|
|
ADAPTERS = {"manual": ManualSource, "csv": CsvSource}
|
|
|
|
def adapter_for(kind: str) -> DiscoverySource:
|
|
try: return ADAPTERS[kind]()
|
|
except KeyError: raise ValueError("unsupported source kind")
|