#!/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()