← back to Labj Cre Banks
scripts/build.py
346 lines
#!/usr/bin/env python3
"""
LABJ CRE + Banking public-source dataset builder.
Combines the LABJ public seed (data/seed.csv) with independently-researched
public company records (data/enriched/*.json) into a normalized, deduped,
RENTV-advertiser-scored master dataset for the RE builds.
Legal posture: LABJ "The Lists" are used ONLY as a discovery/ranking pointer.
Every field is rebuilt from independent public sources with attribution. No
LABJ paid product is reproduced. Fields are never fabricated — a missing value
stays blank.
Stdlib only (no pandas). Re-runnable & idempotent.
"""
import csv, json, glob, os, re, datetime
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
DATA = os.path.join(ROOT, "data")
ENRICHED = os.path.join(DATA, "enriched")
REPORTS = os.path.join(ROOT, "reports")
TODAY = datetime.date.today().isoformat()
# ---- category → canonical LABJ list name + RENTV relevance flags ----------
# is_cre_core: brokerage/dev/pm/contractor/architecture = CRE core business
# is_lender: bank/financial/credit-union serving CRE
CRE_CORE = {
"Commercial Real Estate Brokerage Firms",
"Commercial Real Estate Developers",
"Residential Real Estate Developers",
"Residential Brokerage Firms",
"Property Management Firms",
"Property Management Firms - Office",
"Property Management Firms - Retail",
"Property Management Firms - Industrial",
"General Contractors",
"Architecture Firms",
}
LENDER = {"Banks", "Financial Institutions", "Credit Unions",
"SBA 7(a) Lenders", "SBA 504 Lenders"}
def norm_domain(url):
if not url:
return ""
u = url.strip().lower()
u = re.sub(r"^https?://", "", u)
u = re.sub(r"^www\.", "", u)
u = u.split("/")[0].split("?")[0]
return u.strip()
def norm_name(name):
if not name:
return ""
n = name.lower().strip()
n = re.sub(r"[^a-z0-9]+", " ", n)
for suf in (" inc", " llc", " lp", " group", " co", " company", " corp", " bank",
" corporation", " services", " builders", " partners"):
if n.endswith(suf):
n = n[: -len(suf)]
return n.strip()
def load_seed():
rows = []
with open(os.path.join(DATA, "seed.csv"), newline="", encoding="utf-8") as f:
for r in csv.DictReader(f):
rows.append({
"company_name": r.get("company", "").strip(),
"aliases": "",
"category": r.get("labj_list", "").strip(),
"labj_rank": r.get("rank", "").strip(),
"address": r.get("address", "").strip(),
"city": r.get("city", "").strip(),
"state": r.get("state", "").strip(),
"zip": r.get("zip", "").strip(),
"website": r.get("website", "").strip(),
"main_phone": r.get("phone", "").strip(),
"top_local_executive": r.get("top_executive", "").strip(),
"exec_title": r.get("title", "").strip(),
"marketing_pr_contact": "",
"marketing_pr_title": "",
"public_email": "",
"linkedin_company_url": "",
"specialties": "",
"la_presence": "",
"key_metric": r.get("key_metric", "").strip(),
"source_labj": r.get("source_url", "").strip(),
"source_company": "",
"source_contact": "",
"last_verified": TODAY,
"_origin": "labj_seed",
})
return rows
def load_enriched():
rows = []
if not os.path.isdir(ENRICHED):
return rows
for path in sorted(glob.glob(os.path.join(ENRICHED, "*.json"))):
try:
with open(path, encoding="utf-8") as f:
payload = json.load(f)
except Exception as e:
print(f" ! skipping {os.path.basename(path)}: {e}")
continue
items = payload.get("companies", payload) if isinstance(payload, dict) else payload
for it in items:
srcs = it.get("sources") or []
if isinstance(srcs, list):
srcs = "; ".join(srcs)
rows.append({
"company_name": (it.get("company_name") or "").strip(),
"aliases": (it.get("aliases") or "").strip() if isinstance(it.get("aliases"), str) else "; ".join(it.get("aliases") or []),
"category": (it.get("category") or "").strip(),
"labj_rank": str(it.get("labj_rank") or "").strip(),
"address": (it.get("address") or "").strip(),
"city": (it.get("city") or "").strip(),
"state": (it.get("state") or "").strip(),
"zip": str(it.get("zip") or "").strip(),
"website": (it.get("website") or "").strip(),
"main_phone": (it.get("main_phone") or "").strip(),
"top_local_executive": (it.get("top_local_executive") or "").strip(),
"exec_title": (it.get("exec_title") or "").strip(),
"marketing_pr_contact": (it.get("marketing_pr_contact") or "").strip(),
"marketing_pr_title": (it.get("marketing_pr_title") or "").strip(),
"public_email": (it.get("public_email") or "").strip(),
"linkedin_company_url": (it.get("linkedin_company_url") or "").strip(),
"specialties": (it.get("specialties") or "").strip() if isinstance(it.get("specialties"), str) else "; ".join(it.get("specialties") or []),
"la_presence": (it.get("la_presence") or "").strip(),
"key_metric": (it.get("key_metric") or "").strip(),
"source_labj": (it.get("source_labj") or "").strip(),
"source_company": srcs,
"source_contact": (it.get("source_contact") or "").strip(),
"last_verified": TODAY,
"_origin": "researched",
})
return rows
def merge(rows):
"""Dedup by domain OR normalized name (union-find). Preserve multi-list membership.
Merging on EITHER identifier catches both same-firm/two-domains (Watson Land
Co. / Watson Land Company) and same-domain/two-names (Colliers / Colliers
International).
"""
parent = {}
def find(x):
parent.setdefault(x, x)
root = x
while parent[root] != root:
root = parent[root]
while parent[x] != root:
parent[x], x = root, parent[x]
return root
def union(a, b):
parent[find(a)] = find(b)
row_ids = []
for r in rows:
if not r["company_name"]:
row_ids.append(None); continue
ids = []
dm, nm = norm_domain(r["website"]), norm_name(r["company_name"])
if dm: ids.append(("d", dm))
if nm: ids.append(("n", nm))
for i in ids[1:]:
union(ids[0], i)
row_ids.append(ids[0] if ids else None)
by_key = {}
memberships = []
for r, rid in zip(rows, row_ids):
if rid is None:
continue
key = find(rid)
if r["category"]:
memberships.append((key, r["company_name"], r["category"],
r["labj_rank"], r["source_labj"]))
if key not in by_key:
r = dict(r)
r["_key"] = key
r["_categories"] = set()
by_key[key] = r
cur = by_key[key]
if r["category"]:
cur["_categories"].add(r["category"])
# prefer the most complete company_name (longest) as canonical
if len(r.get("company_name", "")) > len(cur.get("company_name", "")):
cur["company_name"] = r["company_name"]
for fld in ("address", "city", "state", "zip", "website", "main_phone",
"top_local_executive", "exec_title", "marketing_pr_contact",
"marketing_pr_title", "public_email", "linkedin_company_url",
"specialties", "la_presence", "key_metric", "aliases",
"source_company", "source_labj"):
if not cur.get(fld) and r.get(fld):
cur[fld] = r[fld]
return by_key, memberships
def score(company):
cats = company["_categories"]
pts = 0
reasons = []
if any(c in CRE_CORE for c in cats):
pts += 25; reasons.append("CRE core business (+25)")
la = (company.get("state") == "CA") or ("los angeles" in (company.get("la_presence", "").lower())) \
or company.get("city", "") != ""
if la:
pts += 20; reasons.append("active LA presence (+20)")
if company.get("marketing_pr_contact") or company.get("public_email"):
pts += 15; reasons.append("marketing/PR contact identified (+15)")
if len([c for c in cats if c]) > 1:
pts += 15; reasons.append("multiple relevant lists (+15)")
if any(c in LENDER for c in cats):
pts += 10; reasons.append("lender/bank serving CRE (+10)")
if company.get("key_metric"):
pts += 10; reasons.append("major transaction/development footprint (+10)")
# advertising evidence: only credited when a researcher flagged it in la_presence/specialties
blob = (company.get("specialties", "") + " " + company.get("la_presence", "")).lower()
if any(k in blob for k in ("advertis", "sponsor")):
pts += 5; reasons.append("public advertising/sponsorship evidence (+5)")
return min(pts, 100), "; ".join(reasons)
MASTER_COLS = ["company_name", "aliases", "categories", "labj_rank", "address",
"city", "state", "zip", "website", "main_phone",
"top_local_executive", "exec_title", "marketing_pr_contact",
"marketing_pr_title", "public_email", "linkedin_company_url",
"specialties", "la_presence", "key_metric",
"advertiser_fit_score", "advertiser_fit_reason",
"source_labj", "source_company", "origin", "last_verified"]
def write_csv(path, cols, rows):
with open(path, "w", newline="", encoding="utf-8") as f:
w = csv.DictWriter(f, fieldnames=cols, extrasaction="ignore")
w.writeheader()
for r in rows:
w.writerow(r)
def main():
os.makedirs(REPORTS, exist_ok=True)
seed = load_seed()
enriched = load_enriched()
all_rows = seed + enriched
by_key, memberships = merge(all_rows)
master = []
for key, c in by_key.items():
s, reason = score(c)
master.append({
**{k: c.get(k, "") for k in MASTER_COLS if k not in
("categories", "advertiser_fit_score", "advertiser_fit_reason", "origin")},
"categories": " | ".join(sorted(x for x in c["_categories"] if x)),
"advertiser_fit_score": s,
"advertiser_fit_reason": reason,
"origin": c.get("_origin", ""),
})
master.sort(key=lambda r: (-r["advertiser_fit_score"], r["company_name"].lower()))
# 1. master (CSV + JSON for JS-based RE builds to consume directly)
write_csv(os.path.join(DATA, "labj_master.csv"), MASTER_COLS, master)
payload = {"generated": TODAY, "count": len(master), "records": master}
with open(os.path.join(DATA, "labj_master.json"), "w", encoding="utf-8") as f:
json.dump(payload, f, indent=2, ensure_ascii=False)
pub = os.path.join(ROOT, "public")
if os.path.isdir(pub): # keep the viewer's copy in sync
with open(os.path.join(pub, "labj_master.json"), "w", encoding="utf-8") as f:
json.dump(payload, f, indent=2, ensure_ascii=False)
# 2. companies (identity/firmographic subset)
comp_cols = ["company_name", "aliases", "website", "address", "city", "state",
"zip", "main_phone", "specialties", "la_presence", "categories",
"advertiser_fit_score", "source_company", "last_verified"]
write_csv(os.path.join(DATA, "labj_companies.csv"), comp_cols, master)
# 3. contacts (only rows with a real contact)
contact_cols = ["company_name", "website", "top_local_executive", "exec_title",
"marketing_pr_contact", "marketing_pr_title", "public_email",
"linkedin_company_url", "source_contact", "last_verified"]
contact_rows = []
for r in master:
if r.get("top_local_executive") or r.get("marketing_pr_contact") or r.get("public_email"):
contact_rows.append({
"company_name": r["company_name"], "website": r["website"],
"top_local_executive": r.get("top_local_executive", ""),
"exec_title": r.get("exec_title", ""),
"marketing_pr_contact": r.get("marketing_pr_contact", ""),
"marketing_pr_title": r.get("marketing_pr_title", ""),
"public_email": r.get("public_email", ""),
"linkedin_company_url": r.get("linkedin_company_url", ""),
"source_contact": "",
"last_verified": TODAY,
})
write_csv(os.path.join(DATA, "labj_contacts.csv"), contact_cols, contact_rows)
# 4. list memberships
mem_cols = ["company_name", "category", "labj_rank", "source_labj"]
mem_rows = [{"company_name": m[1], "category": m[2], "labj_rank": m[3],
"source_labj": m[4]} for m in memberships if m[2]]
write_csv(os.path.join(DATA, "labj_list_memberships.csv"), mem_cols, mem_rows)
# 5. research queue (records missing key fields)
rq_cols = ["company_name", "website", "missing_fields", "categories", "advertiser_fit_score"]
rq = []
for r in master:
missing = [f for f in ("website", "main_phone", "address",
"top_local_executive", "marketing_pr_contact")
if not r.get(f)]
if missing:
rq.append({"company_name": r["company_name"], "website": r["website"],
"missing_fields": ", ".join(missing),
"categories": r["categories"],
"advertiser_fit_score": r["advertiser_fit_score"]})
write_csv(os.path.join(DATA, "research_queue.csv"), rq_cols, rq)
# 6. top 100 RENTV prospects
write_csv(os.path.join(REPORTS, "top_100_rentv_prospects.csv"), MASTER_COLS,
master[:100])
# 7. source audit
sa_cols = ["company_name", "source_labj", "source_company", "origin"]
write_csv(os.path.join(REPORTS, "source_audit.csv"), sa_cols,
[{"company_name": r["company_name"], "source_labj": r["source_labj"],
"source_company": r["source_company"], "origin": r["origin"]}
for r in master])
# summary
cats = {}
for r in master:
for c in r["categories"].split(" | "):
if c:
cats[c] = cats.get(c, 0) + 1
print(f"companies (deduped): {len(master)}")
print(f"contacts: {len(contact_rows)}")
print(f"list memberships: {len(mem_rows)}")
print(f"research queue: {len(rq)}")
print(f"categories: {len(cats)}")
print("top 10 RENTV prospects:")
for r in master[:10]:
print(f" {r['advertiser_fit_score']:3d} {r['company_name']} [{r['categories']}]")
return master, cats
if __name__ == "__main__":
main()