1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
|
#!/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()
|
Cairnbin