diff --git a/apps/api/app/main.py b/apps/api/app/main.py index 798afbc..6f0b840 100644 --- a/apps/api/app/main.py +++ b/apps/api/app/main.py @@ -9,7 +9,7 @@ from urllib.request import Request, urlopen if __package__ in (None, ""): sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from app.domain import deduplication_key, deduplicate_businesses, is_suppressed, normalize_business, normalize_domain, normalize_phone, match_businesses - from app.sources import adapter_for, contains_secret, available_adapters, normalize_record, circuit_is_open + from app.sources import adapter_for, contains_secret, available_adapters, normalize_record, circuit_is_open, DISCOVERY_CRITERIA_FIELDS from app.domain_intelligence import normalize_registrable_domain, resolve_domain, generate_candidate_domains from app.website_scanner import scan_website, validate_url from app.contact_extractor import extract_contacts, MAX_HTML_BYTES, MAX_RESULTS @@ -22,7 +22,7 @@ if __package__ in (None, ""): from app.provider_config import validate_payload as validate_remote_provider, encrypt as encrypt_provider_secret, decrypt as decrypt_provider_secret, safe_status as remote_provider_status, test_connectivity as test_remote_connectivity else: from .domain import deduplication_key, deduplicate_businesses, is_suppressed, normalize_business, normalize_domain, normalize_phone, match_businesses - from .sources import adapter_for, contains_secret, available_adapters, normalize_record, circuit_is_open + from .sources import adapter_for, contains_secret, available_adapters, normalize_record, circuit_is_open, DISCOVERY_CRITERIA_FIELDS from .domain_intelligence import normalize_registrable_domain, resolve_domain, generate_candidate_domains from .website_scanner import scan_website, validate_url from .contact_extractor import extract_contacts, MAX_HTML_BYTES, MAX_RESULTS @@ -1446,6 +1446,7 @@ class ApiHandler(BaseHTTPRequestHandler): source_code=str(config.get('provider','')).strip().lower() if requested == 'approved_directory' and str(config.get('provider','')).strip().lower() in {'openstreetmap','wikidata','common_crawl'} else requested if not name or requested not in ('csv','manual','google_places','bing_local','approved_directory','openstreetmap','wikidata','common_crawl','public_website','permitted_social','ct_logs','dns','rdap') or not isinstance(config,dict):return self.send_json(400,{"error":"invalid_source"}) if contains_secret(config):return self.send_json(400,{"error":"secret_not_permitted"}) + if any(field in config for field in DISCOVERY_CRITERIA_FIELDS):return self.send_json(400,{"error":"source_configuration_contains_criteria"}) try: validation=adapter_for(source_code).validate_config(config) # Registration may precede the actual local payload or optional provider @@ -1660,15 +1661,19 @@ def _run_source_discovery(db, job, handler): handler.add_job_event(db,job["id"],org,"source.started",f"Starting {source['display_name'] or source['kind']}",5) try: config=json.loads(source["config_json"] or "{}") - if query and query["source_id"]==source["id"]: - config.update(json.loads(query["query_json"] or "{}")) quota=json.loads(source["quota_json"] or "{}") daily_limit=int(quota.get("daily_limit", payload.get("daily_limit", 100000))) per_run_limit=int(quota.get("per_run_limit", max_records)) used_today=db.execute("SELECT COUNT(*) FROM source_records WHERE organization_id=? AND source_id=? AND date(created_at)=date('now')",(org,source["id"])).fetchone()[0] if used_today >= daily_limit or per_run_limit <= 0: handler.add_job_event(db,job["id"],org,"source.quota_exceeded",f"Quota reached for {source['kind']}",10,"SOURCE_QUOTA_EXCEEDED"); continue - page=adapter_for(source["source_code"] or source["kind"]).discover(source_config_with_credentials(db, source, config)) + criteria = json.loads(query["query_json"] or "{}") if query else payload.get("criteria", {}) + limits = { + "max_records": max_records, + "daily_limit": daily_limit, + "per_run_limit": per_run_limit, + } + page=adapter_for(source["source_code"] or source["kind"]).discover(source_config_with_credentials(db, source, config), criteria=criteria, limits=limits) except Exception as exc: blocked_count+=1; failures=int(source["consecutive_failures"])+1 db.execute("UPDATE sources SET health_status='unhealthy',consecutive_failures=?,circuit_open=?,last_failure_at=CURRENT_TIMESTAMP,last_error=?,updated_at=CURRENT_TIMESTAMP WHERE id=?",(failures,int(circuit_is_open(failures)),str(exc)[:300],source["id"])) diff --git a/apps/api/app/sources.py b/apps/api/app/sources.py index 05d0238..a0c286c 100644 --- a/apps/api/app/sources.py +++ b/apps/api/app/sources.py @@ -18,6 +18,7 @@ except ImportError: SECRET_KEYS = {"password", "passwd", "secret", "token", "api_key", "apikey", "authorization", "private_key", "credential"} NETWORK_KINDS = {"google_places", "bing_local", "approved_directory", "public_website", "permitted_social", "ct_logs", "dns", "rdap"} +DISCOVERY_CRITERIA_FIELDS = {"query", "category", "city", "location", "keywords", "province", "country", "language", "search", "phrase", "industry", "keyword"} def contains_secret(value: Any, path: str = "") -> str | None: @@ -123,7 +124,7 @@ class _Base: def health_check(self, config): result = self.validate_config(config) return SourceHealth("healthy" if result.valid else "unhealthy", last_error=None if result.valid else "; ".join(result.errors)) - def discover(self, config, cursor=None): + def discover(self, config, cursor=None, criteria=None, limits=None): result = self.validate_config(config) if not result.valid: raise ValueError(result.errors[0]) raise RuntimeError("source_not_configured") @@ -135,7 +136,7 @@ class ManualSource(_Base): if not result.valid: return result if not isinstance(config.get("rows"), list): return ValidationResult(False, ["rows must be a list"]) return ValidationResult(True) - def discover(self, config, cursor=None): + def discover(self, config, cursor=None, criteria=None, limits=None): result = self.validate_config(config) if not result.valid: raise ValueError(result.errors[0]) return DiscoveryPage([normalize_record(row) for row in config["rows"] if isinstance(row, Mapping)], metadata={"adapter":self.source_code}) @@ -151,7 +152,7 @@ class CsvSource(_Base): 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, cursor=None): + def discover(self, config, cursor=None, criteria=None, limits=None): result = self.validate_config(config) if not result.valid: raise ValueError(result.errors[0]) reader = csv.DictReader(io.StringIO(config["csv"].replace("\r\n", "\n"))) @@ -194,7 +195,7 @@ class PublicWebsiteSource(_Base): except ValueError: return ValidationResult(False, ["unsafe public website URL"]) return ValidationResult(True) - def discover(self, config, cursor=None): + def discover(self, config, cursor=None, criteria=None, limits=None): result = self.validate_config(config) if not result.valid: raise ValueError(result.errors[0]) urls = config.get("urls", config.get("url")) @@ -225,10 +226,10 @@ class CtLogsSource(_HttpJsonSource): return ValidationResult(False, ["domain or query is required"]) return ValidationResult(True) - def discover(self, config, cursor=None): + def discover(self, config, cursor=None, criteria=None, limits=None): result = self.validate_config(config) if not result.valid: raise ValueError(result.errors[0]) - query = str(config.get("domain", config.get("query"))).strip() + query = str(config.get("domain", config.get("query", ""))).strip() endpoint = "https://crt.sh/?" + urlencode({"q": "%25." + query.lstrip("%.") if not query.startswith("%") else query, "output": "json"}) payload, source_url = self._get_json(endpoint) if not isinstance(payload, list): raise ValueError("invalid_ct_response") @@ -254,7 +255,7 @@ class DnsSource(_Base): if not isinstance(domains, list) or not domains or len(domains) > 100: return ValidationResult(False, ["domains must contain 1 to 100 names"]) return ValidationResult(True) - def discover(self, config, cursor=None): + def discover(self, config, cursor=None, criteria=None, limits=None): result = self.validate_config(config) if not result.valid: raise ValueError(result.errors[0]) import socket @@ -280,7 +281,7 @@ class RdapSource(_HttpJsonSource): if not domain or "." not in domain: return ValidationResult(False, ["domain is required"]) return ValidationResult(True) - def discover(self, config, cursor=None): + def discover(self, config, cursor=None, criteria=None, limits=None): result = self.validate_config(config) if not result.valid: raise ValueError(result.errors[0]) domain = str(config["domain"]).strip().lower().rstrip(".") @@ -303,7 +304,7 @@ class GatedSource(_Base): if not isinstance(config.get("rate_limit", 1), (int, float)) or config.get("rate_limit", 1) <= 0: return ValidationResult(False, ["positive rate_limit is required"]) return ValidationResult(True) - def discover(self, config, cursor=None): + def discover(self, config, cursor=None, criteria=None, limits=None): result = self.validate_config(config) if not result.valid: raise ValueError(result.errors[0]) # Network execution is delegated to an explicitly approved provider; never guess. @@ -322,10 +323,9 @@ class ApprovedDirectorySource(GatedSource): if not result.valid:return result provider=str(config.get("provider", "")).strip().lower() if provider not in {"openstreetmap", "wikidata", "common_crawl"}: return ValidationResult(False,["provider must be openstreetmap, wikidata, or common_crawl"]) - if not str(config.get("query", "")).strip(): return ValidationResult(False,["query is required"]) return ValidationResult(True) - def discover(self, config, cursor=None): + def discover(self, config, cursor=None, criteria=None, limits=None): result=self.validate_config(config) if not result.valid: raise ValueError(result.errors[0]) provider=str(config["provider"]).lower(); query=str(config["query"]).strip(); limit=max(1,min(100,int(config.get("max_records",50)))) @@ -363,7 +363,7 @@ class GooglePlacesSource(GatedSource): if not str(config.get("query", "")).strip(): return ValidationResult(False, ["query is required"]) return ValidationResult(True) - def discover(self, config, cursor=None): + def discover(self, config, cursor=None, criteria=None, limits=None): result = self.validate_config(config) if not result.valid: raise ValueError(result.errors[0]) api_key = str(config.get("_api_key", "")).strip() @@ -383,24 +383,24 @@ class OpenStreetMapSource(ApprovedDirectorySource): kind = source_code = "openstreetmap" display_name = "OpenStreetMap / Overpass" optional = False - def discover(self, config, cursor=None): - page = super().discover({**dict(config), "provider": "openstreetmap"}, cursor) + def discover(self, config, cursor=None, criteria=None, limits=None): + page = super().discover({**dict(config), "provider": "openstreetmap"}, cursor, criteria=criteria, limits=limits) return DiscoveryPage(page.records, page.next_cursor, {**page.metadata, "adapter": self.source_code}) class WikidataSource(ApprovedDirectorySource): kind = source_code = "wikidata" display_name = "Wikidata" optional = False - def discover(self, config, cursor=None): - page = super().discover({**dict(config), "provider": "wikidata"}, cursor) + def discover(self, config, cursor=None, criteria=None, limits=None): + page = super().discover({**dict(config), "provider": "wikidata"}, cursor, criteria=criteria, limits=limits) return DiscoveryPage(page.records, page.next_cursor, {**page.metadata, "adapter": self.source_code}) class CommonCrawlSource(ApprovedDirectorySource): kind = source_code = "common_crawl" display_name = "Common Crawl index" optional = False - def discover(self, config, cursor=None): - page = super().discover({**dict(config), "provider": "common_crawl"}, cursor) + def discover(self, config, cursor=None, criteria=None, limits=None): + page = super().discover({**dict(config), "provider": "common_crawl"}, cursor, criteria=criteria, limits=limits) return DiscoveryPage(page.records, page.next_cursor, {**page.metadata, "adapter": self.source_code}) def _gated(code, name): diff --git a/apps/api/tests/test_sources_phase5.py b/apps/api/tests/test_sources_phase5.py index 208d337..66d96e4 100644 --- a/apps/api/tests/test_sources_phase5.py +++ b/apps/api/tests/test_sources_phase5.py @@ -81,8 +81,8 @@ class SourceApiTests(unittest.TestCase): self.assertEqual(self.req('POST',f'/api/v1/sources/{sid}/ingest',payload)[0],409) db=sqlite3.connect(self.tmp.name+'/x.db'); self.assertTrue(db.execute("select 1 from audit_log where action='source.disabled'").fetchone()); db.close() def test_queries_enqueue_and_records_are_tenant_scoped(self): - _,source=self.req('POST','/api/v1/sources',{'name':'CSV','kind':'csv','enabled':True}) - _,q=self.req('POST','/api/v1/discovery-queries',{'source_id':source['id'],'name':'q','query':{'csv':'name\nA'}}) + _,source=self.req('POST','/api/v1/sources',{'name':'CSV','kind':'csv','enabled':True,'config':{'csv':'name\nA'}}) + _,q=self.req('POST','/api/v1/discovery-queries',{'source_id':source['id'],'name':'q','query':{}}) status,job=self.req('POST',f"/api/v1/discovery-queries/{q['id']}/run",{}) self.assertEqual(status,202); self.assertEqual(job['type'],'source_discovery') for _ in range(100): @@ -105,12 +105,52 @@ class SourceApiTests(unittest.TestCase): self.assertEqual(self.req('PATCH', f"/api/v1/sources/{website['id']}", {'enabled': True})[0], 409) def test_duplicate_source_registration_is_idempotent(self): - payload={'name':'OpenStreetMap / Overpass · plumbers','kind':'openstreetmap','config':{'provider':'openstreetmap','query':'plumbers','location':'Cape Town','approved':True,'public_access':True,'terms_accepted':True,'rate_limit':1}} + payload={'name':'OpenStreetMap / Overpass · plumbers','kind':'openstreetmap','config':{'provider':'openstreetmap','approved':True,'public_access':True,'terms_accepted':True,'rate_limit':1}} status, created=self.req('POST','/api/v1/sources',payload); self.assertEqual(status,201); self.assertTrue(created['created']) status, reused=self.req('POST','/api/v1/sources',payload); self.assertEqual(status,200); self.assertFalse(reused['created']); self.assertEqual(reused['id'],created['id']) - second={**payload,'name':'OpenStreetMap / Overpass · plumbers · Durban','config':{**payload['config'],'location':'Durban'}} + second={**payload,'name':'OpenStreetMap / Overpass · plumbers · Durban'} status, other=self.req('POST','/api/v1/sources',second); self.assertEqual(status,201); self.assertTrue(other['created']); self.assertNotEqual(other['id'],created['id']) + def test_source_configuration_rejects_discovery_criteria(self): + forbidden = {'query': 'plumbers', 'category': 'trades', 'city': 'Cape Town', 'location': 'Western Cape'} + for field, value in forbidden.items(): + with self.subTest(field=field): + status, body = self.req('POST', '/api/v1/sources', { + 'name': 'Manual ' + field, 'kind': 'manual', + 'config': {'rows': [], field: value}, + }) + self.assertEqual(status, 400) + self.assertEqual(body['error'], 'source_configuration_contains_criteria') + + def test_source_worker_passes_query_criteria_and_effective_limits_to_connector(self): + status, source = self.req('POST', '/api/v1/sources', { + 'name': 'Criteria manual', 'kind': 'manual', 'enabled': True, + 'config': {'rows': [{'name': 'Criteria Acme'}]}, + }) + self.assertEqual(status, 201) + status, query = self.req('POST', '/api/v1/discovery-queries', { + 'source_id': source['id'], 'name': 'Cape solar', + 'query': {'keywords': ['solar'], 'city': 'Cape Town'}, + 'max_records': 7, 'daily_limit': 9, + }) + self.assertEqual(status, 201) + observed = [] + from app.sources import ManualSource + original = ManualSource.discover + def spy(adapter, config, cursor=None, criteria=None, limits=None): + observed.append((criteria, limits)) + return original(adapter, config, cursor, criteria=criteria, limits=limits) + with patch('app.sources.ManualSource.discover', new=spy): + status, job = self.req('POST', f"/api/v1/discovery-queries/{query['id']}/run", {}) + self.assertEqual(status, 202) + for _ in range(100): + _, current = self.req('GET', f"/api/v1/jobs/{job['id']}") + if current['status'] in ('succeeded', 'failed'): + break + threading.Event().wait(.01) + self.assertEqual(current['status'], 'succeeded') + self.assertEqual(observed, [({'keywords': ['solar'], 'city': 'Cape Town'}, {'max_records': 7, 'daily_limit': 9, 'per_run_limit': 7})]) + def test_source_configuration_can_be_saved_before_enablement(self): status, source = self.req('POST', '/api/v1/sources', {'name': 'DNS', 'kind': 'dns', 'config': {}}) self.assertEqual(status, 201) diff --git a/apps/web/app.js b/apps/web/app.js index 75472f2..17a65f7 100644 --- a/apps/web/app.js +++ b/apps/web/app.js @@ -318,7 +318,7 @@ $('directDiscoveryResultStatus').textContent = discoveryLabel(status); const controls = '
Pause/resume require the job-control API.
'; const sourceRows = run.sources || run.source_health || []; - const health = sourceRows.length ? '

Source health

' + sourceRows.length + ' sources
' + sourceRows.map(source => '
' + esc(source.name || source.source_name || ('Source ' + (source.source_id || ''))) + '' + esc(source.status || source.health || 'Unknown') + '' + esc(source.result_count ?? source.results ?? 0) + ' results · ' + esc(source.error || 'No errors') + '
').join('') + '
' : '

Source health

Per-source health will appear when the API returns source telemetry.

'; + const health = sourceRows.length ? '

Source health

' + sourceRows.length + ' sources
' + sourceRows.map(source => '
' + esc(source.name || source.source_name || ('Source ' + (source.source_id || ''))) + '' + esc(source.status || source.health || 'Unknown') + '' + esc(source.result_count ?? source.results ?? 0) + ' results · ' + esc(source.error || 'No errors') + '
').join('') + '
' : '

Source health

Per-source health will appear when the API returns source telemetry.

'; const log = events.length ? events.map(event => '
  • ' + esc(event.message || event.type || 'Run event') + '' + esc(event.created_at || event.timestamp || '') + '
    • ').join('') : '
  • No persisted events returned yet.
    • '; const stats = '
    Progress' + progress + '%
    Records' + esc(run.result_count ?? candidates.length) + '
    Daily limit' + esc(run.daily_limit ?? '—') + '
    Errors' + (Array.isArray(errors) ? errors.length : 0) + '
    '; const cards = candidates.length ? candidates.slice(0, 50).map(item => '
    ' + esc(item.name || item.business_name || item.title || 'Unnamed candidate') + '' + esc(item.website || item.url || item.domain || 'No URL returned') + '' + esc(item.source_url || item.source || 'Source provenance unavailable') + ' · ' + esc(item.confidence || item.score || 'Confidence unknown') + '
    ').join('') : '

    No candidates returned by this run.

    '; diff --git a/apps/web/asset-manifest.json b/apps/web/asset-manifest.json index 1281e41..85ae77c 100644 --- a/apps/web/asset-manifest.json +++ b/apps/web/asset-manifest.json @@ -14,7 +14,7 @@ ], "integrity": { "config.js": "sha256-7792697d8640937cb0573d314897e8be96230b39418fc1b8a40b66c8138005df", - "app.js": "sha256-fe50db5a4d978922cd761e03c80502b775611e14e2e1118f29147a2de689dddc", + "app.js": "sha256-662a5eb000a9ce90bc4a75c4cb5fbfbe097f5ebff5c06afa9452635872a60bf1", "styles.css": "sha256-ba90290ab11e82a6b2639dfd70d1e74502c1cacb2b26cf1db92b45beb67ac03f", "index.html": "sha256-3bd7fda6c92bdb3b65e61ac4457c855c99e5bbd82fcfc26959ce70d93b064aa9", "health.html": "sha256-c352a6f37aa24628cfc8d5709a70ff2d94181d192ed5fe00916cca9378a61d81", diff --git a/docs/source-discovery-rebuild-plan.md b/docs/source-discovery-rebuild-plan.md new file mode 100644 index 0000000..d8e21db --- /dev/null +++ b/docs/source-discovery-rebuild-plan.md @@ -0,0 +1,46 @@ +# Source and Discovery Rebuild Plan + +## Audit snapshot + +- **Frontend:** `apps/web/index.html`, `apps/web/app.js`, `apps/web/styles.css`; static same-origin delivery via `apps/web/server.py`. +- **Backend:** threaded Python/SQLite HTTP API in `apps/api/app/main.py`; adapters in `apps/api/app/sources.py`; bounded crawler/discovery in `apps/api/app/discovery.py`. +- **Persistence:** additive SQLite schema in `apps/api/schema.sql`, with compatibility upgrades in `connect()`. +- **Background work:** in-process `source_discovery` and `scoped_discovery` job worker plus schedule worker in `main.py`. +- **Existing boundaries to preserve:** tenant filtering, role checks, write-only provider credentials, suppression precedence, deterministic scoring, CRM, evidence, no automated outreach, and same-origin web/API delivery. + +## Acceptance slices + +1. **Reusable source registry** + - Separate connector configuration from discovery criteria. + - Add safe migration fields for owner, terms, rate/quota, credential status, and connector configuration. + - Make duplicate handling identity-based and return the existing registered source visibly. + - Keep previews separate from real persisted source IDs. + +2. **Connector contract and safe adapters** + - Normalize connector methods around configuration validation, connection tests, criteria-based discovery, health, limits, normalized evidence, and structured errors. + - Preserve CSV/manual behavior and repair OSM/Wikidata/Common Crawl/CT/RDAP/DNS/public crawler connectors. + - Add a disabled-by-default Google Browser Search connector that reports challenge/blocked states without bypassing controls. Runtime/dependency requirements will be documented and feature-gated. + +3. **Criteria-based discovery runs** + - Persist explicit criteria independently from source configuration. + - Select only enabled, valid sources; validate limits and dry-runs. + - Pass criteria and limits to connectors, retain source-level result/error status, record provenance, deduplicate candidates, and distinguish complete, partial, blocked, and failed outcomes. + +4. **Evidence/enrichment and AI boundaries** + - Preserve bounded website/domain/contact checks and provenance. + - Keep AI evidence-driven, strict-schema, write-only-provider-configured, suppression-overridden, and review-only. + +5. **Sources and Discovery UI** + - Rebuild the registry, source details, source setup, criteria builder, run history, live logs, per-source status, candidate provenance, and actionable empty/error states. + - Every action presents loading, success, validation, or API-error feedback; no fake IDs, silent actions, demo health, or fabricated records. + +6. **Verification, documentation, and deployment** + - Add migration, source identity, criteria, connector mock, dedupe/provenance, blocked/partial, Google challenge, AI schema, and suppression tests. + - Run backend/full/frontend/desktop/Compose checks and live authenticated workflow checks where available. + - Update deployment/source architecture documentation, deploy only after data backup and verified live results. + +## Known operational limits + +- SQLite/in-process jobs remain a controlled single-node/pilot architecture. +- Public connectors must remain bounded and may return an honest blocked/network/rate-limit result. +- Google Browser Search remains disabled unless its reviewed browser runtime is installed and explicitly enabled; it will never bypass CAPTCHA, authentication, or anti-bot controls.