complete source discovery processing lifecycle
This commit is contained in:
Marco0300
+75
-15
@@ -78,7 +78,7 @@ def connect(db_path: str) -> sqlite3.Connection:
|
||||
"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 '{}'")),
|
||||
"source_records": (("discovery_run_id", "INTEGER"), ("normalized_key", "TEXT NOT NULL DEFAULT ''"), ("provenance_json", "TEXT NOT NULL DEFAULT '{}'"), ("response_metadata_json", "TEXT NOT NULL DEFAULT '{}'")),
|
||||
}.items():
|
||||
existing = {r[1] for r in db.execute(f"PRAGMA table_info({table})")}
|
||||
for col, definition in additions:
|
||||
@@ -1322,7 +1322,7 @@ class ApiHandler(BaseHTTPRequestHandler):
|
||||
rows=db.execute("SELECT * FROM source_records WHERE organization_id=? ORDER BY id DESC LIMIT ? OFFSET ?",(org,limit+1,offset)).fetchall(); out=[]
|
||||
for r in rows[:limit]:
|
||||
x=row_json(r)
|
||||
for k in ('raw_json','normalized_json','query_context_json','cursor_json','rate_policy_json'):
|
||||
for k in ('raw_json','normalized_json','query_context_json','response_metadata_json','cursor_json','rate_policy_json'):
|
||||
try:x[k]=json.loads(x[k])
|
||||
except (ValueError,TypeError):pass
|
||||
out.append(x)
|
||||
@@ -1334,7 +1334,9 @@ class ApiHandler(BaseHTTPRequestHandler):
|
||||
if contains_secret(config):return self.send_json(400,{"error":"secret_not_permitted"})
|
||||
try:
|
||||
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})
|
||||
# Registration may precede the actual local payload or optional provider
|
||||
# approval. Discovery/ingest still validates the effective configuration.
|
||||
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,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,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()))
|
||||
@@ -1355,8 +1357,14 @@ class ApiHandler(BaseHTTPRequestHandler):
|
||||
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):
|
||||
if not db.execute("SELECT id FROM discovery_queries WHERE id=? AND organization_id=?",(qid,user['organization_id'])).fetchone():return self.send_json(404,{"error":"not_found"})
|
||||
return self.create_job({"type":"source_discovery","_accepted":True,"payload":{"discovery_query_id":qid},"idempotency_key":f"discovery-query-{qid}-{int(time.time())}"},db,user)
|
||||
query=db.execute("SELECT * FROM discovery_queries WHERE id=? AND organization_id=?",(qid,user["organization_id"])).fetchone()
|
||||
if not query:return self.send_json(404,{"error":"not_found"})
|
||||
key=f"discovery-query-{qid}-{int(time.time())}"
|
||||
result=self.create_job({"type":"source_discovery","_accepted":True,"payload":{"discovery_query_id":qid,"selected_adapters":json.loads(query["selected_adapters_json"] or "[]"),"max_records":query["max_records"],"daily_limit":query["daily_limit"]},"idempotency_key":key,"_defer_wakeup":True},db,user)
|
||||
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,selected_adapters_json,location,category,max_records,daily_limit,schedule,dry_run,lifecycle,criteria_json,seed_urls_json) VALUES(?,?,?,?,?,?,?,?,?,?,?,?)",(user["organization_id"],job["id"],query["selected_adapters_json"],query["location"],query["category"],query["max_records"],query["daily_limit"],query["schedule"],query["dry_run"],"queued",query["query_json"],"[]")); db.commit()
|
||||
getattr(self.server,"job_wakeup",threading.Event()).set(); return result
|
||||
def test_source(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"})
|
||||
@@ -1486,19 +1494,66 @@ def _run_scoped_discovery(db, job, handler):
|
||||
|
||||
|
||||
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
|
||||
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 []
|
||||
query=None
|
||||
if payload.get("discovery_query_id"):
|
||||
query=db.execute("SELECT * FROM discovery_queries WHERE id=? AND organization_id=?",(payload["discovery_query_id"],org)).fetchone()
|
||||
if query:
|
||||
selected=json.loads(query["selected_adapters_json"] or "[]")
|
||||
if not selected:
|
||||
linked=db.execute("SELECT kind FROM sources WHERE id=? AND organization_id=?",(query["source_id"],org)).fetchone()
|
||||
selected=[linked["kind"]] if linked else []
|
||||
placeholders=','.join('?'*len(selected)) or "NULL"
|
||||
sources=db.execute("SELECT * FROM sources WHERE organization_id=? AND enabled=1 AND circuit_open=0 AND (source_code IN ("+placeholders+") OR kind IN ("+placeholders+"))",[org]+list(selected)+list(selected)).fetchall() if selected else []
|
||||
total=0; max_records=max(0,int(payload.get("max_records", query["max_records"] if query else 100)))
|
||||
for source in sources:
|
||||
try: config=json.loads(source["config_json"] or "{}"); page=adapter_for(source["kind"]).discover(config)
|
||||
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["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()
|
||||
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"]))
|
||||
handler.add_job_event(db,job["id"],org,"source.blocked",f"{source['kind']} unavailable",10,"SOURCE_NOT_CONFIGURED"); continue
|
||||
limit=min(max_records-total, per_run_limit, max(0,daily_limit-used_today))
|
||||
for record in page.records[:limit]:
|
||||
raw=json.dumps(record,sort_keys=True,separators=(",",":")); digest=hashlib.sha256(raw.encode()).hexdigest(); normalized=normalize_business(normalize_record(record)); norm=json.dumps(normalized,sort_keys=True); nkey=hashlib.sha256(norm.encode()).hexdigest()
|
||||
handler.add_job_event(db,job["id"],org,"source.raw_persisted",f"Persisting {source['kind']} record",20)
|
||||
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
|
||||
cur=db.execute("INSERT INTO source_records(organization_id,source_id,discovery_query_id,discovery_run_id,content_hash,raw_json,normalized_json,normalized_key,source_url,provenance_json,query_context_json,response_metadata_json,processing_status) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?)",(org,source["id"],query["id"] if query else None,run["id"] if run else None,digest,raw,norm,nkey,str(config.get("source_url","")),json.dumps({"adapter":source["kind"],"metadata":page.metadata},sort_keys=True),json.dumps(payload.get("criteria",{}),sort_keys=True),json.dumps(page.metadata,sort_keys=True),"raw"))
|
||||
record_id=cur.lastrowid; total+=1
|
||||
except sqlite3.IntegrityError:
|
||||
existing=db.execute("SELECT id,processing_status FROM source_records WHERE organization_id=? AND source_id=? AND content_hash=?",(org,source["id"],digest)).fetchone()
|
||||
if existing and existing["processing_status"] in ("processed","matched"): continue
|
||||
record_id=existing["id"] if existing else None
|
||||
handler.add_job_event(db,job["id"],org,"source.normalized",f"Normalized {normalized.get('name','')}",35)
|
||||
suppressed=is_suppressed(normalized,[dict(r) for r in db.execute("SELECT kind,value FROM suppressions WHERE organization_id=? AND active=1",(org,))])
|
||||
if suppressed:
|
||||
if record_id: db.execute("UPDATE source_records SET processing_status='skipped' WHERE id=? AND organization_id=?",(record_id,org))
|
||||
handler.add_job_event(db,job["id"],org,"review.skipped",f"Suppressed source record {record_id}",40,"SUPPRESSED"); continue
|
||||
existing=db.execute("SELECT * FROM businesses WHERE organization_id=? AND ((website_domain<>'' AND website_domain=?) OR (email<>'' AND email=?) OR (phone<>'' AND phone=?) OR (name=? AND city=?)) ORDER BY id LIMIT 1",(org,normalized["website_domain"],normalized["email"],normalized["phone"],normalized["name"],normalized["city"])).fetchone()
|
||||
if existing: bid=existing["id"]; handler.add_job_event(db,job["id"],org,"business.matched",f"Matched business {bid}",50)
|
||||
else:
|
||||
scored=score_business(normalized); cur=db.execute("INSERT INTO businesses(organization_id,name,website,website_domain,email,phone,description,province,city,suburb,score,score_version,score_factors,website_class,review_status) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",(org,normalized["name"],normalized["website"],normalized["website_domain"],normalized["email"],normalized["phone"],normalized.get("description",""),normalized["province"],normalized["city"],normalized["suburb"],scored["score"],scored["score_version"],json.dumps(scored["factors"]),scored["website_class"],"pending")); bid=cur.lastrowid
|
||||
db.execute("INSERT INTO score_history(organization_id,business_id,score,eligible,priority_band,score_version,explanations_json,signals_json) VALUES(?,?,?,?,?,?,?,?)",(org,bid,scored["score"],1,"high" if scored["score"]>=70 else "medium" if scored["score"]>=40 else "low",scored["score_version"],json.dumps(scored["factors"]),json.dumps({"source":source["kind"]})))
|
||||
handler.add_job_event(db,job["id"],org,"business.created",f"Created business {bid}",60)
|
||||
if normalized["website_domain"] and not db.execute("SELECT 1 FROM domains WHERE organization_id=? AND business_id=? AND domain=?",(org,bid,normalized["website_domain"])).fetchone(): db.execute("INSERT INTO domains(business_id,organization_id,domain,kind) VALUES(?,?,?,?)",(bid,org,normalized["website_domain"],"website"))
|
||||
if normalized["website"] and not db.execute("SELECT 1 FROM websites WHERE organization_id=? AND business_id=? AND url=?",(org,bid,normalized["website"])).fetchone(): db.execute("INSERT INTO websites(business_id,organization_id,url,website_class) VALUES(?,?,?,?)",(bid,org,normalized["website"],"business_site"))
|
||||
if normalized["website"] or normalized["website_domain"]: db.execute("INSERT OR IGNORE INTO evidence(business_id,organization_id,kind,url,claim) VALUES(?,?,?,?,?)",(bid,org,"source_record",normalized["website"],"Discovered by "+source["kind"]))
|
||||
if normalized["email"] or normalized["phone"]:
|
||||
if not db.execute("SELECT 1 FROM contacts WHERE organization_id=? AND business_id=? AND email=? AND phone=?",(org,bid,normalized["email"],normalized["phone"])).fetchone(): db.execute("INSERT INTO contacts(business_id,organization_id,email,phone) VALUES(?,?,?,?)",(bid,org,normalized["email"],normalized["phone"]))
|
||||
if record_id: db.execute("UPDATE source_records SET processing_status='processed',normalized_json=?,normalized_key=? WHERE id=? AND organization_id=?",(norm,nkey,record_id,org)); db.execute("INSERT OR IGNORE INTO enrichment_queue(organization_id,source_record_id,status) VALUES(?,?,?)",(org,record_id,"completed")); db.execute("UPDATE enrichment_queue SET status='completed',updated_at=CURRENT_TIMESTAMP WHERE source_record_id=? AND organization_id=?",(record_id,org))
|
||||
handler.add_job_event(db,job["id"],org,"enrichment.queued",f"Enrichment complete for business {bid}",75); handler.add_job_event(db,job["id"],org,"review.queued",f"Business {bid} queued for review",90)
|
||||
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)
|
||||
|
||||
@@ -1525,6 +1580,11 @@ def _job_worker(server):
|
||||
continue
|
||||
if job["type"] == "source_discovery":
|
||||
try:
|
||||
run_state=db.execute("SELECT lifecycle FROM discovery_runs WHERE organization_id=? AND job_id=?",(org,jid)).fetchone()
|
||||
if run_state and run_state["lifecycle"] == "cancelled":
|
||||
db.execute("UPDATE jobs SET status='cancelled',completed_at=CURRENT_TIMESTAMP,updated_at=CURRENT_TIMESTAMP WHERE id=?",(jid,)); server_handler.add_job_event(db,jid,org,"cancelled","Discovery cancelled",job["progress"]); db.commit(); continue
|
||||
if run_state and run_state["lifecycle"] == "paused":
|
||||
db.execute("UPDATE jobs SET status='queued',updated_at=CURRENT_TIMESTAMP WHERE id=?",(jid,)); server_handler.add_job_event(db,jid,org,"paused","Discovery paused",job["progress"]); db.commit(); continue
|
||||
_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:
|
||||
|
||||
Reference in New Issue
Block a user