- ${reasons.map(reason => `
- ${esc(typeof reason === 'string' ? reason : reason.text || reason.description || JSON.stringify(reason))} `).join('')}
diff --git a/README.md b/README.md index 99dc7b3..4227039 100644 --- a/README.md +++ b/README.md @@ -1,12 +1,12 @@ # Prospect Intelligence Platform -A safety-first Phase 5 design/implementation boundary for **manual**, evidence-led prospect qualification and controlled source ingestion. The current runtime remains a manual vertical slice: it stores tenant-owned businesses and child intelligence records, keeps provenance with each evidence item, supports a review pipeline, and records operationally relevant changes. Phase 5 defines source adapters, discovery-query records, raw-source retention, and health controls; it does **not** enable network discovery. **Automated outreach is disabled, and no live source may be enabled without explicit approval.** +A safety-first Phase 6 design/implementation boundary for **manual**, evidence-led prospect qualification and controlled source ingestion. The current runtime remains a manual vertical slice: it stores tenant-owned businesses and child intelligence records, keeps provenance with each evidence item, supports a review pipeline, and records operationally relevant changes. Phase 6 defines deterministic South African normalization and deduplication review semantics in addition to the Phase 5 source controls; it does **not** enable network discovery. **Automated outreach is disabled, and no live source may be enabled without explicit approval.** ## Included - Dependency-free Python/SQLite API under `apps/api`. - Tenant-scoped business detail APIs with child intelligence/evidence records, provenance fields, notes, pipeline state, and audit history. -- Server-side normalization, conservative website classification, exact deduplication, versioned scoring, and suppression checks. +- Server-side normalization, conservative website classification, exact deduplication, versioned scoring, and suppression checks. Phase 6 documents the SA phone/location canonical forms and the review-only fuzzy-match contract. - Bounded list pagination and server-side filters so a tenant cannot request an unbounded prospect collection. - Responsive static dashboard under `apps/web` with authenticated explorer filters, paginated results, detail review, manual intake, notes/pipeline context, evidence provenance, and browser-only CSV preview. - Docker Compose runtime with non-root containers, read-only filesystems, health checks, and a named SQLite data volume. @@ -86,6 +86,16 @@ A discovery query is a tenant-scoped, bounded, auditable request that can be val SQLite, the in-process worker, and the named local volume are suitable for the pilot only; production migration, durable queue/worker leases, event retention/backup, SSE delivery, and tested backup/restore remain unfinished. Redis and Celery are not implemented. The development password fallback is PBKDF2 rather than production Argon2id. Before production, complete the gates in `docs/SECURITY.md` and `docs/OPERATIONS.md`, including MFA, TLS, CSRF protection, source approval and terms review, rate limiting, circuit monitoring, tenant-scoped job/event authorization, idempotent side-effect handling, durable raw-source/audit retention, SSRF-safe fetching if a future scanner is approved, and tested backups/restores. +## Phase 6 normalization and deduplication boundary + +Normalization is deterministic and versioned. For South African data, phone values are stripped to digits, local 10-digit `0` forms and `00 27` forms are converted to canonical `+27...`, and unknown international numbers retain their explicit country code; presentation punctuation must not create a second identity. Locations derive whitespace/case/diacritic-folded province, city, and suburb fields. A normalized value is not proof that the underlying observation is correct. + +Exact keys (for example, canonical domain, email, or phone) may identify duplicate candidates. Fuzzy matching is deterministic and suggestion-only: the same inputs and normalization version produce the same candidate, score, and reason. A suggested match must never merge automatically. Use the documented thresholds: `>=0.90` is a strong suggestion, `0.75–0.8999` is a review suggestion, and `<0.75` is not surfaced as a suggestion. A human with permission must explicitly confirm each merge. + +Every confirmed merge must create a tenant-scoped, immutable-enough merge snapshot before mutation, recording the surviving and absorbed IDs, normalized comparison inputs, score/reasons, acting user, timestamp, and schema/normalization versions. The operation must be reversible from that snapshot. It must preserve or re-parent every child, evidence item, provenance/source-record link, note, pipeline/audit history, and original source identity; conflicts remain visible for human resolution rather than being silently overwritten. Cross-tenant candidates are never comparable or mergeable, and each suggestion, confirmation, rejection, reversal, and preservation/conflict decision belongs in the audit trail. + +The MVP now exposes deterministic match suggestions at `GET /api/v1/businesses/{id}/matches`, explicit merge confirmation in the web review dialog, tenant-scoped merge history, and `POST /api/v1/merge-history/{id}/reverse`. The implementation remains a pilot boundary: hardening is still needed for a dedicated merge permission, stronger server-side confirmation semantics, full snapshot conflict handling, and production-grade rollback guarantees. Do not describe a normalized or suggested match as verified identity, discovery, enrichment, or outreach authorization. + ## Verification ```bash diff --git a/apps/api/README.md b/apps/api/README.md index 1661adb..487b157 100644 --- a/apps/api/README.md +++ b/apps/api/README.md @@ -1,6 +1,6 @@ -# Prospect Platform API — Phase 5 boundary +# Prospect Platform API — Phase 6 boundary -Dependency-light JSON API for tenant-scoped, **manual** prospect workflows and the Phase 5 source-ingestion contract. Core domain rules use Python's standard library and persistence is SQLite. The API stores businesses plus child intelligence/evidence records, pipeline state, notes, and audit context. Source queries and adapter results must remain auditable and fail closed; the current runtime does not perform network discovery, DNS/website scanning, or outreach. +Dependency-light JSON API for tenant-scoped, **manual** prospect workflows and the Phase 6 normalization/deduplication plus Phase 5 source-ingestion contracts. Core domain rules use Python's standard library and persistence is SQLite. The API stores businesses plus child intelligence/evidence records, pipeline state, notes, and audit context. Source queries and adapter results must remain auditable and fail closed; the current runtime does not perform network discovery, DNS/website scanning, or outreach. ## Run @@ -88,6 +88,16 @@ Pipeline state and verification are review metadata, not outreach authorization. All SQL uses parameters and all responses are JSON. Scores include `score_version` and `score_factors` for traceability. Provenance is supplied by the operator/source record; the MVP does not validate external sources or independently refresh evidence. +### Phase 6 normalization and duplicate-review contract + +The server is the normalization authority. SA phone input is canonicalized using explicit `+27` context for local `0` numbers, with punctuation/spacing removed while retaining a display/original value. Location input retains the raw observation and derives a comparison form plus country/province/municipality/city tokens; missing or ambiguous locality must remain missing/ambiguous, not guessed. The normalization/schema version must be stored with derived values so reprocessing is deterministic. + +Exact duplicate keys are deterministic. `GET /api/v1/businesses/{id}/matches` compares only active businesses in the authenticated organization and produces a sorted, deterministic score, score version, and explainable reasons. Its default candidate cutoff is `0.72`; policy bands are `>=0.90` strong suggestion, `0.75–0.8999` review suggestion, and `<0.75` no suggestion. Fuzzy comparison is **suggestion-only** and there is no automatic merge at any score. `POST /api/v1/businesses/{id}/merge` requires an authenticated mutating-role user and an explicit target; the web client also requires a human confirmation. A production merge permission and server-verifiable confirmation token remain hardening work. + +Before a confirmed merge, the current route persists a tenant-scoped `merge_history` snapshot of the source business and child rows (`business_identifiers`, `contacts`, `domains`, `websites`, `evidence`, `pipeline_entries`, `interactions`, and `notes`), plus child IDs/counts, actor, and timestamp. It re-parents those children to the target without deleting source records; `GET /api/v1/merge-history` reads the ledger and `POST /api/v1/merge-history/{id}/reverse` restores the source/child links. Candidate queries, merges, snapshots, reversal, and history reads apply the organization predicate; a cross-tenant ID behaves as not found. Audit events record merge and reversal actions. + +Remaining limitations: the snapshot currently focuses on the source graph rather than a full two-parent conflict snapshot; the merge route does not yet enforce a dedicated merge permission or cryptographically bound confirmation payload; and preservation/conflict semantics need production-grade transactional and concurrency tests. It must not claim that normalization proves identity or that deduplication performs discovery or outreach. + ## Pagination and filtering rules List and child-record endpoints are deliberately bounded. For business lists, use `page` (starting at 1) and `page_size` within the server-enforced maximum; invalid values are rejected rather than allowing an unbounded query. Supported filters are applied inside the tenant-scoped query before pagination: `q`, `score_min`, `score_max`, `website_class`, and `pipeline_stage`. The UI's page and filter controls are convenience clients, not authorization controls. A filtered page is not a count of the entire unfiltered tenant unless the response explicitly says so. diff --git a/apps/api/app/domain.py b/apps/api/app/domain.py index 9ad28ae..14681ad 100644 --- a/apps/api/app/domain.py +++ b/apps/api/app/domain.py @@ -2,9 +2,12 @@ from __future__ import annotations import re +import unicodedata +from difflib import SequenceMatcher from urllib.parse import urlparse SCORE_VERSION = "mvp-1" +MATCH_SCORE_VERSION = "phase6-1" _SOCIAL = {"facebook.com", "instagram.com", "linkedin.com", "twitter.com", "x.com", "youtube.com", "tiktok.com"} @@ -20,7 +23,38 @@ def normalize_domain(value: str | None) -> str: def normalize_phone(value: str | None) -> str: - return re.sub(r"[^0-9+]", "", (value or "").strip()) + raw = str(value or "").strip() + if not raw: + return "" + # Keep a leading international plus and digits only; never invent a country + # code for an unknown number. South African local and 00 prefixes are safe + # canonicalization cases because their numbering plan is unambiguous. + compact = re.sub(r"[^0-9+]", "", raw) + if compact.startswith("00"): + compact = "+" + compact[2:] + if compact.startswith("+27"): + rest = compact[3:] + if rest.startswith("0"): + rest = rest[1:] + return "+27" + rest + if compact.startswith("0") and len(compact) == 10: + return "+27" + compact[1:] + if compact.startswith("+"): + return "+" + re.sub(r"\D", "", compact[1:]) + return re.sub(r"\D", "", compact) + + +def _location_part(value: object) -> str: + text = " ".join(str(value or "").split()).strip().lower() + return "".join(c for c in unicodedata.normalize("NFKD", text) if not unicodedata.combining(c)) + + +def normalize_location(value: object = None, *, province=None, city=None, suburb=None) -> dict: + if isinstance(value, dict): + province, city, suburb = value.get("province", province), value.get("city", city), value.get("suburb", suburb) + elif value is not None and not any(x is not None for x in (province, city, suburb)): + province = value + return {"province": _location_part(province), "city": _location_part(city), "suburb": _location_part(suburb)} def normalize_business(raw: dict) -> dict: @@ -31,9 +65,37 @@ def normalize_business(raw: dict) -> dict: phone = normalize_phone(raw.get("phone")) result = dict(raw) result.update({"name": name, "email": email, "website": website, "website_domain": domain, "phone": phone}) + result.update(normalize_location(raw.get("location", raw))) return result +def match_businesses(source: dict, candidates: list[dict], threshold: float = 0.72) -> list[dict]: + """Return deterministic, explainable suggestions; this function never merges.""" + left = normalize_business(source) + output = [] + for raw in candidates: + right = normalize_business(raw) + signals = [] + if left["website_domain"] and left["website_domain"] == right["website_domain"]: + signals.append((1.0, "exact_website_domain")) + if left["email"] and left["email"] == right["email"]: + signals.append((1.0, "exact_email")) + if left["phone"] and left["phone"] == right["phone"]: + signals.append((1.0, "exact_phone")) + if left["name"] and right["name"]: + similarity = SequenceMatcher(None, re.sub(r"[^a-z0-9]", "", left["name"].lower()), re.sub(r"[^a-z0-9]", "", right["name"].lower())).ratio() + if similarity >= 0.65: signals.append((similarity, "similar_name")) + for field, reason in (("province", "same_province"), ("city", "same_city"), ("suburb", "same_suburb")): + if left[field] and left[field] == right[field]: signals.append((0.08, reason)) + if not signals: continue + exact = [s for s, r in signals if r.startswith("exact_")] + name = next((s for s, r in signals if r == "similar_name"), 0.0) + confidence = max(exact or [0.0]) if exact else min(0.99, 0.65 * name + sum(s for s, r in signals if r.startswith("same_"))) + if confidence >= threshold: + output.append({"id": raw.get("id"), "confidence": round(confidence, 4), "reasons": [r for _, r in signals], "score_version": MATCH_SCORE_VERSION}) + return sorted(output, key=lambda x: (-x["confidence"], x["id"] if isinstance(x["id"], int) else str(x["id"]))) + + def classify_website(website_or_domain: str | None) -> str: domain = normalize_domain(website_or_domain) if not domain: diff --git a/apps/api/app/main.py b/apps/api/app/main.py index 57ceb9e..32b1b7e 100644 --- a/apps/api/app/main.py +++ b/apps/api/app/main.py @@ -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): diff --git a/apps/api/schema.sql b/apps/api/schema.sql index ffd2154..c4276a2 100644 --- a/apps/api/schema.sql +++ b/apps/api/schema.sql @@ -25,6 +25,14 @@ CREATE TABLE IF NOT EXISTS businesses ( score_version TEXT NOT NULL DEFAULT 'mvp-1', score_factors TEXT NOT NULL DEFAULT '[]', website_class TEXT NOT NULL DEFAULT 'missing', verified INTEGER NOT NULL DEFAULT 0, verified_at TEXT, created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP ); +-- Phase 6 normalized location and reversible merge ledger. +CREATE TABLE IF NOT EXISTS merge_history ( + id INTEGER PRIMARY KEY AUTOINCREMENT, organization_id TEXT NOT NULL REFERENCES organizations(id), + source_business_id INTEGER NOT NULL REFERENCES businesses(id), target_business_id INTEGER NOT NULL REFERENCES businesses(id), + source_snapshot_json TEXT NOT NULL, child_reassignment_json TEXT NOT NULL DEFAULT '{}', actor_user_id INTEGER REFERENCES users(id) ON DELETE SET NULL, + reversible INTEGER NOT NULL DEFAULT 1, reversed_at TEXT, created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP +); +CREATE INDEX IF NOT EXISTS idx_merge_history_org ON merge_history(organization_id,created_at DESC,id DESC); CREATE INDEX IF NOT EXISTS idx_businesses_org ON businesses(organization_id); CREATE INDEX IF NOT EXISTS idx_businesses_score ON businesses(organization_id,score DESC,id); CREATE INDEX IF NOT EXISTS idx_businesses_class ON businesses(organization_id,website_class); diff --git a/apps/api/tests/test_phase6.py b/apps/api/tests/test_phase6.py new file mode 100644 index 0000000..ae7ee89 --- /dev/null +++ b/apps/api/tests/test_phase6.py @@ -0,0 +1,30 @@ +import unittest +from app.domain import normalize_phone, normalize_location, normalize_business, match_businesses + +class Phase6DomainTests(unittest.TestCase): + def test_south_african_phone_formats_share_canonical_value(self): + self.assertEqual(normalize_phone("082 555 1234"), "+27825551234") + self.assertEqual(normalize_phone("0027 82 555 1234"), "+27825551234") + self.assertEqual(normalize_phone("+27 (82) 555-1234"), "+27825551234") + + def test_unknown_international_phone_is_not_rewritten(self): + self.assertEqual(normalize_phone("+44 (20) 1234 5678"), "+442012345678") + self.assertEqual(normalize_phone("555-1234"), "5551234") + + def test_location_normalization_and_business_fields(self): + self.assertEqual(normalize_location({"province": " Gauteng ", "city": " Johannesburg ", "suburb": " Sandton "}), {"province": "gauteng", "city": "johannesburg", "suburb": "sandton"}) + b = normalize_business({"name":" Acme ", "province":" Gauteng ", "city":" Johannesburg ", "suburb":" Sandton "}) + self.assertEqual((b["province"], b["city"], b["suburb"]), ("gauteng", "johannesburg", "sandton")) + + def test_matching_has_deterministic_confidence_and_reasons(self): + a = {"name":"Acme Consulting", "email":"hello@acme.co.za", "city":"Johannesburg"} + b = {"name":"Acme Consultng", "email":"hello@acme.co.za", "city":"Johannesburg"} + first = match_businesses(a, [dict(b, id=2), {"id":3,"name":"Unrelated Shop"}], threshold=0.5) + second = match_businesses(a, [dict(b, id=2), {"id":3,"name":"Unrelated Shop"}], threshold=0.5) + self.assertEqual(first, second) + self.assertEqual(first[0]["id"], 2) + self.assertGreaterEqual(first[0]["confidence"], 0.5) + self.assertTrue(first[0]["reasons"]) + self.assertEqual([x["id"] for x in match_businesses(a, [{"id":3,"name":"Unrelated Shop"}], threshold=0.8)], []) + +if __name__ == "__main__": unittest.main() diff --git a/apps/api/tests/test_phase6_api.py b/apps/api/tests/test_phase6_api.py new file mode 100644 index 0000000..2cbc2d5 --- /dev/null +++ b/apps/api/tests/test_phase6_api.py @@ -0,0 +1,35 @@ +import json, os, sqlite3, threading, unittest +from http.client import HTTPConnection +from tempfile import TemporaryDirectory +from app.main import create_server + +class Phase6ApiTests(unittest.TestCase): + def setUp(self): + self.tmp=TemporaryDirectory(); os.environ['BOOTSTRAP_ADMIN_EMAIL']='owner@phase6.test'; os.environ['BOOTSTRAP_ADMIN_PASSWORD']='password' + self.server=create_server('127.0.0.1',0,self.tmp.name+'/db.sqlite'); self.thread=threading.Thread(target=self.server.serve_forever,daemon=True); self.thread.start(); self.conn=HTTPConnection('127.0.0.1',self.server.server_port,timeout=3); self.cookie=None + self.request('POST','/api/v1/auth/login',{'email':'owner@phase6.test','password':'password'}) + def tearDown(self): self.server.shutdown(); self.server.server_close(); self.thread.join(2); self.tmp.cleanup() + def request(self,method,path,payload=None): + body=json.dumps(payload).encode() if payload is not None else None; headers={'Content-Type':'application/json'} if body else {} + if self.cookie: headers['Cookie']=self.cookie + self.conn.request(method,path,body,headers); r=self.conn.getresponse(); c=r.getheader('Set-Cookie'); + if c: self.cookie=c.split(';',1)[0] + return r.status,json.loads(r.read() or b'{}') + def test_merge_preserves_children_and_reverse_restores_ownership(self): + _, source=self.request('POST','/api/v1/businesses',{'name':'Acme Consulting','phone':'082 555 1234'}) + _, target=self.request('POST','/api/v1/businesses',{'name':'Acme Consulting HQ','city':'Johannesburg'}) + self.assertEqual(self.request('POST',f"/api/v1/businesses/{source['id']}/notes",{'body':'evidence'})[0],201) + self.assertEqual(self.request('POST',f"/api/v1/businesses/{source['id']}/evidence",{'kind':'source','claim':'claim'})[0],201) + status, merged=self.request('POST',f"/api/v1/businesses/{source['id']}/merge",{'target_business_id':target['id']}) + self.assertEqual(status,200); self.assertEqual(merged['status'],'merged') + status, detail=self.request('GET',f"/api/v1/businesses/{target['id']}"); self.assertEqual(status,200); self.assertEqual(len(detail['notes']),1); self.assertEqual(len(detail['evidence']),1) + status, source_detail=self.request('GET',f"/api/v1/businesses/{source['id']}"); self.assertEqual(status,200); self.assertEqual(source_detail['merge_status'],'merged') + status, reversed_=self.request('POST',f"/api/v1/merge-history/{merged['merge_history_id']}/reverse",{}); self.assertEqual(status,200); self.assertEqual(reversed_['status'],'reversed') + _, restored=self.request('GET',f"/api/v1/businesses/{source['id']}"); self.assertEqual(len(restored['notes']),1); self.assertEqual(len(restored['evidence']),1) + def test_matches_endpoint_is_thresholded_and_history_is_tenant_scoped(self): + _, a=self.request('POST','/api/v1/businesses',{'name':'Bright Co','email':'hello@bright.test'}) + _, b=self.request('POST','/api/v1/businesses',{'name':'Bright Company','website':'https://bright.test'}) + status, matches=self.request('GET',f"/api/v1/businesses/{a['id']}/matches?threshold=0.4"); self.assertEqual(status,200); self.assertEqual(matches['items'][0]['id'],b['id']); self.assertIn('similar_name',matches['items'][0]['reasons']) + self.assertEqual(self.request('GET','/api/v1/merge-history')[0],200) + +if __name__=='__main__': unittest.main() diff --git a/apps/web/README.md b/apps/web/README.md index 1d7a4c9..15203a9 100644 --- a/apps/web/README.md +++ b/apps/web/README.md @@ -1,6 +1,6 @@ -# ProspectOS web — Phase 5 boundary +# ProspectOS web — Phase 6 boundary -Self-contained static frontend for the Prospect Platform API. There is no bundler or runtime dependency: serve this directory with any static HTTP server. The UI supports a manual, tenant-scoped review workflow and a Phase 4 MVP job monitor. Phase 5 source concepts are display/contract boundaries only; the UI does not perform network discovery, scan DNS/websites, or send outreach. +Self-contained static frontend for the Prospect Platform API. There is no bundler or runtime dependency: serve this directory with any static HTTP server. The UI supports a manual, tenant-scoped review workflow and a Phase 4 MVP job monitor. Phase 5 source concepts and Phase 6 normalization/deduplication concepts are display/contract boundaries only; the UI does not perform network discovery, scan DNS/websites, or send outreach. ## Configure and run @@ -24,6 +24,14 @@ If not set, the UI uses `localStorage.prospect_api_base` when present; otherwise The API remains the source of truth for tenant isolation, pagination bounds, filters, pipeline transitions, notes, audit records, and suppression. See `apps/api/README.md` for the route contract. +## Phase 6 normalization and deduplication UI contract + +The UI may display the API's normalized SA phone and location values alongside the original observation, normalization version, and any ambiguity warning. It must not silently replace the source value or imply that a canonical form verifies identity. Duplicate candidates must show deterministic score, threshold band (`strong` `>=0.90`, `review` `0.75–0.8999`, or `none` `<0.75`), and explainable matching reasons. + +Suggestions are review aids only. A merge flow must identify the surviving record, list all parents/children/evidence/notes/source records that will be preserved, show conflicts, and require an explicit human confirmation before calling an authorized API mutation. The UI must offer rejection and, where implemented, reversal using the merge snapshot; it must never auto-merge based on a score. Candidate, merge, snapshot, and audit data are tenant-scoped by the API, not by hidden UI state. + +The current static MVP requests `/matches`, renders a **Human review required** list with confidence/reasons, asks for **Confirm merge**, and displays merge history with **Reverse merge** when the API marks it reversible. The API remains authoritative; these controls are not a substitute for server-side authorization. Existing normalization and match display remain suggestion-only; no merge happens without explicit operator confirmation. + ## Phase 5 source UI contract The web client may display registered source metadata, query mode, approval/terms state, rate-limit status, retention class, health, and circuit state returned by the API. It must label `dry_run` as a plan/validation result and distinguish operator-supplied CSV/manual references from independently verified evidence. It must not offer a live-source control unless the API reports explicit approval and operational enablement; client visibility is never an authorization control. diff --git a/apps/web/app.js b/apps/web/app.js index 3e7cd86..c40132b 100644 --- a/apps/web/app.js +++ b/apps/web/app.js @@ -25,7 +25,38 @@ async function selectProspect(id){selectedId=Number(id);selectedDetail=null;renderRows();$('detailPanel').innerHTML='
${esc(error.message)}
${empty}
`; - function renderDetail(p){const s=scoreFor(p),st=statusOf(p),f=freshness(p),factors=p.score_factors||p.factors||[],blocked=st==='review'||st==='suppressed',stage=p.pipeline_stage||p.pipeline?.stage||(Array.isArray(p.pipeline)?p.pipeline.at(-1)?.stage:'')||'new',contacts=p.contacts||[],domains=p.domains||[],websites=p.websites||[],evidence=p.evidence||p.evidence_timeline||[],notes=p.notes||[],review=p.review_status||p.review|| (st==='review'?'Needs review':st[0].toUpperCase()+st.slice(1));$('detailPanel').innerHTML=`PROSPECT DETAIL
${esc(p.website_domain||'no detected website')}
✓ ${esc(labelFactor(x))}${esc(p.confidence||'Medium')}
`).join(''):''}${esc(review)}
${st!=='suppressed'?'':''}${st==='suppressed'?'Suppressed records cannot be contacted.':'Review this prospect before outreach is available.'}
`:''}`;} + function renderDetail(p){const s=scoreFor(p),st=statusOf(p),f=freshness(p),factors=p.score_factors||p.factors||[],blocked=st==='review'||st==='suppressed',stage=p.pipeline_stage||p.pipeline?.stage||(Array.isArray(p.pipeline)?p.pipeline.at(-1)?.stage:'')||'new',contacts=p.contacts||[],domains=p.domains||[],websites=p.websites||[],evidence=p.evidence||p.evidence_timeline||[],notes=p.notes||[],review=p.review_status||p.review|| (st==='review'?'Needs review':st[0].toUpperCase()+st.slice(1));$('detailPanel').innerHTML=`PROSPECT DETAIL
${esc(p.website_domain||'no detected website')}
✓ ${esc(labelFactor(x))}${esc(p.confidence||'Medium')}
`).join(''):''}${esc(review)}
${st!=='suppressed'?'':''}${st==='suppressed'?'Suppressed records cannot be contacted.':'Review this prospect before outreach is available.'}
`:''}`;renderDedupPanel();} + let mergeSource = null, mergeTarget = null, mergeBusy = false; + const payloadItems = (payload, keys) => { for (const key of keys) if (Array.isArray(payload?.[key])) return payload[key]; return Array.isArray(payload) ? payload : []; }; + const suggestionId = item => item.target_id ?? item.business_id ?? item.prospect_id ?? item.id; + const suggestionName = item => item.target_name || item.business_name || item.prospect_name || item.name || `Prospect ${suggestionId(item)}`; + const suggestionConfidence = item => item.confidence ?? item.score ?? item.match_confidence ?? 'Unknown'; + const suggestionReasons = item => item.reasons || item.reason || item.match_reasons || item.explanation || []; + const reasonItems = reasons => Array.isArray(reasons) ? reasons : [reasons]; + function renderDedupPanel() { + const detail = $('detailPanel'); if (!detail || !selectedId) return; + let panel = $('dedupPanel'); + if (!panel) { panel = document.createElement('section'); panel.id = 'dedupPanel'; panel.className = 'dedup-panel'; panel.dataset.smoke = 'deduplication'; detail.appendChild(panel); } + panel.innerHTML = 'DEDUPLICATION
No possible matches returned. Nothing was merged automatically.
'}${esc(error.message)}
No merges recorded for this prospect.
'}`; } catch (error) { if (error.message !== 'unauthorized') history.innerHTML = `${esc(error.message)}
REVIEW REQUIRED