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

add reversible prospect deduplication

Browse Source
This commit is contained in:
Marco0300
2026-09-03 08:46:22 +02:00
parent 46cc1f6182
commit 25ee7931ab
14 changed files with 290 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/api/app/main.py
+59 -4
View File
@@ -7,10 +7,10 @@ from pathlib import Path
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.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
else:
from .domain import deduplication_key, deduplicate_businesses, is_suppressed, normalize_business, score_business, normalize_domain, normalize_phone
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
ORGANIZATION_ID = "demo-tenant"
SCHEMA = Path(__file__).resolve().parents[1] / "schema.sql"
@@ -44,7 +44,7 @@ def connect(db_path: str) -> sqlite3.Connection:
db = sqlite3.connect(db_path); db.row_factory = sqlite3.Row; db.execute("PRAGMA foreign_keys = ON"); db.executescript(SCHEMA.read_text())
# Upgrade databases created by Phase 1/2 without destroying data.
cols = {r[1] for r in db.execute("PRAGMA table_info(businesses)")}
for col, definition in (("verified", "INTEGER NOT NULL DEFAULT 0"), ("verified_at", "TEXT"), ("updated_at", "TEXT")):
for col, definition in (("verified", "INTEGER NOT NULL DEFAULT 0"), ("verified_at", "TEXT"), ("updated_at", "TEXT"), ("province", "TEXT NOT NULL DEFAULT ''"), ("city", "TEXT NOT NULL DEFAULT ''"), ("suburb", "TEXT NOT NULL DEFAULT ''"), ("merge_status", "TEXT NOT NULL DEFAULT 'active'"), ("merged_into_id", "INTEGER")):
if col not in cols: db.execute(f"ALTER TABLE businesses ADD COLUMN {col} {definition}")
db.execute("UPDATE businesses SET updated_at=COALESCE(updated_at,created_at) WHERE updated_at IS NULL")
db.execute("INSERT OR IGNORE INTO organizations (id,name) VALUES (?,?)", (ORGANIZATION_ID, "Demo organization"))
@@ -112,6 +112,7 @@ 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/merge-history": return self.list_merge_history(db,org)
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))
@@ -122,6 +123,7 @@ class ApiHandler(BaseHTTPRequestHandler):
if not ident.isdigit(): return self.send_json(404,{"error":"not_found"})
row=self.business(db,int(ident),org)
if not row:return self.send_json(404,{"error":"not_found"})
if len(bits)==6 and bits[5]=="matches": return self.matches(int(ident),db,org)
payload=row_json(row); payload.update(self.nested(db,int(ident),org)); return self.send_json(200,payload)
return self.send_json(404,{"error":"not_found"})
finally: db.close()
@@ -219,6 +221,9 @@ class ApiHandler(BaseHTTPRequestHandler):
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)
if path.startswith("/api/v1/merge-history/") and path.endswith("/reverse"):
ident=path.split("/")[4]
return self.reverse_merge(int(ident) if ident.isdigit() else -1,db,user)
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)
@@ -226,6 +231,7 @@ class ApiHandler(BaseHTTPRequestHandler):
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)
if len(bits)==6 and bits[:4]==["","api","v1","businesses"] and bits[5]=="merge":return self.merge_business(int(bits[4]) if bits[4].isdigit() else -1,payload,db,user)
return self.send_json(404,{"error":"not_found"})
finally:db.close()
def do_PATCH(self):
@@ -252,7 +258,7 @@ class ApiHandler(BaseHTTPRequestHandler):
if is_suppressed(b,suppressions):return self.send_json(409,{"error":"suppressed"})
fields=[(c,b[c]) for c in ("website_domain","email","phone") if b[c]]
if fields and db.execute("SELECT id FROM businesses WHERE organization_id=? AND ("+" OR ".join(f"{c}=?" for c,_ in fields)+")",[org]+[v for _,v in fields]).fetchone():return self.send_json(409,{"error":"duplicate"})
scored=score_business(b);cur=db.execute("INSERT INTO businesses(organization_id,name,website,website_domain,email,phone,description,score,score_version,score_factors,website_class) VALUES(?,?,?,?,?,?,?,?,?,?,?)",(org,b["name"],b["website"],b["website_domain"],b["email"],b["phone"],str(b.get("description","")),scored["score"],scored["score_version"],json.dumps(scored["factors"]),scored["website_class"])); self.audit(db,user,"business.created",str(cur.lastrowid));db.commit();return self.send_json(201,row_json(db.execute("SELECT * FROM businesses WHERE id=?",(cur.lastrowid,)).fetchone()))
scored=score_business(b);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) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?)",(org,b["name"],b["website"],b["website_domain"],b["email"],b["phone"],str(b.get("description","")),b["province"],b["city"],b["suburb"],scored["score"],scored["score_version"],json.dumps(scored["factors"]),scored["website_class"])); self.audit(db,user,"business.created",str(cur.lastrowid));db.commit();return self.send_json(201,row_json(db.execute("SELECT * FROM businesses WHERE id=?",(cur.lastrowid,)).fetchone()))
def create_suppression(self,payload,db,user):
kind,value=payload.get("kind"),str(payload.get("value","")).strip().lower()
if kind not in {"email","domain","phone"} or not value:return self.send_json(400,{"error":"invalid_suppression"})
@@ -364,6 +370,55 @@ class ApiHandler(BaseHTTPRequestHandler):
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 matches(self,bid,db,org):
source=self.business(db,bid,org)
if not source:return self.send_json(404,{"error":"not_found"})
try: threshold=float(parse_qs(urlparse(self.path).query).get("threshold",["0.72"])[0])
except ValueError:return self.send_json(400,{"error":"invalid_threshold"})
rows=[row_json(r) for r in db.execute("SELECT * FROM businesses WHERE organization_id=? AND id<>? AND merge_status='active' ORDER BY id",(org,bid))]
return self.send_json(200,{"business_id":bid,"threshold":threshold,"items":match_businesses(row_json(source),rows,threshold)})
def list_merge_history(self,db,org):
rows=[row_json(r) for r in db.execute("SELECT * FROM merge_history WHERE organization_id=? ORDER BY id DESC",(org,))]
for row in rows:
for key in ("source_snapshot_json","child_reassignment_json"):
try: row[key]=json.loads(row[key])
except (TypeError,ValueError): pass
return self.send_json(200,{"organization_id":org,"items":rows})
def merge_business(self,bid,payload,db,user):
org=user["organization_id"]; target_id=payload.get("target_business_id",payload.get("target_id"))
if not isinstance(target_id,int) or target_id==bid:return self.send_json(400,{"error":"target_required"})
source=self.business(db,bid,org); target=self.business(db,target_id,org)
if not source or not target:return self.send_json(404,{"error":"not_found"})
if source["merge_status"] != "active":return self.send_json(409,{"error":"source_already_merged"})
snapshot={"business":row_json(source),"children":{}}
child_meta={}
for table in ("business_identifiers","contacts","domains","websites","evidence","pipeline_entries","interactions","notes"):
rows=[row_json(r) for r in db.execute(f"SELECT * FROM {table} WHERE business_id=? AND organization_id=? ORDER BY id",(bid,org))]
snapshot["children"][table]=rows; child_meta[table]={"ids":[r["id"] for r in rows],"count":len(rows)}
if rows: db.execute(f"UPDATE {table} SET business_id=? WHERE business_id=? AND organization_id=?",(target_id,bid,org))
cur=db.execute("INSERT INTO merge_history(organization_id,source_business_id,target_business_id,source_snapshot_json,child_reassignment_json,actor_user_id) VALUES(?,?,?,?,?,?)",(org,bid,target_id,json.dumps(snapshot,sort_keys=True),json.dumps(child_meta,sort_keys=True),user["id"]))
db.execute("UPDATE businesses SET merge_status='merged',merged_into_id=?,updated_at=CURRENT_TIMESTAMP WHERE id=? AND organization_id=?",(target_id,bid,org))
self.audit(db,user,"business.merged",f"{bid}->{target_id}");db.commit()
return self.send_json(200,{"merge_history_id":cur.lastrowid,"source_business_id":bid,"target_business_id":target_id,"status":"merged","reassigned":child_meta})
def reverse_merge(self,hid,db,user):
row=db.execute("SELECT * FROM merge_history WHERE id=? AND organization_id=?",(hid,user["organization_id"])).fetchone()
if not row:return self.send_json(404,{"error":"not_found"})
if not row["reversible"]:return self.send_json(409,{"error":"merge_not_reversible"})
source=self.business(db,row["source_business_id"],user["organization_id"]); target=self.business(db,row["target_business_id"],user["organization_id"])
if not source or not target:return self.send_json(409,{"error":"business_missing"})
snapshot=json.loads(row["source_snapshot_json"]); ids=json.loads(row["child_reassignment_json"])
for table, meta in ids.items():
if not meta.get("ids"):continue
marks=",".join("?" for _ in meta["ids"])
db.execute(f"UPDATE {table} SET business_id=? WHERE business_id=? AND organization_id=? AND id IN ({marks})",[source["id"],target["id"],user["organization_id"]]+meta["ids"])
db.execute("UPDATE businesses SET merge_status='active',merged_into_id=NULL,updated_at=CURRENT_TIMESTAMP WHERE id=? AND organization_id=?",(source["id"],user["organization_id"]))
db.execute("UPDATE merge_history SET reversible=0,reversed_at=CURRENT_TIMESTAMP WHERE id=? AND organization_id=?",(hid,user["organization_id"]))
self.audit(db,user,"business.merge_reversed",str(hid));db.commit()
return self.send_json(200,{"id":hid,"status":"reversed","source_business_id":source["id"],"target_business_id":target["id"]})
def log_message(self,*_):pass
def _job_worker(server):