root/MarketingTool
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add policy-aware source adapter framework

Browse Source
This commit is contained in:
Marco0300
2026-09-02 18:58:10 +02:00
parent cf034288e6
commit 46cc1f6182
13 changed files with 359 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/api/app/main.py
+80 -2
View File
@@ -8,14 +8,16 @@ 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
from app.sources import adapter_for, contains_secret
else:
from .domain import deduplication_key, deduplicate_businesses, is_suppressed, normalize_business, score_business, normalize_domain, normalize_phone
from .sources import adapter_for, contains_secret
ORGANIZATION_ID = "demo-tenant"
SCHEMA = Path(__file__).resolve().parents[1] / "schema.sql"
SESSION_DAYS = 7
PBKDF2_ITERATIONS = 300_000
MUTATING_ROLES = {"owner", "admin", "researcher"}
JOB_TYPES = {"noop", "prospect_recalculate"}
JOB_TYPES = {"noop", "prospect_recalculate", "source_discovery"}
JOB_PAGE_SIZE = 100
SECRET_KEYS = {"password", "passwd", "secret", "token", "api_key", "apikey", "authorization", "credential", "private_key"}
CHILD_TABLES = {"contacts": ("name", "email", "phone", "title", "do_not_contact"), "domains": ("domain", "kind"), "websites": ("url", "website_class"), "evidence": ("kind", "url", "claim"), "notes": ("body",)}
@@ -110,6 +112,9 @@ class ApiHandler(BaseHTTPRequestHandler):
if path=="/api/v1/dashboard/summary":
row=db.execute("SELECT COUNT(*) businesses,COALESCE(AVG(score),0) average_score FROM businesses WHERE organization_id=?",(org,)).fetchone(); return self.send_json(200,{"organization_id":org,"businesses":row["businesses"],"average_score":round(row["average_score"],2),"suppressed":db.execute("SELECT COUNT(*) FROM suppressions WHERE organization_id=?",(org,)).fetchone()[0]})
if path=="/api/v1/businesses": return self.list_businesses(db,org,parse_qs(parsed.query))
if path=="/api/v1/sources": return self.list_sources(db,org)
if path=="/api/v1/discovery-queries": return self.list_queries(db,org)
if path=="/api/v1/source-records": return self.list_source_records(db,org,parse_qs(parsed.query))
if path=="/api/v1/jobs": return self.list_jobs(db,org,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/businesses/"):
@@ -153,7 +158,7 @@ class ApiHandler(BaseHTTPRequestHandler):
try:
cur=db.execute("INSERT INTO jobs(organization_id,idempotency_key,type,payload,max_attempts) VALUES(?,?,?,?,?)",(user["organization_id"],key,kind,safe,max_attempts)); jid=cur.lastrowid
self.add_job_event(db,jid,user["organization_id"],"queued","Job queued",0); self.audit(db,user,"job.created",str(jid)); db.commit(); getattr(self.server,"job_wakeup",threading.Event()).set()
return self.send_json(201,job_json(db.execute("SELECT * FROM jobs WHERE id=?",(jid,)).fetchone()))
return self.send_json(202 if payload.get("_accepted") else 201,job_json(db.execute("SELECT * FROM jobs WHERE id=?",(jid,)).fetchone()))
except sqlite3.IntegrityError:
row=db.execute("SELECT * FROM jobs WHERE organization_id=? AND idempotency_key=?",(user["organization_id"],key)).fetchone(); return self.send_json(200,job_json(row))
@@ -210,9 +215,14 @@ class ApiHandler(BaseHTTPRequestHandler):
if path.startswith("/api/v1/jobs/"):
return self.job_action(db,user,path)
if path=="/api/v1/businesses":return self.create_business(payload,db,user)
if path=="/api/v1/sources":return self.create_source(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/imports/preview":return self.preview_import(payload,db,org)
bits=path.split("/")
if len(bits)==6 and bits[3] == "sources" and bits[4].isdigit() and bits[5] in {"test","ingest"}:
return self.test_source(int(bits[4]),db,user) if bits[5]=="test" else self.ingest_source(int(bits[4]),payload,db,user)
if len(bits)==6 and bits[3] == "discovery-queries" and bits[4].isdigit() and bits[5]=="run": return self.run_query(int(bits[4]),db,user)
if len(bits)==7 and bits[:4]==["","api","v1","businesses"] and bits[5] in CHILD_TABLES and bits[6]=="": pass
if len(bits)==6 and bits[:4]==["","api","v1","businesses"] and bits[5] in CHILD_TABLES:return self.create_child(int(bits[4]) if bits[4].isdigit() else -1,bits[5],payload,db,user)
if len(bits)==6 and bits[:4]==["","api","v1","businesses"] and bits[5]=="verify":return self.verify_business(int(bits[4]) if bits[4].isdigit() else -1,payload,db,user)
@@ -225,6 +235,7 @@ class ApiHandler(BaseHTTPRequestHandler):
if not user:return
if user["role"] not in MUTATING_ROLES:return self.send_json(403,{"error":"forbidden"})
bits=path.split("/")
if len(bits)==5 and bits[:4]==["","api","v1","sources"] and bits[4].isdigit(): return self.update_source(int(bits[4]),self.read_json(),db,user)
if len(bits)==6 and bits[:4]==["","api","v1","businesses"] and bits[5]=="pipeline":return self.update_pipeline(int(bits[4]) if bits[4].isdigit() else -1,self.read_json(),db,user)
return self.send_json(404,{"error":"not_found"})
finally:db.close()
@@ -286,6 +297,73 @@ class ApiHandler(BaseHTTPRequestHandler):
elif key in existing_keys or key in seen:continue
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'
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,))]})
def list_source_records(self,db,org,q):
try:
limit=int(q.get('page_size',[50])[0]); offset=max(0,int(q.get('offset',[0])[0]))
if limit<1 or limit>100: raise ValueError
except (ValueError,TypeError): return self.send_json(400,{"error":"invalid_pagination"})
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'):
try:x[k]=json.loads(x[k])
except (ValueError,TypeError):pass
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"})
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)))
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()))
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',{})
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)))
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)
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"})
try: result=adapter_for(source['kind']).validate(json.loads(source['config_json'])); ok=result.valid; error='; '.join(result.errors) if not ok else None
except Exception as exc:ok=False;error=str(exc)[:300]
if ok:db.execute("UPDATE sources SET health_status='healthy',consecutive_failures=0,circuit_open=0,last_success_at=CURRENT_TIMESTAMP,last_error=NULL WHERE id=?",(sid,));action='source.test.succeeded'
else:db.execute("UPDATE sources SET health_status='unhealthy',consecutive_failures=consecutive_failures+1,circuit_open=CASE WHEN consecutive_failures+1>=3 THEN 1 ELSE circuit_open END,last_failure_at=CURRENT_TIMESTAMP,last_error=? WHERE id=?",(error,sid));action='source.test.failed'
self.audit(db,user,action,str(sid));db.commit();return self.send_json(200,{"ok":ok,"errors":[] if ok else [error]})
def ingest_source(self,sid,payload,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"})
if not source['enabled'] or source['circuit_open']:return self.send_json(409,{"error":"source_disabled"})
if contains_secret(payload):return self.send_json(400,{"error":"secret_not_permitted"})
config={k:v for k,v in payload.items() if k not in ('source_url','query_context','cursor','rate_policy')}
if len(json.dumps(config).encode())>5*1024*1024:return self.send_json(400,{"error":"ingest_limits"})
if isinstance(config.get('rows'),list) and (len(config['rows'])>1000 or any(not isinstance(r,dict) or len(r)>50 or any(len(str(v))>10000 for v in r.values()) for r in config['rows'])):return self.send_json(400,{"error":"ingest_limits"})
if isinstance(config.get('csv'),str) and config['csv'].count('\n')>1001:return self.send_json(400,{"error":"ingest_limits"})
try:page=adapter_for(source['kind']).discover(config)
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
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 log_message(self,*_):pass
def _job_worker(server):