diff --git a/apps/api/app/main.py b/apps/api/app/main.py index 603907a..4e92fae 100644 --- a/apps/api/app/main.py +++ b/apps/api/app/main.py @@ -7,12 +7,12 @@ 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, match_businesses + from app.domain import deduplication_key, deduplicate_businesses, is_suppressed, normalize_business, normalize_domain, normalize_phone, match_businesses 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 - from app.scoring import DEFAULT_RULES, signals_for_business, evaluate_score, SCORE_VERSION + from app.scoring import DEFAULT_RULES, signals_for_business, evaluate_score, score_business_opportunity, SCORE_VERSION from app.ai_assistance import generate as generate_ai, input_fingerprint, provider_status, MAX_INPUT_ITEMS, MAX_OUTPUT_CHARS from app.discovery import discover as scoped_discover from app.ai_research import provider_status as ai_research_provider_status, configure_db as configure_ai_research_db, validate_criteria as validate_ai_research_criteria, AIResearchConfigError @@ -20,12 +20,12 @@ if __package__ in (None, ""): from app.config import load_config 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 .domain import deduplication_key, deduplicate_businesses, is_suppressed, normalize_business, normalize_domain, normalize_phone, match_businesses 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 - from .scoring import DEFAULT_RULES, signals_for_business, evaluate_score, SCORE_VERSION + from .scoring import DEFAULT_RULES, signals_for_business, evaluate_score, score_business_opportunity, SCORE_VERSION from .ai_assistance import generate as generate_ai, input_fingerprint, provider_status, MAX_INPUT_ITEMS, MAX_OUTPUT_CHARS from .discovery import discover as scoped_discover from .ai_research import provider_status as ai_research_provider_status, configure_db as configure_ai_research_db, validate_criteria as validate_ai_research_criteria, AIResearchConfigError @@ -1180,7 +1180,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,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())) + scored=score_business_opportunity(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["explanations"]),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"}) @@ -1470,11 +1470,11 @@ def _run_scoped_discovery(db, job, handler): existing = db.execute("SELECT id FROM businesses WHERE organization_id=? AND website_domain=?", (org, b["website_domain"])).fetchone() if existing: bid = existing["id"] else: - 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"],b.get("description", ""),b["province"],b["city"],b["suburb"],scored["score"],scored["score_version"],json.dumps(scored["factors"]),scored["website_class"])) + scored = score_business_opportunity(b, sources=["scoped_discovery"]) + 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"],b.get("description", ""),b["province"],b["city"],b["suburb"],scored["score"],scored["score_version"],json.dumps(scored["explanations"]),scored["website_class"])) bid = cur.lastrowid priority = "high" if scored["score"] >= 70 else "medium" if scored["score"] >= 40 else "low" - 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,priority,scored["score_version"],json.dumps(scored["factors"]),json.dumps({"source": "scoped_discovery"}, sort_keys=True))) + 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"],int(scored["eligible"]),scored["priority_band"],scored["score_version"],json.dumps(scored["explanations"]),json.dumps(scored["signals"], sort_keys=True))) scan_ids = {} for page in candidate.get("pages", []): scan_key = hashlib.sha256((org + ":" + page["url"]).encode()).hexdigest() @@ -1545,8 +1545,8 @@ def _run_source_discovery(db, job, handler): 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"]}))) + scored=score_business_opportunity(normalized, sources=[source["kind"]]); 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["explanations"]),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"],int(scored["eligible"]),scored["priority_band"],scored["score_version"],json.dumps(scored["explanations"]),json.dumps(scored["signals"],sort_keys=True))) 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")) diff --git a/apps/api/app/scoring.py b/apps/api/app/scoring.py index 0b71702..acf7df7 100644 --- a/apps/api/app/scoring.py +++ b/apps/api/app/scoring.py @@ -1,20 +1,36 @@ -"""Deterministic, explainable qualification scoring.""" +"""Deterministic, transparent opportunity scoring.""" from __future__ import annotations import json -from datetime import datetime, timezone +from urllib.parse import urlparse + +SCORE_VERSION = "opportunity-v1" + + +def _rule(code, name, points, description): + return {"code": code, "name": name, "description": description, + "condition_json": {"signal": f"opportunity.{code}", "operator": "truthy"}, + "points": points, "max_applications": 1, "enabled": 1, "version": 1} + -SCORE_VERSION = "phase10-1" DEFAULT_RULES = [ - {"code": "business_name", "name": "Named business", "description": "Business has a usable name", "condition_json": {"signal": "business.name", "operator": "present"}, "points": 15, "max_applications": 1, "enabled": 1, "version": 1}, - {"code": "business_site", "name": "Business website", "description": "Business has a non-social website", "condition_json": {"signal": "business.website_class", "operator": "equals", "value": "business_site"}, "points": 20, "max_applications": 1, "enabled": 1, "version": 1}, - {"code": "website_healthy", "name": "Healthy website", "description": "Latest website scan is healthy", "condition_json": {"signal": "website.classification", "operator": "equals", "value": "healthy"}, "points": 15, "max_applications": 1, "enabled": 1, "version": 1}, - {"code": "contact_email", "name": "Email contact", "description": "A direct business email is available", "condition_json": {"signal": "business.email", "operator": "present"}, "points": 15, "max_applications": 1, "enabled": 1, "version": 1}, - {"code": "contact_phone", "name": "Phone contact", "description": "A business phone is available", "condition_json": {"signal": "business.phone", "operator": "present"}, "points": 10, "max_applications": 1, "enabled": 1, "version": 1}, - {"code": "extracted_contact", "name": "Extracted contact", "description": "A public, non-suppressed contact was extracted", "condition_json": {"signal": "contacts.public_count", "operator": "gte", "value": 1}, "points": 10, "max_applications": 1, "enabled": 1, "version": 1}, - {"code": "domain_verified", "name": "Domain check", "description": "Domain check resolved successfully", "condition_json": {"signal": "domain.status", "operator": "in", "value": ["resolved", "ok", "healthy"]}, "points": 5, "max_applications": 1, "enabled": 1, "version": 1}, - {"code": "verified_business", "name": "Verified business", "description": "Business has been verified", "condition_json": {"signal": "state.verified", "operator": "truthy"}, "points": 10, "max_applications": 1, "enabled": 1, "version": 1}, + _rule("no_detected_website", "No detected website", 30, "No website was detected for the business."), + _rule("no_official_domain", "No official domain", 25, "No official business domain was corroborated."), + _rule("no_functioning_web_service", "Domain but no functioning web service", 25, "A domain exists but no functioning web service was observed."), + _rule("broken_website", "Broken website", 25, "The observed website is broken."), + _rule("parked_default_placeholder", "Parked/default/placeholder website", 20, "The website is parked, default, or a placeholder."), + _rule("public_free_mail", "Public free-mail address", 15, "A public business contact uses a free-mail provider."), + _rule("human_reviewed_outdated", "Human-reviewed outdated website", 15, "A human reviewer marked the website outdated."), + _rule("no_working_https", "No working HTTPS", 10, "No working HTTPS service was verified."), + _rule("severe_performance", "Severe performance issue", 10, "The website has a severe performance issue."), + _rule("active_social", "Active social presence", 10, "An active social presence was detected."), + _rule("valid_public_business_phone", "Valid public business phone", 5, "A valid public business phone is available."), + _rule("multiple_corroborating_sources", "Multiple corroborating sources", 5, "Multiple independent sources corroborate the business."), + _rule("possibly_closed", "Possibly closed", -30, "Evidence suggests the business may be closed."), + _rule("healthy_modern_website", "Healthy modern website", -30, "The website is healthy and modern."), + _rule("stale_or_uncertain", "Stale or uncertain evidence", -15, "The evidence is stale or uncertain."), ] + def _get(data, path): value = data for part in str(path).split("."): @@ -22,14 +38,18 @@ def _get(data, path): value = value.get(part) return value + def _match(condition, signals): if not isinstance(condition, dict): return False if "all" in condition: return all(_match(c, signals) for c in condition["all"]) if "any" in condition: return any(_match(c, signals) for c in condition["any"]) if "not" in condition: return not _match(condition["not"], signals) - value = _get(signals, condition.get("signal", "")); op = condition.get("operator", "truthy"); expected = condition.get("value") - section = signals.get(str(condition.get("signal", "")).split(".")[0], {}) - if isinstance(section, dict) and (section.get("stale") or section.get("uncertain")): return False + path = str(condition.get("signal", "")); value = _get(signals, path) + op = condition.get("operator", "truthy"); expected = condition.get("value") + section = signals.get(path.split(".")[0], {}) if isinstance(signals, dict) else {} + # Positive evidence is suppressed when its evidence section is stale/uncertain; + # the explicit opportunity.stale_or_uncertain rule remains evaluable. + if path != "opportunity.stale_or_uncertain" and isinstance(section, dict) and (section.get("stale") or section.get("uncertain")): return False if op in ("truthy", "present"): return bool(value) if op == "truthy" else value not in (None, "", [], {}) if op == "equals": return value == expected if op == "in": return value in (expected if isinstance(expected, list) else [expected]) @@ -38,6 +58,7 @@ def _match(condition, signals): except (TypeError, ValueError): return False return False + def evaluate_score(signals, rules): total = 0; explanations = [] ordered = sorted((dict(r) for r in rules), key=lambda r: (str(r.get("code", "")), int(r.get("id", 0) or 0))) @@ -48,12 +69,10 @@ def evaluate_score(signals, rules): explanations.append({"code": rule.get("code", ""), "name": rule.get("name", rule.get("code", "")), "version": int(rule.get("version", 1) or 1), "enabled": enabled, "applied": applied, "points": points, "reason": (rule.get("description") or rule.get("name") or rule.get("code") or "Rule") + (" (matched)" if applied else " (not matched)")}) total = max(0, min(100, total)); state = signals.get("state", {}) if isinstance(signals, dict) else {} eligible = not bool(state.get("suppressed")) and str(state.get("merge_status", "active")) == "active" - if not eligible: band = "ineligible" - elif total >= 70: band = "high" - elif total >= 40: band = "medium" - else: band = "low" + band = "ineligible" if not eligible else "high" if total >= 70 else "medium" if total >= 40 else "low" return {"score": total, "score_version": SCORE_VERSION, "eligible": eligible, "priority_band": band, "explanations": explanations} + def _condition(rule): raw = rule.get("condition_json", {}) if isinstance(raw, str): @@ -61,7 +80,47 @@ def _condition(rule): except (TypeError, ValueError): return {} return raw -def signals_for_business(business, website=None, contacts=None, domain=None, suppressed=False): - b = dict(business); website = website or {}; contacts = contacts or []; domain = domain or {} + +def _has_working_https(website): + url = website.get("final_url") or website.get("input_url") or "" + return urlparse(str(url)).scheme.lower() == "https" and website.get("tls") is not False and website.get("certificate_status", "valid") not in {"invalid", "error"} + + +def signals_for_business(business, website=None, contacts=None, domain=None, suppressed=False, sources=None): + b = dict(business); website = dict(website or {}); contacts = contacts or []; domain = dict(domain or {}) public = [c for c in contacts if c.get("public_business") and not c.get("suppressed") and not c.get("do_not_contact")] - return {"business": {"name": b.get("name", ""), "email": b.get("email", ""), "phone": b.get("phone", ""), "description": b.get("description", ""), "website_domain": b.get("website_domain", ""), "website_class": b.get("website_class", "")}, "website": website, "contacts": {"count": len(contacts), "public_count": len(public)}, "domain": domain, "state": {"verified": bool(b.get("verified")), "suppressed": bool(suppressed), "merge_status": b.get("merge_status", "active"), "merged": b.get("merge_status") == "merged"}} + free_mail = any(str(c.get("classification", "")).lower() == "free_mail" for c in public) + phone = str(b.get("phone", "") or "") + valid_phone = sum(ch.isdigit() for ch in phone) >= 7 or any(c.get("kind") == "phone" for c in public) + classification = str(website.get("classification") or b.get("website_class") or "").lower() + has_domain = bool(b.get("website_domain") or b.get("website") or domain.get("domain")) + stale = any(isinstance(x, dict) and (x.get("stale") or x.get("uncertain")) for x in (b, website, domain)) or bool(b.get("stale") or b.get("uncertain")) + closed = str(b.get("status", "")).lower() in {"closed", "possibly_closed"} or bool(b.get("possibly_closed")) + source_count = len(sources or b.get("sources", []) or []) + opportunity = { + "no_detected_website": not has_domain, + "no_official_domain": not bool(domain.get("official", domain.get("status") in {"resolved", "ok", "healthy"}) and has_domain), + "no_functioning_web_service": has_domain and classification not in {"healthy", "modern", "healthy_modern"}, + "broken_website": classification == "broken", + "parked_default_placeholder": classification in {"parked", "placeholder", "default", "under_construction"}, + "public_free_mail": free_mail or str(b.get("email", "")).lower().split("@")[-1] in {"gmail.com", "yahoo.com", "hotmail.com", "outlook.com", "icloud.com"}, + "human_reviewed_outdated": bool(b.get("human_reviewed_outdated") or website.get("human_reviewed_outdated")), + "no_working_https": has_domain and not _has_working_https(website), + "severe_performance": str(website.get("performance", website.get("performance_severity", ""))).lower() == "severe" or bool(website.get("severe_performance")), + "active_social": bool(website.get("social_signal") or b.get("active_social")), + "valid_public_business_phone": valid_phone, + "multiple_corroborating_sources": source_count >= 2, + "possibly_closed": closed, + "healthy_modern_website": classification in {"healthy_modern", "modern"} or (classification == "healthy" and bool(website.get("modern") or website.get("modern_signal"))), + "stale_or_uncertain": stale, + } + return {"business": {"name": b.get("name", ""), "email": b.get("email", ""), "phone": b.get("phone", ""), "description": b.get("description", ""), "website_domain": b.get("website_domain", ""), "website_class": b.get("website_class", "")}, "website": website, "contacts": {"count": len(contacts), "public_count": len(public)}, "domain": domain, "opportunity": opportunity, "state": {"verified": bool(b.get("verified")), "suppressed": bool(suppressed), "merge_status": b.get("merge_status", "active"), "merged": b.get("merge_status") == "merged"}} + + +def score_business_opportunity(business, website=None, contacts=None, domain=None, suppressed=False, sources=None): + """Score a normalized record with the immutable built-in opportunity model.""" + signals = signals_for_business(business, website, contacts, domain, suppressed, sources) + result = evaluate_score(signals, DEFAULT_RULES) + result["factors"] = [item["code"] for item in result["explanations"] if item["applied"]] + result["website_class"] = str((website or {}).get("classification") or business.get("website_class") or ("business_site" if business.get("website") else "")) + return result | {"signals": signals} diff --git a/apps/api/tests/test_phase10_scoring.py b/apps/api/tests/test_phase10_scoring.py index bf3a750..cc729e2 100644 --- a/apps/api/tests/test_phase10_scoring.py +++ b/apps/api/tests/test_phase10_scoring.py @@ -7,16 +7,54 @@ from http.client import HTTPConnection from tempfile import TemporaryDirectory from app.main import create_server -from app.scoring import DEFAULT_RULES, evaluate_score +from app.scoring import DEFAULT_RULES, SCORE_VERSION, evaluate_score, signals_for_business class ScoringEngineTests(unittest.TestCase): + def test_opportunity_defaults_are_the_exact_versioned_transparent_model(self): + expected = { + "no_detected_website": 30, "no_official_domain": 25, + "no_functioning_web_service": 25, "broken_website": 25, + "parked_default_placeholder": 20, "public_free_mail": 15, + "human_reviewed_outdated": 15, "no_working_https": 10, + "severe_performance": 10, "active_social": 10, + "valid_public_business_phone": 5, "multiple_corroborating_sources": 5, + "possibly_closed": -30, "healthy_modern_website": -30, + "stale_or_uncertain": -15, + } + self.assertEqual(SCORE_VERSION, "opportunity-v1") + self.assertEqual({r["code"]: r["points"] for r in DEFAULT_RULES}, expected) + self.assertTrue(all(r["version"] == 1 and r["enabled"] == 1 for r in DEFAULT_RULES)) + + def test_each_opportunity_signal_applies_only_when_present_and_explains_rule_metadata(self): + for rule in DEFAULT_RULES: + signals = {"state": {"suppressed": False, "merge_status": "active"}, "opportunity": {rule["code"]: True}} + result = evaluate_score(signals, [rule]) + self.assertEqual(result["score"], max(0, rule["points"]), rule["code"]) + explanation = result["explanations"][0] + self.assertEqual({explanation[k] for k in ("code", "name", "points", "version")}, {rule["code"], rule["name"], rule["points"], 1}) + + def test_signal_extraction_maps_website_contacts_and_provenance_to_opportunity_signals(self): + signals = signals_for_business({"name": "Acme", "website": "http://acme.test", "website_domain": "acme.test", "phone": "+27123456789"}, + {"classification": "parked", "social_signal": True, "performance": "severe", "human_reviewed_outdated": True}, + [{"public_business": True, "classification": "free_mail", "suppressed": False, "do_not_contact": False}], + {"status": "unknown", "official": False}, False, sources=["directory", "registry"]) + self.assertTrue(signals["opportunity"]["parked_default_placeholder"]) + self.assertTrue(signals["opportunity"]["public_free_mail"]) + self.assertTrue(signals["opportunity"]["active_social"]) + self.assertTrue(signals["opportunity"]["multiple_corroborating_sources"]) + + def test_score_cap_and_negative_signals_are_deterministic(self): + signals = {"opportunity": {r["code"]: True for r in DEFAULT_RULES}, "state": {"suppressed": False, "merge_status": "active"}} + first = evaluate_score(signals, DEFAULT_RULES) + self.assertEqual(first["score"], 100) + self.assertEqual(first, evaluate_score(signals, list(reversed(DEFAULT_RULES)))) def test_defaults_are_deterministic_and_emit_explanations_and_band(self): signals = {"business": {"name": "Acme", "email": "a@acme.test", "website_domain": "acme.test"}, "website": {"classification": "healthy"}, "state": {"suppressed": False}} first = evaluate_score(signals, DEFAULT_RULES) self.assertEqual(first, evaluate_score(signals, DEFAULT_RULES)) self.assertEqual(0 <= first["score"] <= 100, True) - self.assertEqual(first["priority_band"], "medium") + self.assertEqual(first["priority_band"], "low") self.assertTrue(all("code" in item and "reason" in item for item in first["explanations"])) def test_disabled_and_versioned_rules_change_score_without_nondeterminism(self): @@ -26,12 +64,14 @@ class ScoringEngineTests(unittest.TestCase): self.assertEqual(enabled["score"], 30) self.assertEqual(disabled["score"], 0) - def test_suppression_is_ineligible_and_stale_uncertain_signals_do_not_penalize(self): - signals = {"business": {"name": "Acme"}, "website": {"classification": "unknown", "stale": True}, "domain": {"status": "error"}, "state": {"suppressed": True}} + def test_suppression_is_ineligible_and_stale_uncertain_signal_is_transparent(self): + signals = {"business": {"name": "Acme"}, "website": {"classification": "unknown", "stale": True}, "domain": {"status": "error"}, "opportunity": {"stale_or_uncertain": True}, "state": {"suppressed": True}} result = evaluate_score(signals, DEFAULT_RULES) self.assertFalse(result["eligible"]) self.assertEqual(result["priority_band"], "ineligible") - self.assertNotIn("negative", json.dumps(result["explanations"]).lower()) + stale = next(item for item in result["explanations"] if item["code"] == "stale_or_uncertain") + self.assertTrue(stale["applied"]) + self.assertEqual(stale["points"], -15) class ScoringApiTests(unittest.TestCase): diff --git a/apps/api/tests/test_sources_phase5.py b/apps/api/tests/test_sources_phase5.py index 604ac22..003a68a 100644 --- a/apps/api/tests/test_sources_phase5.py +++ b/apps/api/tests/test_sources_phase5.py @@ -93,6 +93,8 @@ class SourceApiTests(unittest.TestCase): detail = self.req('GET', f"/api/v1/businesses/{businesses[0]['id']}")[1] self.assertTrue(detail['domains']); self.assertTrue(detail['websites']); self.assertTrue(detail['evidence']) self.assertTrue(detail['contacts']); self.assertEqual(detail['review_status'], 'pending') + self.assertEqual(detail['score_version'], 'opportunity-v1') + self.assertTrue(all({'code', 'name', 'points', 'version'} <= set(item) for item in detail['score_factors'])) db = sqlite3.connect(self.tmp.name + '/x.db') self.assertEqual(db.execute('SELECT processing_status FROM source_records').fetchone()[0], 'processed') self.assertEqual(db.execute('SELECT status FROM enrichment_queue').fetchone()[0], 'completed')