{html.escape(r["owner_business"] or "Not listed")}Applicant: {html.escape(r["applicant_business"] or "Not listed")}
'
f'
{html.escape(r["service_matches"])}{description}
'
)
counts = {k: sum(r["priority"] == k for r in rows) for k in ("HIGH", "MEDIUM", "LOW")}
head = f"""
BCG NYC Project Intelligence
BCG / NYC Construction Project Intelligence
Public NYC DOB NOW filing research · Generated {html.escape(now)} UTC · Filing dates from {date_cutoff} onward
{len(rows)}Listings scanned
{counts['HIGH']}High technical fit
{counts['MEDIUM']}Medium technical fit
{new_count}New since previous scan
{update_count}Changed since previous scan
{"First run: baseline established; newly detected items will appear on later runs." if baseline else "Changes are tracked against the previous stored scan."}
The score is a project relevance heuristic, NOT evidence of available contracts, procurement, bid status or buyer intent. Cost and area are self-reported.
Fit
Project
Filing
Public organization
Potential service / description
"""
foot = """
"""
with open(dest, "w", encoding="utf-8") as f:
f.write(head + "\n".join(blocks) + foot)
def demo_test():
sample = {"job_filing_number": "M00999999-I1", "filing_date": "2026-11-01T00:00:00.000",
"filing_status": "Filed", "job_type": "Alteration", "job_description":
"Commercial office tenant HVAC and structural renovation", "borough": "Manhattan",
"house_no": "100", "street_name": "TEST STREET", "owner_s_business_name": "TEST OWNER",
"applicant_business_name": "TEST ARCHITECT", "initial_cost": "800000",
"mechanical_systems_work_type_": "YES", "structural_work_type_": "YES"}
score, p, services, reasons = classify(sample)
assert score >= 40 and "HVAC" in services
conn = sqlite3.connect(":memory:")
conn.execute("CREATE TABLE filings(job_filing_number TEXT PRIMARY KEY,raw_json TEXT,fingerprint TEXT,first_seen TEXT,last_seen TEXT,changed_at TEXT,last_change_type TEXT)")
events, first = save_records(conn, [sample], "2026-11-03T00:00:00")
assert first and events[sample["job_filing_number"]] == "BASELINE"
events, first = save_records(conn, [sample], "2026-11-03T01:00:00")
assert not first and events[sample["job_filing_number"]] == "UNCHANGED"
sample2 = dict(sample, filing_status="Approved")
events, first = save_records(conn, [sample2], "2026-11-03T02:00:00")
assert events[sample["job_filing_number"]] == "UPDATED"
assert safe_csv_cell("=HYPERLINK(1)").startswith("'")
print("Self-test passed: scoring, baseline, change detection, CSV safety.")
def main():
parser = argparse.ArgumentParser(description="BCG NYC Construction Early Warning")
parser.add_argument("--days", type=int, default=14, help="Look back days from UTC today (default: 14)")
parser.add_argument("--max-rows", type=int, default=1500, help="Max listings fetched per run (default: 1500)")
parser.add_argument("--page-size", type=int, default=250)
parser.add_argument("--out", default=".", help="Output directory, default current directory")
parser.add_argument("--self-test", action="store_true")
args = parser.parse_args()
if args.self_test:
demo_test()
return 0
if not 1 <= args.days <= 365 or not 1 <= args.max_rows <= 10000 or not 1 <= args.page_size <= 1000:
parser.error("days=1..365; max-rows=1..10000; page-size=1..1000")
folder = Path(args.out)
folder.mkdir(parents=True, exist_ok=True)
now = datetime.now(timezone.utc)
cutoff = (now - timedelta(days=args.days)).strftime("%Y-%m-%d")
print(f"BCG NYC Project Watch | UTC {now.strftime('%Y-%m-%d %H:%M')} | since {cutoff}")
print(f"NYC Open Data source: {DATASET_PAGE}")
try:
data = list(api_query(cutoff, args.page_size, args.max_rows))
except (RuntimeError, ValueError, json.JSONDecodeError) as exc:
print(f"ERROR: {exc}", file=sys.stderr)
print("Test connectivity: curl -6 -I --max-time 12 https://data.cityofnewyork.us/", file=sys.stderr)
return 2
if not data:
print("No filings returned for selected lookback; database left unchanged.")
return 0
with init_db(str(folder / "bcg_nyc.db")) as conn:
events, baseline = save_records(conn, data, now.isoformat(timespec="seconds"))
rows = [record_for_export(item, events.get(clean(item.get("job_filing_number")), "")) for item in data]
rows.sort(key=lambda r: (r["score"], r["filing_date"]), reverse=True)
changes = [row for row in rows if row["change_type"] in ("NEW", "UPDATED")]
# Output the full ranked sample and only newly seen or modified rows.
write_csv(folder / "bcg_nyc_opportunities.csv", rows)
write_csv(folder / "bcg_nyc_changes.csv", changes)
write_html(folder / "bcg_nyc_dashboard.html", rows, cutoff, now.strftime("%Y-%m-%d %H:%M"),
baseline, sum(e == "NEW" for e in events.values()), sum(e == "UPDATED" for e in events.values()))
print(f"Rows evaluated: {len(rows)} | High: {sum(r['priority'] == 'HIGH' for r in rows)} | Medium: {sum(r['priority'] == 'MEDIUM' for r in rows)}")
print(f"New: {sum(e == 'NEW' for e in events.values())} | Updated: {sum(e == 'UPDATED' for e in events.values())} | Baseline: {baseline}")
print("Files: bcg_nyc_opportunities.csv, bcg_nyc_changes.csv, bcg_nyc_dashboard.html, bcg_nyc.db")
for r in rows[:5]:
print(f" {r['score']:>3} {r['priority']:<6} {r['borough']:<12} {r['address'][:35]:<35} {r['job_filing_number']}")
return 0
if __name__ == "__main__":
sys.exit(main())