#!/usr/bin/env python3 """BCG Radar Lite: public-site research, not automated messaging.""" import csv import re import time from datetime import datetime, timezone from urllib.parse import urljoin, urlparse from urllib.robotparser import RobotFileParser import requests from bs4 import BeautifulSoup DIRECTORY = "https://members.asaonline.com/directory/Find" USER_AGENT = "BCG-Radar/1.0 (respectful business website research)" LIMIT = 40 SEEDS = [ "https://layneglass.com", "https://rm4hvac.com", "https://www.metroelectric-rgv.com", "https://sgkcontractinginc.com", "https://www.tmechtx.com", "https://www.bvmechanical.com", "https://www.gdacontractors.com", "https://marekbros.com", "https://alliedglazingmgmt.com", "https://www.prime-controls.com", "https://www.gbshades.com", "https://bakerbrotherselectric.com", "https://www.premierconcrete.biz", "https://dfwelectricgroup.com", "https://criticalpowergroup.com", ] GROUPS = { "Glazing": (18, ("curtain wall", "storefront", "glazing", "glass contractor")), "Mechanical": (18, ("mechanical contractor", "hvac", "ductwork", "piping", "sheet metal")), "Electrical": (16, ("electrical contractor", "electrical construction", "low voltage", "building automation")), "Interior trades": (16, ("millwork", "drywall", "casework", "acoustical ceilings")), "Structural trades": (12, ("structural steel", "steel fabrication", "concrete contractor", "reinforcing steel")), "Commercial projects": (16, ("commercial construction", "commercial projects", "healthcare projects", "school construction", "institutional")), "Technical drawings": (14, ("shop drawings", "revit", "bim coordination", "construction documents")), "Preconstruction": (10, ("preconstruction", "estimating", "quantity takeoff", "design-build")), } BLOCKED_DOMAINS = ("asaonline.com", "growthzone", "facebook.com", "linkedin.com", "instagram.com", "youtube.com", "google.com", "x.com", "twitter.com", "yelp.com") SESSION = requests.Session() SESSION.headers.update({"User-Agent": USER_AGENT, "Accept": "text/html,application/xhtml+xml"}) ROBOTS = {} def domain(url): return urlparse(url).netloc.lower().removeprefix("www.") def allowed(url): host = urlparse(url).netloc if not host: return False if host not in ROBOTS: robots_url = urljoin(f"{urlparse(url).scheme}://{host}/", "robots.txt") try: r = SESSION.get(robots_url, timeout=8) if r.status_code == 404 or r.status_code == 410: ROBOTS[host] = None # Explicitly no robots file elif r.status_code == 200: rp = RobotFileParser() rp.parse(r.text.splitlines()) ROBOTS[host] = rp else: ROBOTS[host] = False # Fail closed except requests.RequestException: ROBOTS[host] = False rule = ROBOTS[host] return rule is None or (rule is not False and rule.can_fetch(USER_AGENT, url)) def fetch(url): if not allowed(url): return None try: r = SESSION.get(url, timeout=(5, 12), stream=True, allow_redirects=True) if r.status_code != 200 or "html" not in r.headers.get("Content-Type", "").lower(): r.close() return None # 450 KB is enough for business research and keeps RAM very low. html = r.raw.read(450_000, decode_content=True) final_url = r.url r.close() if not allowed(final_url): return None return final_url, BeautifulSoup(html, "html.parser") except (requests.RequestException, ValueError): return None def discover(): urls = list(SEEDS) page = fetch(DIRECTORY) if page: _, soup = page for link in soup.select("a[href]"): href = urljoin(DIRECTORY, link.get("href", "")) host = domain(href) if not host or any(x in host for x in BLOCKED_DOMAINS): continue if urlparse(href).scheme not in ("http", "https"): continue if href.lower().endswith((".pdf", ".jpg", ".png", ".zip")): continue urls.append(f"{urlparse(href).scheme}://{urlparse(href).netloc}/") distinct, seen = [], set() for u in urls: if domain(u) not in seen: seen.add(domain(u)) distinct.append(u) return distinct[:LIMIT] def analyze(url): page = fetch(url) if not page: return None final_url, soup = page title = soup.title.get_text(" ", strip=True) if soup.title else domain(url) contact = "" for a in soup.select("a[href]"): label = a.get_text(" ", strip=True).lower() href = urljoin(final_url, a.get("href", "")) if "contact" in label and domain(href) == domain(final_url) and href.startswith("http"): contact = href break for element in soup(["script", "style", "nav", "footer", "noscript"]): element.decompose() text = re.sub(r"\s+", " ", soup.get_text(" ", strip=True)).lower()[:25000] matches = [label for label, (_, terms) in GROUPS.items() if any(term in text for term in terms)] score = min(100, sum(GROUPS[label][0] for label in matches)) return { "company_title": title[:100], "website": final_url, "service_fit_score": score, "fit_level": "HIGH" if score >= 45 else "MEDIUM" if score >= 22 else "LOW", "matching_signals": "; ".join(matches), "public_contact_page": contact, "evidence_url": final_url, "source_directory": DIRECTORY, "research_date_utc": datetime.now(timezone.utc).strftime("%Y-%m-%d"), "review_status": "Needs human verification", } def main(): urls = discover() print(f"Discovered {len(urls)} candidate websites; checking public pages...", flush=True) rows = [] for idx, url in enumerate(urls, 1): result = analyze(url) if result: rows.append(result) print(f"[{idx}/{len(urls)}] {domain(url)} -> {result['service_fit_score'] if result else 'unavailable'}", flush=True) time.sleep(0.8) rows.sort(key=lambda x: x["service_fit_score"], reverse=True) fields = ["company_title", "website", "service_fit_score", "fit_level", "matching_signals", "public_contact_page", "evidence_url", "source_directory", "research_date_utc", "review_status"] with open("bcg_leads.csv", "w", encoding="utf-8-sig", newline="") as f: writer = csv.DictWriter(f, fieldnames=fields) writer.writeheader() writer.writerows(rows) print(f"\nSaved {len(rows)} reviewed candidates to bcg_leads.csv") print("Scores indicate service fit, NOT proof of hiring need. No messages were sent.") if __name__ == "__main__": main()