Clean contact, directory and company-name data from discovery

harvey/collectors/discover.py

@@ -24,6 +24,7 @@
 import base64
 import logging
 import re
+from html import unescape
 from dataclasses import dataclass, field
 from urllib.parse import urlparse
 
@@ -69,6 +70,9 @@
 JUNK_NAME_PATTERNS = tuple(re.compile(p, re.I) for p in (
     r"^\s*(top|best|\d+)\s+\d*\s*(best\s+)?\w+.*\b(in|near|of)\b",
     r"\b(directory|listings?|reviews? of|compare|find a)\b",
+    # "Ohio HOA Management Company List", "List of Plumbers in Austin"
+    r"\b(compan(y|ies)|firms?|contractors?|businesses|providers?)\s+list\b",
+    r"\blist of\b",
 ))
 
 
@@ -87,6 +91,110 @@ def normalize_domain(url_or_domain: str) -> str:
     return host[4:] if host.startswith("www.") else host
 
 
+# ── Business names ────────────────────────────────────────────────────
+# SERP providers return a page TITLE, not a business name: "Denver, CO | HOA
+# Management Firm", "Home Page - NVCM | Community Association Management".
+# Stored as-is it lands in an email as "...at Denver, CO | HOA Management
+# Firm". The business is the title segment that shares a DISTINCTIVE word with
+# the domain — "management" matches managementharbor.com but says nothing.
+
+_TITLE_SEPARATORS = re.compile(r"\s+[|\-–—·•]\s+|\s*[|•·]\s*|:\s+")
+_GENERIC_NAME_WORDS = frozenset("""
+home page welcome official site website hoa hoas management manager managers
+mgmt mgt company companies co inc llc ltd corp firm group services service
+property properties community communities association associations condo
+condominium residential homeowners owners real estate the and of in for best
+top leading trusted local full arizona phoenix az texas tx florida fl
+california ca nevada nv colorado co georgia ga carolina nc usa us america
+""".split())
+
+
+def _words(text: str) -> list[str]:
+    return [w for w in re.findall(r"[a-z0-9]+", text.lower()) if w]
+
+
+def _domain_stem(domain: str) -> str:
+    host = normalize_domain(domain)
+    return re.sub(r"[^a-z0-9]", "", host.rsplit(".", 1)[0]) if host else ""
+
+
+def _segment_matches_domain(segment: str, stem: str) -> bool:
+    words = _words(segment)
+    distinctive = [w for w in words if w not in _GENERIC_NAME_WORDS]
+    if any(len(w) >= 3 and w in stem for w in distinctive):
+        return True
+    # "Northvale Association of Community Managers" ↔ nacm.org
+    initials = "".join(w[0] for w in words if w not in ("of", "the", "and", "&"))
+    return len(initials) >= 3 and stem.startswith(initials)
+
+
+_SPLIT_WORDS = sorted((w for w in _GENERIC_NAME_WORDS | {"with", "your"}
+                       if len(w) >= 4), key=len, reverse=True)
+
+
+def _split_stem(stem: str) -> list[str]:
+    """managementharbor → [management, harbor]; kqresidential → [kq, residential]."""
+    parts, buf, i = [], "", 0
+    while i < len(stem):
+        word = next((w for w in _SPLIT_WORDS if stem.startswith(w, i)), "")
+        if word:
+            if buf:
+                parts.append(buf)
+                buf = ""
+            parts.append(word)
+            i += len(word)
+        else:
+            buf += stem[i]
+            i += 1
+    if buf:
+        parts.append(buf)
+    return parts
+
+
+def name_from_domain(domain: str) -> str:
+    stem = normalize_domain(domain).rsplit(".", 1)[0]
+    words = [w for chunk in re.split(r"[-_.]+", stem) if chunk for w in _split_stem(chunk)]
+    # Short vowel-less leftovers are initials: "kq" → "KQ".
+    return " ".join(w.upper() if len(w) <= 3 and not re.search(r"[aeiouy]", w)
+                    else w.capitalize() for w in words)
+
+
+def name_matches_domain(name: str, domain: str) -> bool:
+    """True when the stored name demonstrably belongs to this domain."""
+    stem = _domain_stem(domain)
+    return bool(stem and name and _segment_matches_domain(name, stem))
+
+
+def site_name_from_html(html: str, domain: str) -> str:
+    """The business name a homepage gives itself (og:site_name, then <title>),
+    or "" when neither names it in a way the domain backs up."""
+    stem = _domain_stem(domain)
+    candidates = []
+    for pat in (r'<meta[^>]+property=["\']og:site_name["\'][^>]*content=["\']([^"\']+)',
+                r'<meta[^>]+content=["\']([^"\']+)["\'][^>]*property=["\']og:site_name',
+                r"(?is)<title[^>]*>(.*?)</title>"):
+        m = re.search(pat, html or "", re.I)
+        if m:
+            candidates.append(unescape(re.sub(r"\s+", " ", m.group(1))).strip())
+    # The fullest name wins: "The Management Harbor" over og's "Management Harbor".
+    matches = [seg for text in candidates
+               for seg in (x.strip(" ,.-") for x in _TITLE_SEPARATORS.split(text))
+               if seg and len(seg) <= 60 and _segment_matches_domain(seg, stem)]
+    return max(matches, key=len, default="")
+
+
+def business_name(title: str, domain: str) -> str:
+    """The business's name from a SERP title, or one derived from its domain."""
+    stem = _domain_stem(domain)
+    title = re.sub(r"\s*(\.\.\.|…)\s*$", "", (title or "").strip())
+    if stem and title:
+        for segment in _TITLE_SEPARATORS.split(title):
+            segment = segment.strip(" ,.-")
+            if segment and len(segment) <= 60 and _segment_matches_domain(segment, stem):
+                return segment
+    return name_from_domain(domain) if stem else title
+
+
 def is_junk(domain: str, name: str = "") -> bool:
     """True when this row would pollute the database rather than fill it."""
     if domain:
@@ -387,7 +495,7 @@ async def fetch(self, client, env, query):
             if not domain:
                 continue
             out.append(Business(
-                name=item.get("title") or domain,
+                name=business_name(item.get("title") or "", domain),
                 domain=domain,
                 website=item.get("url") or "",
                 location=query.location,
@@ -438,7 +546,7 @@ async def fetch(self, client, env, query):
             if not domain:
                 continue
             out.append(Business(
-                name=item.get("title") or domain,
+                name=business_name(item.get("title") or "", domain),
                 domain=domain,
                 website=item.get("link") or "",
                 location=query.location,

harvey/collectors/profile.py

@@ -18,6 +18,10 @@
 
 import httpx
 
+from harvey.collectors.discover import (
+    name_from_domain, name_matches_domain, site_name_from_html,
+)
+
 logger = logging.getLogger("harvey.collectors.profile")
 
 TIMEOUT = httpx.Timeout(connect=8.0, read=12.0, write=8.0, pool=8.0)
@@ -122,13 +126,63 @@
 privacy policy terms conditions sitemap blog news events why choose us
 areas served locations location city county state north south east west
 español espanol menu skip navigation search close open toggle
+hoa hoas condo condos association associations board boards community
+communities management property properties owners residents customer care
+scope alignment implementation consistency communication process driven
+focused structured
 """.split())
 
+# Tokens that are words, not names, whatever the vertical: abstract nouns
+# ("Scope Alignment", "Board Communication") and shouted acronyms ("HOA").
+_NOUN_SUFFIX = re.compile(r"(tion|tions|sion|sions|ness)$")
+
+
+def _is_name_token(token: str) -> bool:
+    for part in token.split("-"):
+        bare = part.strip(".,'")
+        low = bare.lower()
+        if low in NON_NAME_WORDS or _NOUN_SUFFIX.search(low):
+            return False
+        if len(bare) >= 3 and bare.isupper():
+            return False
+    return True
+
+
+# Block-level tags separate a name, its title, and the bio that follows; the
+# first text chunk after the name heading is the title, nothing further.
+_BLOCK_BREAK = re.compile(r"(?i)<(?:/?(?:p|div|li|ul|ol|section|article|td|tr)\b[^>]*|br\s*/?)>")
+_CONNECTORS = frozenset("of and the for & to in at".split())
+
+
+def _title_chunks(raw_after: str) -> list[str]:
+    raw_after = re.sub(r"<[^>]*$", "", raw_after)   # a tag cut off by the window
+    chunks = []
+    for piece in _BLOCK_BREAK.split(raw_after):
+        text = html_unescape(re.sub(r"(?s)<[^>]+>", " ", piece))
+        text = re.sub(r"\s+", " ", text).strip(" ,.|-–—·")
+        if text:
+            chunks.append(text)
+    return chunks
+
+
+def _reads_like_title(text: str) -> bool:
+    """A job title is short and mostly capitalised; a sentence is neither.
+
+    "Desiring consistent oversight and a trusted management partner" contains
+    the role word "partner" but is marketing copy, not a title.
+    """
+    words = text.split()
+    if not words or len(words) > 10:
+        return False
+    lower = [w for w in words[1:] if w[:1].islower() and w.lower() not in _CONNECTORS]
+    return len(lower) <= 2 and title_looks_like_role(text)
+
 DECISION_MAKER_TITLES = (
     "owner", "founder", "co-founder", "president", "ceo", "principal",
     "partner", "general manager", "operations manager", "practice manager",
     "office manager", "marketing director", "marketing manager",
     "director of marketing", "vice president", "vp ", "managing director",
+    "chief", "director",
 )
 
 
@@ -177,6 +231,7 @@ def __init__(self, state, confirmed: set[str] | None = None, timeout: float = 12
         self.confirmed = confirmed if confirmed is not None else set()
         self.timeout = timeout
         self._client: httpx.AsyncClient | None = None
+        self.site_names: dict[str, str] = {}
 
     def wants(self, code: str) -> bool:
         return code in self.confirmed
@@ -284,9 +339,11 @@ def detect_people(self, html: str) -> list[dict]:
         text_html = _strip_code(html)
         people, seen = [], set()
         # Name in a heading, title in the nearby text.
+        # Tag names match any case; the NAME must not — under a blanket (?i)
+        # "Customer care" passes the Firstname Lastname capitalisation test.
         for m in re.finditer(
-            r"(?is)<(h[1-5]|strong|b)[^>]*>\s*([A-Z][a-z'\-]+(?:\s+[A-Z]\.?)?\s+"
-            r"[A-Z][a-zA-Z'\-]+)\s*</\1>(?=(.{0,220}))", text_html
+            r"(?s)<((?i:h[1-5]|strong|b))[^>]*>\s*([A-Z][a-z'\-]+(?:\s+[A-Z]\.?)?\s+"
+            r"[A-Z][a-zA-Z'\-]+)\s*</(?i:\1)>(?=(.{0,220}))", text_html
         ):
             # NOTE the lookahead above: capturing the trailing context normally
             # would CONSUME it, so finditer would skip every person whose entry
@@ -299,34 +356,23 @@ def detect_people(self, html: str) -> list[dict]:
             nxt = re.search(r"(?i)<(h[1-5]|strong|b)\b", raw_after)
             if nxt:
                 raw_after = raw_after[: nxt.start()]
-            after = html_unescape(re.sub(r"(?s)<[^>]+>", " ", raw_after))
-            after = re.sub(r"\s+", " ", after).strip()
             if name.lower() in seen or len(name) > 45:
                 continue
             # Reject obvious non-names
             if re.search(r"\d|@|\bthe\b|\bour\b|\bwe\b", name, re.I):
                 continue
             # Reject headings that merely LOOK like names. Every token must be
             # plausibly a name part; one business word disqualifies the whole
-            # match ("Request Service", "Baker History", "Commercial Services").
-            tokens = [t.strip(".,'-").lower() for t in name.split()]
-            if any(t in NON_NAME_WORDS for t in tokens):
+            # match ("Request Service", "Baker History", "HOA Boards").
+            if not all(_is_name_token(t) for t in name.split()):
                 continue
             # A real contact needs a title that reads like a role; a bare
-            # heading with no role text is almost always navigation.
-            if not title_looks_like_role(after):
-                continue
-            title = ""
-            low = after.lower()
-            for t in DECISION_MAKER_TITLES:
-                if t in low:
-                    idx = low.index(t)
-                    title = after[idx:idx + 60].strip(" ,.|-—·")
-                    break
+            # heading with no role text is almost always navigation, and a
+            # heading followed by a sentence is a feature list.
+            title = next((c for c in _title_chunks(raw_after)[:2]
+                          if _reads_like_title(c)), "")
             if not title:
-                m2 = re.match(r"[\s,|\-–—·]*([A-Za-z][A-Za-z /&'\-]{2,45})", after)
-                if m2:
-                    title = m2.group(1).strip()
+                continue
             seen.add(name.lower())
             parts = name.split()
             people.append({
@@ -359,6 +405,7 @@ def add(code, *, num=None, text="", conf=1.0, url=""):
             # A failure IS an observation — silent failures look like clean results.
             logger.info(f"profile: {domain} unreachable")
             return obs
+        self.site_names[company_id] = site_name_from_html(home, domain)
 
         # Ads (they are spending TODAY — the strongest budget signal here)
         add("RUNNING_GOOGLE_ADS", num=1 if self._any(home, GOOGLE_ADS) else 0)
@@ -455,6 +502,22 @@ def add(code, *, num=None, text="", conf=1.0, url=""):
 
         return obs
 
+    async def _improve_name(self, company: dict, domain: str):
+        """Replace a name the domain doesn't back up ("Denver, CO | HOA
+        Management Firm") or one merely derived from the domain ("Grow With
+        Us") with the one the homepage gives itself. A real match is kept."""
+        better = self.site_names.pop(company["id"], "")
+        current = company.get("name") or ""
+        fallback = current == name_from_domain(domain)
+        if not better or better == current or (
+                name_matches_domain(current, domain) and not fallback):
+            return
+        try:
+            await self.state.rename_company(company["id"], better)
+            logger.info(f"profile: {domain} named '{better}' (was '{current}')")
+        except Exception as e:
+            logger.debug(f"rename_company failed for {domain}: {e}")
+
     async def _mark_checked(self, company_id: str):
         try:
             await self.state.mark_company_profiled(company_id)
@@ -482,6 +545,7 @@ async def run(self, companies: list[dict], run_id: str = "") -> int:
             # as checked, or the same handful gets retried every cycle
             # forever instead of rotating through the rest of the backlog.
             await self._mark_checked(company["id"])
+            await self._improve_name(company, domain)
             if obs:
                 total += await self.state.add_observations(obs, run_id=run_id)
                 logger.info(f"profile: {domain} → {len(obs)} observation(s)")

harvey/state.py

@@ -1254,6 +1254,20 @@ async def mark_company_profiled(self, company_id: str):
             )
             await db.commit()
 
+    async def rename_company(self, company_id: str, name: str):
+        """Set a company's name, and the copy each of its prospects carries."""
+        async with self._connect() as db:
+            await db.execute(
+                "UPDATE companies SET name = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?",
+                (name, company_id),
+            )
+            await db.execute(
+                "UPDATE prospects SET company = ?, updated_at = CURRENT_TIMESTAMP "
+                "WHERE company_id = ?",
+                (name, company_id),
+            )
+            await db.commit()
+
     async def companies_needing_prospects(
         self, limit: int = 10, stale_days: int = 14
     ) -> list[dict]: