diff --git a/apps/api/app/main.py b/apps/api/app/main.py index eb21a7d..af1227d 100644 --- a/apps/api/app/main.py +++ b/apps/api/app/main.py @@ -8,7 +8,7 @@ from urllib.parse import parse_qs, urlparse 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, score_business, normalize_domain, normalize_phone, match_businesses - from app.sources import adapter_for, contains_secret + from app.sources import adapter_for, contains_secret, available_adapters, normalize_record, circuit_is_open 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 @@ -21,7 +21,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, score_business, normalize_domain, normalize_phone, match_businesses - from .sources import adapter_for, contains_secret + from .sources import adapter_for, contains_secret, available_adapters, normalize_record, circuit_is_open 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 @@ -75,6 +75,10 @@ def connect(db_path: str) -> sqlite3.Connection: "pipeline_entries": (("notes", "TEXT NOT NULL DEFAULT ''"), ("next_action", "TEXT NOT NULL DEFAULT ''"), ("follow_up_at", "TEXT"), ("actor_user_id", "INTEGER"), ("idempotency_key", "TEXT"), ("version", "INTEGER NOT NULL DEFAULT 1")), "interactions": (("outcome", "TEXT NOT NULL DEFAULT 'other'"), ("notes", "TEXT NOT NULL DEFAULT ''"), ("next_action", "TEXT NOT NULL DEFAULT ''"), ("follow_up_at", "TEXT"), ("actor_user_id", "INTEGER"), ("idempotency_key", "TEXT")), "suppressions": (("active", "INTEGER NOT NULL DEFAULT 1"), ("updated_at", "TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP"), ("actor_user_id", "INTEGER")), + "sources": (("source_code", "TEXT NOT NULL DEFAULT ''"), ("display_name", "TEXT NOT NULL DEFAULT ''"), ("approved", "INTEGER NOT NULL DEFAULT 0"), ("policy_json", "TEXT NOT NULL DEFAULT '{}'"), ("quota_json", "TEXT NOT NULL DEFAULT '{}'")), + "discovery_queries": (("selected_adapters_json", "TEXT NOT NULL DEFAULT '[]'"), ("location", "TEXT NOT NULL DEFAULT ''"), ("category", "TEXT NOT NULL DEFAULT ''"), ("max_records", "INTEGER NOT NULL DEFAULT 100"), ("daily_limit", "INTEGER NOT NULL DEFAULT 1000"), ("schedule", "TEXT NOT NULL DEFAULT ''"), ("dry_run", "INTEGER NOT NULL DEFAULT 0"), ("lifecycle", "TEXT NOT NULL DEFAULT 'draft'")), + "discovery_runs": (("selected_adapters_json", "TEXT NOT NULL DEFAULT '[]'"), ("location", "TEXT NOT NULL DEFAULT ''"), ("category", "TEXT NOT NULL DEFAULT ''"), ("max_records", "INTEGER NOT NULL DEFAULT 100"), ("daily_limit", "INTEGER NOT NULL DEFAULT 1000"), ("schedule", "TEXT NOT NULL DEFAULT ''"), ("dry_run", "INTEGER NOT NULL DEFAULT 0"), ("lifecycle", "TEXT NOT NULL DEFAULT 'draft'"), ("paused_at", "TEXT")), + "source_records": (("discovery_run_id", "INTEGER"), ("normalized_key", "TEXT NOT NULL DEFAULT ''"), ("provenance_json", "TEXT NOT NULL DEFAULT '{}'")), }.items(): existing = {r[1] for r in db.execute(f"PRAGMA table_info({table})")} for col, definition in additions: @@ -631,7 +635,13 @@ class ApiHandler(BaseHTTPRequestHandler): if path=="/api/v1/businesses": return self.list_businesses(db,org,parse_qs(parsed.query)) if path=="/api/v1/merge-history": return self.list_merge_history(db,org) if path=="/api/v1/sources": return self.list_sources(db,org) + if path=="/api/v1/sources/adapters": return self.send_json(200,{"items":available_adapters()}) if path=="/api/v1/discovery-queries": return self.list_queries(db,org) + if path.startswith("/api/v1/discovery-runs/"): + bits=path.split("/") + if len(bits)==5 and bits[4].isdigit(): + run=db.execute("SELECT * FROM discovery_runs WHERE id=? AND organization_id=?",(int(bits[4]),org)).fetchone() + return self.send_json(200,self._discovery_run_json(run)) if run else self.send_json(404,{"error":"not_found"}) if path=="/api/v1/discovery-runs": return self.list_discovery_runs(db,org,parse_qs(parsed.query)) if path=="/api/v1/discovery/provider-status": return self.send_json(200, ai_research_provider_status()) if path=="/api/v1/discovery/ai-provider-status": return self.send_json(200, ai_research_provider_status()) @@ -652,6 +662,9 @@ class ApiHandler(BaseHTTPRequestHandler): if path=="/api/v1/outreach/provider-config": return self.provider_config(db,user) if path in ("/api/v1/reports/pipeline","/api/v1/reports/outcomes","/api/v1/reports/activity"): return self.report(db,org,path.rsplit('/',1)[1],parse_qs(parsed.query)) if path.startswith("/api/v1/jobs/"): return self.get_job_route(db,org,path,parse_qs(parsed.query)) + if path.startswith("/api/v1/sources/"): + bits=path.split("/") + if len(bits)==6 and bits[4].isdigit() and bits[5] == "health": return self.source_health(int(bits[4]),db,user) if path.startswith("/api/v1/businesses/"): bits=path.split("/"); ident=bits[4] if len(bits)>4 else "" if not ident.isdigit(): return self.send_json(404,{"error":"not_found"}) @@ -833,7 +846,7 @@ class ApiHandler(BaseHTTPRequestHandler): def _discovery_run_json(self, row): item = row_json(row) - for field, default in (("criteria_json", {}), ("seed_urls_json", []), ("result_json", {})): + for field, default in (("criteria_json", {}), ("seed_urls_json", []), ("result_json", {}), ("selected_adapters_json", [])): try: item[field[:-5]] = json.loads(item.pop(field) or json.dumps(default)) except (TypeError, ValueError): item[field[:-5]] = default return item @@ -846,9 +859,11 @@ class ApiHandler(BaseHTTPRequestHandler): def create_scoped_discovery(self, payload, db, user): criteria = payload.get("criteria", {}); seeds = payload.get("seed_urls") - criteria_only = seeds is None + selected_adapters = payload.get("selected_adapters", payload.get("sources", [])) + source_mode = bool(selected_adapters) and seeds is None + criteria_only = seeds is None and not source_mode if not isinstance(criteria, dict): return self.send_json(400, {"error": "invalid_criteria"}) - if not criteria_only and (not isinstance(seeds, list) or not seeds): return self.send_json(400, {"error": "seed_urls_required"}) + if not criteria_only and not source_mode and (not isinstance(seeds, list) or not seeds): return self.send_json(400, {"error": "seed_urls_required"}) if criteria_only: try: validate_ai_research_criteria(criteria) except AIResearchConfigError as exc: @@ -859,26 +874,26 @@ class ApiHandler(BaseHTTPRequestHandler): legacy = search_provider_status() if status["status"] != "ready" and legacy["status"] != "ready": return self.send_json(503, {"error": status["status"], "provider": status["provider"]}) try: - if not criteria_only and len(seeds) > 5: raise ValueError("invalid_criteria") + if not criteria_only and not source_mode and len(seeds) > 5: raise ValueError("invalid_criteria") if len(json.dumps(criteria).encode()) > 8192: raise ValueError("invalid_criteria") if not isinstance(criteria.get("keywords", criteria.get("keyword", [])), (list, str)): raise ValueError("invalid_criteria") max_pages = int(payload.get("max_pages", 20)); max_candidates = int(payload.get("max_candidates", 50)) if not 1 <= max_pages <= 20 or not 1 <= max_candidates <= 50: raise ValueError("invalid_limits") except (ValueError, TypeError) as exc: return self.send_json(400, {"error": str(exc) or "invalid_criteria"}) - if not criteria_only: + if not criteria_only and not source_mode: try: for url in seeds: validate_url(url) except (ValueError, TypeError): return self.send_json(400, {"error": "unsafe_seed_url"}) key = str(payload.get("idempotency_key", "")).strip() if not key or len(key) > 200: return self.send_json(400, {"error": "invalid_idempotency_key"}) - job_payload = {"criteria": criteria, "seed_urls": seeds if not criteria_only else None, "max_pages": max_pages, "max_candidates": max_candidates} - result = self.create_job({"type": "scoped_discovery", "payload": job_payload, "idempotency_key": key, "_accepted": True, "_defer_wakeup": True}, db, user) + job_payload = {"criteria": criteria, "seed_urls": seeds if not criteria_only else None, "selected_adapters": selected_adapters, "location": payload.get("location", ""), "category": payload.get("category", ""), "max_records": payload.get("max_records", 100), "daily_limit": payload.get("daily_limit", 1000), "dry_run": bool(payload.get("dry_run", False)), "max_pages": max_pages, "max_candidates": max_candidates} + result = self.create_job({"type": "source_discovery" if source_mode else "scoped_discovery", "payload": job_payload, "idempotency_key": key, "_accepted": True, "_defer_wakeup": True}, db, user) # create_job has already committed; read its id from the response is not available, # so resolve by the tenant-scoped idempotency key. job = db.execute("SELECT * FROM jobs WHERE organization_id=? AND idempotency_key=?", (user["organization_id"], key)).fetchone() if not db.execute("SELECT id FROM discovery_runs WHERE organization_id=? AND job_id=?", (user["organization_id"], job["id"])).fetchone(): - db.execute("INSERT INTO discovery_runs(organization_id,job_id,criteria_json,seed_urls_json) VALUES(?,?,?,?)", (user["organization_id"], job["id"], json.dumps(criteria, sort_keys=True), json.dumps(seeds if not criteria_only else []))) + db.execute("INSERT INTO discovery_runs(organization_id,job_id,selected_adapters_json,location,category,max_records,daily_limit,schedule,dry_run,lifecycle,criteria_json,seed_urls_json) VALUES(?,?,?,?,?,?,?,?,?,?,?,?)", (user["organization_id"], job["id"], json.dumps(selected_adapters), str(payload.get("location", "")), str(payload.get("category", "")), int(payload.get("max_records", 100)), int(payload.get("daily_limit", 1000)), str(payload.get("schedule", "")), int(bool(payload.get("dry_run", False))), "queued", json.dumps(criteria, sort_keys=True), json.dumps(seeds if not criteria_only else []))) self.audit(db, user, "discovery.created", str(job["id"])); db.commit() getattr(self.server, "job_wakeup", threading.Event()).set() return result @@ -1080,6 +1095,13 @@ class ApiHandler(BaseHTTPRequestHandler): if len(bits_outreach)==7 and bits_outreach[:4]==["","api","v1","outreach"] and bits_outreach[4]=="drafts" and bits_outreach[5].isdigit() and bits_outreach[6] in {"approve","send"}: return self.approve_outreach_draft(int(bits_outreach[5]),db,user) if bits_outreach[6]=="approve" else self.send_outreach_draft(int(bits_outreach[5]),db,user) if path=="/api/v1/sources":return self.create_source(payload,db,user) if path=="/api/v1/discovery":return self.create_scoped_discovery(payload,db,user) + if path.startswith("/api/v1/discovery-runs/"): + bits=path.split("/") + if len(bits)==6 and bits[4].isdigit() and bits[5] in {"pause","resume","cancel"}: return self.discovery_run_action(int(bits[4]),bits[5],db,user) + if path=="/api/v1/discovery-runs": + payload.setdefault('criteria',{}) + payload.setdefault('idempotency_key', 'discovery-run-' + hashlib.sha256(json.dumps(payload,sort_keys=True).encode()).hexdigest()[:24]) + return self.create_scoped_discovery(payload,db,user) if path=="/api/v1/discovery-queries":return self.create_query(payload,db,user) if path=="/api/v1/suppressions":return self.create_suppression(payload,db,user) if path=="/api/v1/suppressions/import":return self.import_suppressions(payload,db,user) @@ -1288,7 +1310,7 @@ class ApiHandler(BaseHTTPRequestHandler): else:seen.add(key);accepted.append(b) return self.send_json(200,{"accepted":len(accepted),"duplicates":len(rows)-len(normalized)+len(normalized)-len(accepted)-suppressed,"suppressed":suppressed,"rows":accepted}) def list_sources(self,db,org): - cols='id,organization_id,name,kind,enabled,health_status,consecutive_failures,circuit_open,last_success_at,last_failure_at,last_error,created_at,updated_at' + cols='id,organization_id,name,kind,source_code,display_name,enabled,approved,policy_json,quota_json,health_status,consecutive_failures,circuit_open,last_success_at,last_failure_at,last_error,created_at,updated_at' return self.send_json(200,{"organization_id":org,"items":[row_json(r) for r in db.execute(f"SELECT {cols} FROM sources WHERE organization_id=? ORDER BY id",(org,))]}) def list_queries(self,db,org): return self.send_json(200,{"organization_id":org,"items":[row_json(r) for r in db.execute("SELECT * FROM discovery_queries WHERE organization_id=? ORDER BY id",(org,))]}) @@ -1306,24 +1328,30 @@ class ApiHandler(BaseHTTPRequestHandler): out.append(x) return self.send_json(200,{"organization_id":org,"items":out,"limit":limit,"offset":offset,"has_more":len(rows)>limit}) def create_source(self,payload,db,user): - name=str(payload.get('name','')).strip(); kind=str(payload.get('kind','')).strip().lower(); config=payload.get('config',{}) - if not name or kind not in ('csv','manual') or not isinstance(config,dict):return self.send_json(400,{"error":"invalid_source"}) + name=str(payload.get('name','')).strip(); kind=str(payload.get('source_code',payload.get('kind',''))).strip().lower(); config=payload.get('config',{}) + optional = kind not in ('csv','manual') + if not name or kind not in ('csv','manual','google_places','bing_local','approved_directory','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"}) try: - validation=adapter_for(kind).validate(config) - if config and not validation.valid:return self.send_json(400,{"error":"invalid_source_config","details":validation.errors}) - cur=db.execute("INSERT INTO sources(organization_id,name,kind,enabled,config_json) VALUES(?,?,?,?,?)",(user['organization_id'],name,kind,int(bool(payload.get('enabled',False))),json.dumps(config,sort_keys=True))) + validation=adapter_for(kind).validate_config(config) + if (config or optional) and not validation.valid:return self.send_json(400,{"error":"invalid_source_config","details":validation.errors}) + cur=db.execute("INSERT INTO sources(organization_id,name,kind,source_code,display_name,enabled,approved,config_json,policy_json,quota_json) VALUES(?,?,?,?,?,?,?,?,?,?)",(user['organization_id'],name,kind,kind,str(payload.get('display_name') or adapter_for(kind).display_name),int(bool(payload.get('enabled',False))),int(bool(payload.get('approved',config.get('approved',False)))),json.dumps(config,sort_keys=True),json.dumps(payload.get('policy',{}),sort_keys=True),json.dumps(payload.get('quota',{}),sort_keys=True))) except sqlite3.IntegrityError:return self.send_json(409,{"error":"duplicate_source"}) - self.audit(db,user,'source.created',str(cur.lastrowid));db.commit();return self.send_json(201,row_json(db.execute("SELECT id,organization_id,name,kind,enabled,health_status,consecutive_failures,circuit_open,last_success_at,last_failure_at,last_error,created_at,updated_at FROM sources WHERE id=?",(cur.lastrowid,)).fetchone())) + self.audit(db,user,'source.created',str(cur.lastrowid));db.commit();return self.send_json(201,row_json(db.execute("SELECT id,organization_id,name,kind,source_code,display_name,enabled,approved,policy_json,quota_json,health_status,consecutive_failures,circuit_open,last_success_at,last_failure_at,last_error,created_at,updated_at FROM sources WHERE id=?",(cur.lastrowid,)).fetchone())) def update_source(self,sid,payload,db,user): if not db.execute("SELECT id FROM sources WHERE id=? AND organization_id=?",(sid,user['organization_id'])).fetchone():return self.send_json(404,{"error":"not_found"}) if 'enabled' not in payload:return self.send_json(400,{"error":"enabled_required"}) value=int(bool(payload['enabled']));db.execute("UPDATE sources SET enabled=?,updated_at=CURRENT_TIMESTAMP WHERE id=?",(value,sid));self.audit(db,user,'source.enabled' if value else 'source.disabled',str(sid));db.commit();return self.send_json(200,row_json(db.execute("SELECT * FROM sources WHERE id=?",(sid,)).fetchone())) def create_query(self,payload,db,user): sid=payload.get('source_id');name=str(payload.get('name','')).strip();query=payload.get('query',{}) + selected=payload.get('selected_adapters',payload.get('sources',[])); location=str(payload.get('location','')).strip(); category=str(payload.get('category','')).strip(); schedule=str(payload.get('schedule','')).strip() + try: max_records=int(payload.get('max_records',100)); daily_limit=int(payload.get('daily_limit',1000)) + except (TypeError,ValueError): return self.send_json(400,{"error":"invalid_limits"}) + if not isinstance(selected,list) or any(str(x) not in {a["source_code"] for a in available_adapters()} for x in selected): return self.send_json(400,{"error":"invalid_adapters"}) + if max_records<1 or max_records>10000 or daily_limit<1 or daily_limit>100000: return self.send_json(400,{"error":"invalid_limits"}) if not isinstance(sid,int) or not name or not isinstance(query,dict) or contains_secret(query):return self.send_json(400,{"error":"invalid_query"}) if not db.execute("SELECT id FROM sources WHERE id=? AND organization_id=?",(sid,user['organization_id'])).fetchone():return self.send_json(404,{"error":"not_found"}) - try:cur=db.execute("INSERT INTO discovery_queries(organization_id,source_id,name,query_json) VALUES(?,?,?,?)",(user['organization_id'],sid,name,json.dumps(query,sort_keys=True))) + try:cur=db.execute("INSERT INTO discovery_queries(organization_id,source_id,name,query_json,selected_adapters_json,location,category,max_records,daily_limit,schedule,dry_run,lifecycle) VALUES(?,?,?,?,?,?,?,?,?,?,?,?)",(user['organization_id'],sid,name,json.dumps(query,sort_keys=True),json.dumps(selected),location,category,max_records,daily_limit,schedule,int(bool(payload.get('dry_run',False))),str(payload.get('lifecycle','draft')))) except sqlite3.IntegrityError:return self.send_json(409,{"error":"duplicate_query"}) self.audit(db,user,'discovery_query.created',str(cur.lastrowid));db.commit();return self.send_json(201,row_json(db.execute("SELECT * FROM discovery_queries WHERE id=?",(cur.lastrowid,)).fetchone())) def run_query(self,qid,db,user): @@ -1350,10 +1378,27 @@ class ApiHandler(BaseHTTPRequestHandler): except (ValueError,KeyError) as exc:return self.send_json(400,{"error":"invalid_ingest","detail":str(exc)}) inserted=0 for record in page.records[:1000]: - raw=json.dumps(record,sort_keys=True,separators=(',',':'));digest=hashlib.sha256(raw.encode()).hexdigest() - try:db.execute("INSERT INTO source_records(organization_id,source_id,content_hash,raw_json,normalized_json,source_url,query_context_json,cursor_json,rate_policy_json) VALUES(?,?,?,?,?,?,?,?,?)",(user['organization_id'],sid,digest,raw,raw,str(payload.get('source_url','')),json.dumps(payload.get('query_context',{}),sort_keys=True),json.dumps(payload.get('cursor',{}),sort_keys=True),json.dumps(payload.get('rate_policy',{}),sort_keys=True)));inserted+=1 + raw=json.dumps(record,sort_keys=True,separators=(',',':'));digest=hashlib.sha256(raw.encode()).hexdigest(); normalized_key=hashlib.sha256(json.dumps(normalize_record(record),sort_keys=True,separators=(',',':')).encode()).hexdigest() + try: + db.execute("INSERT INTO source_records(organization_id,source_id,content_hash,raw_json,normalized_json,normalized_key,source_url,provenance_json,query_context_json,cursor_json,rate_policy_json) VALUES(?,?,?,?,?,?,?,?,?,?,?)",(user['organization_id'],sid,digest,raw,json.dumps(normalize_record(record),sort_keys=True),normalized_key,str(payload.get('source_url','')),json.dumps({'adapter':source['kind']},sort_keys=True),json.dumps(payload.get('query_context',{}),sort_keys=True),json.dumps(payload.get('cursor',{}),sort_keys=True),json.dumps(payload.get('rate_policy',{}),sort_keys=True))) + record_id=db.execute("SELECT last_insert_rowid()").fetchone()[0]; db.execute("INSERT OR IGNORE INTO enrichment_queue(organization_id,source_record_id) VALUES(?,?)",(user['organization_id'],record_id)); inserted+=1 except sqlite3.IntegrityError:pass db.execute("UPDATE sources SET health_status='healthy',consecutive_failures=0,last_success_at=CURRENT_TIMESTAMP,last_error=NULL WHERE id=?",(sid,));self.audit(db,user,'source.ingested',f'{sid}:{inserted}');db.commit();return self.send_json(201 if inserted else 200,{"inserted":inserted,"records":len(page.records)}) + def source_health(self,sid,db,user): + source=db.execute("SELECT * FROM sources WHERE id=? AND organization_id=?",(sid,user["organization_id"])).fetchone() + if not source:return self.send_json(404,{"error":"not_found"}) + try: config=json.loads(source["config_json"] or "{}") + except (TypeError,ValueError): config={} + health=adapter_for(source["kind"]).health_check(config) + return self.send_json(200,{"id":sid,"source_code":source["source_code"] or source["kind"],"display_name":source["display_name"] or source["name"],"status":source["health_status"],"configured":health.status=="healthy","circuit_open":bool(source["circuit_open"]),"consecutive_failures":source["consecutive_failures"],"last_error":source["last_error"]}) + def discovery_run_action(self,rid,action,db,user): + run=db.execute("SELECT * FROM discovery_runs WHERE id=? AND organization_id=?",(rid,user["organization_id"])).fetchone() + if not run:return self.send_json(404,{"error":"not_found"}) + lifecycle={"pause":"paused","resume":"queued","cancel":"cancelled"}[action] + db.execute("UPDATE discovery_runs SET lifecycle=?,paused_at=CASE WHEN ?='paused' THEN CURRENT_TIMESTAMP ELSE paused_at END,updated_at=CURRENT_TIMESTAMP WHERE id=? AND organization_id=?",(lifecycle,lifecycle,rid,user["organization_id"])) + if action=="cancel": db.execute("UPDATE jobs SET status='cancelled',completed_at=CURRENT_TIMESTAMP,updated_at=CURRENT_TIMESTAMP WHERE id=? AND organization_id=? AND status IN ('queued','running')",(run["job_id"],user["organization_id"])) + self.audit(db,user,"discovery."+action,str(rid)); db.commit() + return self.send_json(200,self._discovery_run_json(db.execute("SELECT * FROM discovery_runs WHERE id=?",(rid,)).fetchone())) def matches(self,bid,db,org): source=self.business(db,bid,org) if not source:return self.send_json(404,{"error":"not_found"}) @@ -1440,6 +1485,24 @@ def _run_scoped_discovery(db, job, handler): handler.add_job_event(db, job["id"], org, "discovery.completed", f"Persisted {len(persisted)} candidates", 100) +def _run_source_discovery(db, job, handler): + payload=json.loads(job["payload"] or "{}"); org=job["organization_id"]; run=db.execute("SELECT * FROM discovery_runs WHERE organization_id=? AND job_id=?",(org,job["id"])).fetchone(); selected=payload.get("selected_adapters") or [] + sources=db.execute("SELECT * FROM sources WHERE organization_id=? AND enabled=1 AND (source_code IN ("+(','.join('?'*len(selected)) or "NULL")+") OR kind IN ("+(','.join('?'*len(selected)) or "NULL")+"))",[org]+list(selected)+list(selected)).fetchall() if selected else [] + total=0 + for source in sources: + try: config=json.loads(source["config_json"] or "{}"); page=adapter_for(source["kind"]).discover(config) + except Exception as exc: + handler.add_job_event(db,job["id"],org,"source.blocked",f"{source['kind']} unavailable",0,"SOURCE_NOT_CONFIGURED"); continue + for record in page.records[:max(0,int(payload.get("max_records",100))-total)]: + raw=json.dumps(record,sort_keys=True,separators=(",",":")); digest=hashlib.sha256(raw.encode()).hexdigest(); norm=json.dumps(normalize_record(record),sort_keys=True); nkey=hashlib.sha256(norm.encode()).hexdigest() + try: + cur=db.execute("INSERT INTO source_records(organization_id,source_id,discovery_run_id,content_hash,raw_json,normalized_json,normalized_key,source_url,provenance_json) VALUES(?,?,?,?,?,?,?,?,?)",(org,source["id"],run["id"] if run else None,digest,raw,norm,nkey,str(config.get("source_url","")),json.dumps({"adapter":source["kind"]}))) + db.execute("INSERT OR IGNORE INTO enrichment_queue(organization_id,source_record_id) VALUES(?,?)",(org,cur.lastrowid)); total+=1 + except sqlite3.IntegrityError: pass + if run: db.execute("UPDATE discovery_runs SET result_count=?,lifecycle='succeeded',updated_at=CURRENT_TIMESTAMP WHERE id=?",(total,run["id"])) + handler.add_job_event(db,job["id"],org,"discovery.completed",f"Persisted {total} source records",100) + + def _job_worker(server): while not server.job_stop.is_set(): db=connect(server.db_path) @@ -1460,6 +1523,13 @@ def _job_worker(server): except Exception as exc: db.execute("UPDATE jobs SET status='failed',error_code=?,completed_at=CURRENT_TIMESTAMP,updated_at=CURRENT_TIMESTAMP WHERE id=?", (str(exc)[:80] or "DISCOVERY_FAILED", jid)); server_handler.add_job_event(db,jid,org,"failed","Discovery failed",job["progress"],str(exc)[:80]); db.commit() continue + if job["type"] == "source_discovery": + try: + _run_source_discovery(db, job, server_handler) + db.execute("UPDATE jobs SET status='succeeded',progress=100,completed_at=CURRENT_TIMESTAMP,updated_at=CURRENT_TIMESTAMP WHERE id=?", (jid,)); db.commit() + except Exception as exc: + db.execute("UPDATE jobs SET status='failed',error_code=?,completed_at=CURRENT_TIMESTAMP,updated_at=CURRENT_TIMESTAMP WHERE id=?", (str(exc)[:80] or "SOURCE_DISCOVERY_FAILED", jid)); server_handler.add_job_event(db,jid,org,"failed","Source discovery failed",job["progress"],str(exc)[:80]); db.commit() + continue try: steps=1 if job["type"]=="noop" else max(1,min(int(payload.get("steps",5)),20)) except (ValueError,TypeError): steps=5 cancelled=False diff --git a/apps/api/app/sources.py b/apps/api/app/sources.py index 0e84f9e..0acd32d 100644 --- a/apps/api/app/sources.py +++ b/apps/api/app/sources.py @@ -1,10 +1,17 @@ -"""Deterministic, network-free discovery source contracts and adapters.""" +"""Safe, tenant-neutral source adapter contracts. + +Network adapters are intentionally capability gated: configuration must explicitly +approve public access, terms, rate limits and credentials (where required). +Adapters never emit or persist credential values. +""" from __future__ import annotations from dataclasses import dataclass, field from typing import Any, Mapping, Protocol, Sequence -import csv, io, re +import csv, io, random, time 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"} + def contains_secret(value: Any, path: str = "") -> str | None: if isinstance(value, Mapping): @@ -38,57 +45,140 @@ class SourceHealth: circuit_open: bool = False last_error: str | None = None +@dataclass(frozen=True) +class NormalizedRecord: + name: str = "" + website: str = "" + email: str = "" + phone: str = "" + description: str = "" + location: str = "" + source_url: str = "" + provenance: str = "" + raw: Mapping[str, Any] = field(default_factory=dict) + class DiscoverySource(Protocol): - kind: str - def validate(self, config: Mapping[str, Any]) -> ValidationResult: ... + source_code: str + display_name: str + def validate_config(self, config: Mapping[str, Any]) -> ValidationResult: ... def discover(self, config: Mapping[str, Any], cursor: str | None = None) -> DiscoveryPage: ... + def health_check(self, config: Mapping[str, Any]) -> SourceHealth: ... -_FIELDS = ("name", "website", "email", "phone", "description") - +_FIELDS = ("name", "website", "email", "phone", "description", "location") 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"} + result = {field: str(row.get(field, "") or "").strip() for field in _FIELDS} + aliases = {"company":"name", "business":"name", "url":"website", "domain":"website", "address":"location", "address_line":"location"} 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}) +def normalized_record(row: Mapping[str, Any], *, source_url="", provenance="") -> NormalizedRecord: + x = normalize_record(row) + return NormalizedRecord(**x, source_url=source_url, provenance=provenance, raw=dict(row)) -class CsvSource: - kind = "csv" - def validate(self, config: Mapping[str, Any]) -> ValidationResult: +def exponential_backoff(attempt: int, base: float = 0.5, maximum: float = 30.0, jitter: float = 0.25) -> float: + """Bounded exponential delay with symmetric jitter; no sleeping occurs here.""" + delay = min(maximum, base * (2 ** max(0, int(attempt)))) + return max(0.0, delay + random.uniform(-jitter * delay, jitter * delay)) + +def backoff_delay(attempt: int, base: float = 0.5, maximum: float = 30.0, jitter: float = 0.25) -> float: + return exponential_backoff(attempt, base, maximum, jitter) + +def circuit_is_open(consecutive_failures: int, threshold: int = 3) -> bool: + return int(consecutive_failures) >= threshold + +def quota_allowed(used: int, limit: int | None) -> bool: + return limit is None or (limit >= 0 and used < limit) + +def rate_limit_delay(last_request: float | None, min_interval: float) -> float: + if last_request is None: return 0.0 + return max(0.0, float(min_interval) - (time.monotonic() - last_request)) + +class _Base: + kind = "" + source_code = "" + display_name = "" + def validate_config(self, config): 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}"]) + return ValidationResult(True) + validate = validate_config + 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): + result = self.validate_config(config) + if not result.valid: raise ValueError(result.errors[0]) + raise RuntimeError("source_not_configured") + +class ManualSource(_Base): + kind = source_code = "manual"; display_name = "Manual records" + def validate_config(self, config): + result = super().validate_config(config) + 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): + 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}) + +class CsvSource(_Base): + kind = source_code = "csv"; display_name = "CSV import" + def validate_config(self, config): + result = super().validate_config(config) + if not result.valid: return result if not isinstance(config.get("csv"), str): return ValidationResult(False, ["csv must be text"]) try: - reader = csv.DictReader(io.StringIO(config["csv"])); + 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]) + def discover(self, config, cursor=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"))) - 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 []}) + return DiscoveryPage([normalize_record({str(k).strip().lower():v for k,v in row.items()}) for row in reader], metadata={"adapter":self.source_code,"columns":reader.fieldnames or []}) -ADAPTERS = {"manual": ManualSource, "csv": CsvSource} +class GatedSource(_Base): + required = "approved" + def validate_config(self, config): + result = super().validate_config(config) + if not result.valid: return result + if config.get("approved") is not True: return ValidationResult(False, ["source approval is required"]) + if config.get("public_access") is not True: return ValidationResult(False, ["public_access approval is required"]) + if config.get("terms_accepted") is not True: return ValidationResult(False, ["terms_accepted is required"]) + if self.source_code in {"google_places", "bing_local"} and not config.get("credential_ref"): + return ValidationResult(False, ["approved credential_ref is required"]) + 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): + 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. + raise RuntimeError("network_adapter_not_configured") + + +def _gated(code, name): + return type(name.replace(" ", ""), (GatedSource,), {"kind":code, "source_code":code, "display_name":name}) +GooglePlacesSource = _gated("google_places", "Google Places") +BingLocalSource = _gated("bing_local", "Bing / approved local API") +ApprovedDirectorySource = _gated("approved_directory", "Approved directory") +PublicWebsiteSource = _gated("public_website", "Public website") +PermittedSocialSource = _gated("permitted_social", "Permitted social") +CtLogsSource = _gated("ct_logs", "Certificate transparency logs") +DnsSource = _gated("dns", "DNS") +RdapSource = _gated("rdap", "RDAP") + +ADAPTERS = {x.source_code: x for x in (ManualSource, CsvSource, GooglePlacesSource, BingLocalSource, ApprovedDirectorySource, PublicWebsiteSource, PermittedSocialSource, CtLogsSource, DnsSource, RdapSource)} +# common aliases used by clients +ADAPTER_REGISTRY = ADAPTERS def adapter_for(kind: str) -> DiscoverySource: - try: return ADAPTERS[kind]() + try: return ADAPTERS[str(kind).strip().lower()]() except KeyError: raise ValueError("unsupported source kind") + +def available_adapters() -> list[dict[str, str]]: + return [{"source_code": cls.source_code, "display_name": cls.display_name} for cls in ADAPTERS.values()] diff --git a/apps/api/schema.sql b/apps/api/schema.sql index f920014..5f3ce6a 100644 --- a/apps/api/schema.sql +++ b/apps/api/schema.sql @@ -136,8 +136,8 @@ CREATE INDEX IF NOT EXISTS idx_job_events_job_sequence ON job_events(job_id,sequ -- Phase 5 source framework (additive-safe; credentials contain metadata only). CREATE TABLE IF NOT EXISTS sources ( id INTEGER PRIMARY KEY AUTOINCREMENT, organization_id TEXT NOT NULL REFERENCES organizations(id), - name TEXT NOT NULL, kind TEXT NOT NULL CHECK(kind IN ('csv','manual')), enabled INTEGER NOT NULL DEFAULT 0, - config_json TEXT NOT NULL DEFAULT '{}', health_status TEXT NOT NULL DEFAULT 'unknown', + name TEXT NOT NULL, kind TEXT NOT NULL CHECK(kind IN ('csv','manual','google_places','bing_local','approved_directory','public_website','permitted_social','ct_logs','dns','rdap')), source_code TEXT NOT NULL DEFAULT '', display_name TEXT NOT NULL DEFAULT '', enabled INTEGER NOT NULL DEFAULT 0, approved INTEGER NOT NULL DEFAULT 0, + config_json TEXT NOT NULL DEFAULT '{}', policy_json TEXT NOT NULL DEFAULT '{}', quota_json TEXT NOT NULL DEFAULT '{}', health_status TEXT NOT NULL DEFAULT 'unknown', consecutive_failures INTEGER NOT NULL DEFAULT 0, circuit_open INTEGER NOT NULL DEFAULT 0, last_success_at TEXT, last_failure_at TEXT, last_error TEXT, created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, UNIQUE(organization_id,name) @@ -151,18 +151,25 @@ CREATE TABLE IF NOT EXISTS source_credentials ( ); CREATE TABLE IF NOT EXISTS discovery_queries ( id INTEGER PRIMARY KEY AUTOINCREMENT, organization_id TEXT NOT NULL REFERENCES organizations(id), source_id INTEGER NOT NULL REFERENCES sources(id) ON DELETE CASCADE, - name TEXT NOT NULL, query_json TEXT NOT NULL DEFAULT '{}', enabled INTEGER NOT NULL DEFAULT 1, created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, + name TEXT NOT NULL, query_json TEXT NOT NULL DEFAULT '{}', enabled INTEGER NOT NULL DEFAULT 1, selected_adapters_json TEXT NOT NULL DEFAULT '[]', location TEXT NOT NULL DEFAULT '', category TEXT NOT NULL DEFAULT '', max_records INTEGER NOT NULL DEFAULT 100, daily_limit INTEGER NOT NULL DEFAULT 1000, schedule TEXT NOT NULL DEFAULT '', dry_run INTEGER NOT NULL DEFAULT 0, lifecycle TEXT NOT NULL DEFAULT 'draft', created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, UNIQUE(organization_id,name) ); CREATE INDEX IF NOT EXISTS idx_discovery_queries_org ON discovery_queries(organization_id,id); CREATE TABLE IF NOT EXISTS source_records ( id INTEGER PRIMARY KEY AUTOINCREMENT, organization_id TEXT NOT NULL REFERENCES organizations(id), source_id INTEGER NOT NULL REFERENCES sources(id) ON DELETE CASCADE, - discovery_query_id INTEGER REFERENCES discovery_queries(id) ON DELETE SET NULL, content_hash TEXT NOT NULL, raw_json TEXT NOT NULL, - normalized_json TEXT NOT NULL, source_url TEXT NOT NULL DEFAULT '', query_context_json TEXT NOT NULL DEFAULT '{}', processing_status TEXT NOT NULL DEFAULT 'raw', + discovery_query_id INTEGER REFERENCES discovery_queries(id) ON DELETE SET NULL, discovery_run_id INTEGER REFERENCES discovery_runs(id) ON DELETE SET NULL, content_hash TEXT NOT NULL, raw_json TEXT NOT NULL, + normalized_json TEXT NOT NULL, normalized_key TEXT NOT NULL DEFAULT '', source_url TEXT NOT NULL DEFAULT '', provenance_json TEXT NOT NULL DEFAULT '{}', query_context_json TEXT NOT NULL DEFAULT '{}', processing_status TEXT NOT NULL DEFAULT 'raw', cursor_json TEXT NOT NULL DEFAULT '{}', rate_policy_json TEXT NOT NULL DEFAULT '{}', created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, UNIQUE(organization_id,source_id,content_hash) ); CREATE INDEX IF NOT EXISTS idx_source_records_org ON source_records(organization_id,id DESC); +CREATE TABLE IF NOT EXISTS enrichment_queue ( + id INTEGER PRIMARY KEY AUTOINCREMENT, organization_id TEXT NOT NULL REFERENCES organizations(id) ON DELETE CASCADE, source_record_id INTEGER NOT NULL REFERENCES source_records(id) ON DELETE CASCADE, status TEXT NOT NULL DEFAULT 'queued', attempts INTEGER NOT NULL DEFAULT 0, created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, UNIQUE(organization_id,source_record_id) +); +CREATE TABLE IF NOT EXISTS source_events ( + id INTEGER PRIMARY KEY AUTOINCREMENT, organization_id TEXT NOT NULL REFERENCES organizations(id) ON DELETE CASCADE, source_id INTEGER REFERENCES sources(id) ON DELETE SET NULL, discovery_run_id INTEGER REFERENCES discovery_runs(id) ON DELETE SET NULL, event_type TEXT NOT NULL, message TEXT NOT NULL DEFAULT '', metadata_json TEXT NOT NULL DEFAULT '{}', created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP +); +CREATE INDEX IF NOT EXISTS idx_source_events_org ON source_events(organization_id,created_at DESC,id DESC); -- Phase 7 domain intelligence (additive-safe; DNS results are explicitly stateful). CREATE TABLE IF NOT EXISTS domain_checks ( @@ -341,9 +348,8 @@ CREATE INDEX IF NOT EXISTS idx_outreach_drafts_business ON outreach_drafts(organ -- Built-in scoped discovery runs. Results are reviewable business records only; -- this feature never sends outreach. CREATE TABLE IF NOT EXISTS discovery_runs ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - organization_id TEXT NOT NULL REFERENCES organizations(id) ON DELETE CASCADE, - job_id INTEGER NOT NULL REFERENCES jobs(id) ON DELETE CASCADE, + id INTEGER PRIMARY KEY AUTOINCREMENT, organization_id TEXT NOT NULL REFERENCES organizations(id), job_id INTEGER NOT NULL REFERENCES jobs(id) ON DELETE CASCADE, + selected_adapters_json TEXT NOT NULL DEFAULT '[]', location TEXT NOT NULL DEFAULT '', category TEXT NOT NULL DEFAULT '', max_records INTEGER NOT NULL DEFAULT 100, daily_limit INTEGER NOT NULL DEFAULT 1000, schedule TEXT NOT NULL DEFAULT '', dry_run INTEGER NOT NULL DEFAULT 0, lifecycle TEXT NOT NULL DEFAULT 'draft', paused_at TEXT, criteria_json TEXT NOT NULL DEFAULT '{}', seed_urls_json TEXT NOT NULL DEFAULT '[]', result_json TEXT NOT NULL DEFAULT '{}', result_count INTEGER NOT NULL DEFAULT 0, created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,