# ============================================================ # SPID — project taxonomy & lightweight value extractors # # Shared by both the rule-based and the LLM extractor so that # `project_type`, `technology`, money, location and dates are # normalised the same way regardless of backend. # ============================================================ from __future__ import annotations import re # ---------------------------------------------------------------------- # Canonical project types. `signals` are high-precision regex fragments # (already lower-cased) used by the rule extractor and as the controlled # vocabulary the LLM is asked to classify into. # ---------------------------------------------------------------------- PROJECT_TYPES: dict[str, dict] = { "manufacturing_expansion": { "label": "Manufacturing / Capacity Expansion", "signals": [ r"(expand|increas\w+|add\w*|ramp\w*|scal\w+|new|build\w*|construct\w*|open\w*)" r"[^.]{0,60}(manufactur\w+|production|plant|factory|facilit\w+|capacity|assembly|fab\b)", ], }, "plant_construction": { "label": "Plant / Site Construction", "signals": [r"(construct\w*|build\w*|broke ground|commission\w*|new (plant|site|facility))" r"[^.]{0,60}(plant|facility|site|campus|complex|refinery|mill)"], }, "data_center": { "label": "Data Center", "signals": [r"data\s?cent(er|re)s?", r"hyperscal\w+", r"server farm"], }, "cloud_migration": { "label": "Cloud Migration", "signals": [r"(migrat\w+|mov\w+|transition\w*)[^.]{0,40}(cloud)", r"cloud (migration|platform|adoption|infrastructure|first)"], }, "ai_initiative": { "label": "AI / ML Initiative", "signals": [r"\b(artificial intelligence|machine learning|generative ai|gen ?ai|deep learning|" r"large language model|\bllm\b|neural network)\b", r"\bai[- ](powered|driven|enabled|capabilit\w+|platform|model|tools?)\b"], }, "digital_transformation": { "label": "Digital Transformation", "signals": [r"digital (transformation|transition|initiative|strateg\w+|platform)", r"digitali[sz]ation"], }, "erp_implementation": { "label": "ERP / Systems Implementation", "signals": [r"\b(erp|sap|oracle|workday|s/?4 ?hana)\b[^.]{0,40}(implement\w+|deploy\w+|migrat\w+|system|platform)", r"(implement\w+|deploy\w+|upgrad\w+)[^.]{0,40}\berp\b", r"enterprise resource planning"], }, "rd_program": { "label": "R&D Program", "signals": [r"research and development", r"\br&d\b", r"(develop\w+|invest\w+)[^.]{0,40}(next[- ]generation|new technolog\w+|new product)"], }, "product_launch": { "label": "New Product / Launch", "signals": [r"(launch\w*|introduc\w+|releas\w+|roll\w*[- ]?out|unveil\w+)" r"[^.]{0,50}(product|platform|service|offering|solution|model|line)", r"new product (line|launch|introduction)"], }, "drug_pipeline": { "label": "Drug / Clinical Pipeline", "signals": [r"(phase\s?(i{1,3}|1|2|3)\b)", r"clinical trial", r"\bfda (approval|clearance|submission)\b", r"(drug|therap\w+|candidate|molecule|pipeline)[^.]{0,40}(develop\w+|trial|approval)"], }, "ma_integration": { "label": "M&A / Integration", "signals": [r"(acqui\w+|merg\w+|takeover|business combination)", r"integration of [a-z]", r"purchase of [a-z]"], }, "geographic_expansion": { "label": "Geographic Expansion", "signals": [r"(expand\w*|enter\w*|launch\w*|establish\w*)[^.]{0,50}" r"(market|region|countr\w+|geograph\w+|international|overseas|abroad)", r"market entry"], }, "capex_program": { "label": "Capital Expenditure Program", "signals": [r"capital (expenditure|expenditures|investment|spending|program|project|budget)", r"\bcapex\b", r"invest\w+[^.]{0,30}(\$|usd)"], }, "supply_chain": { "label": "Supply Chain Initiative", "signals": [r"supply chain", r"(reshor\w+|nearshor\w+|onshor\w+|offshor\w+)", r"(distribution|logistics|fulfillment|warehouse|inventory)[^.]{0,40}" r"(network|center|optimi[sz]\w+|expansion|investment)"], }, "automation": { "label": "Automation / Robotics", "signals": [r"automat\w+", r"robot\w+", r"(industrial|warehouse) automation"], }, "sustainability": { "label": "Sustainability / ESG", "signals": [r"sustainabilit\w+", r"\besg\b", r"net[- ]zero", r"carbon (neutral|reduction|footprint|emissions?)", r"decarboni[sz]\w+", r"emission reduction"], }, "energy_transition": { "label": "Energy Transition / Renewables", "signals": [r"renewable (energy|power)", r"(solar|wind|hydrogen|geothermal|battery storage)[^.]{0,40}" r"(project|facility|farm|plant|capacity|investment)", r"clean energy", r"energy transition"], }, "infrastructure": { "label": "Infrastructure Investment", "signals": [r"infrastructure (investment|project|upgrade|expansion|spending|build\w*)", r"(network|grid|pipeline|broadband|fiber) (expansion|build\w*|upgrade|deployment)"], }, "cost_reduction": { "label": "Cost Reduction / Restructuring", "signals": [r"(cost (reduction|savings|cutting)|restructur\w+|efficiency (program|initiative)|" r"productivity (program|initiative)|streamlin\w+|optimi[sz]ation (program|initiative))"], }, "partnership": { "label": "Strategic Partnership", "signals": [r"(strategic (partnership|alliance|collaboration)|joint venture|\bjv\b|" r"(partner\w+|collaborat\w+|team\w* up)[^.]{0,40}(with))"], }, "technology_deployment": { "label": "Technology Deployment", "signals": [r"(deploy\w+|implement\w+|roll\w*[- ]?out|adopt\w+|install\w+)[^.]{0,50}" r"(technolog\w+|system|platform|software|solution|tool)"], }, } # ---------------------------------------------------------------------- # Technology vocabulary -> canonical tag # ---------------------------------------------------------------------- TECHNOLOGY_TERMS: dict[str, str] = { r"\b(artificial intelligence|machine learning|generative ai|gen ?ai|deep learning|llm|neural)\b": "AI/ML", r"\bcloud\b": "Cloud", r"\b(erp|sap|oracle|workday|s/?4 ?hana|peoplesoft)\b": "ERP", r"\b(robot\w+|automation|cobots?)\b": "Robotics/Automation", r"\b(semiconductor|chip|wafer|fab|node\b|nanometer|foundry)\b": "Semiconductors", r"\b(solar|wind|renewable|hydrogen|geothermal)\b": "Renewable Energy", r"\b(battery|batteries|energy storage|lithium)\b": "Battery/Storage", r"\b(ev|electric vehicle|charging)\b": "Electric Vehicles", r"\b(blockchain|crypto|digital asset|tokeni[sz])\b": "Blockchain", r"\b(5g|broadband|fiber|fibre|network infrastructure)\b": "Connectivity", r"\b(data\s?cent(er|re)|hyperscale|server)\b": "Data Center", r"\b(cyber ?security|encryption|zero trust)\b": "Cybersecurity", r"\b(iot|internet of things|sensors?|edge computing)\b": "IoT/Edge", r"\b(quantum)\b": "Quantum", r"\b(biolog\w+|gene|mrna|crispr|cell therapy|molecul\w+)\b": "Biotech", r"\b(saas|software[- ]as[- ]a[- ]service|platform)\b": "Software/SaaS", } # ---------------------------------------------------------------------- # Status keywords -> canonical status # ---------------------------------------------------------------------- STATUS_TERMS: list[tuple[str, str]] = [ (r"\b(completed|commission\w+|operational|in service|came online|opened|finalized)\b", "completed"), (r"\b(under construction|underway|in progress|ongoing|ramping|being (built|implemented))\b", "in_progress"), (r"\b(announc\w+|plan\w+|intend\w+|expect\w+|will (build|invest|launch|open)|propos\w+|" r"anticipat\w+|target\w+|by (20\d\d)|slated)\b", "planned"), ] # ---------------------------------------------------------------------- # Geography # ---------------------------------------------------------------------- US_STATES = { "alabama","alaska","arizona","arkansas","california","colorado","connecticut","delaware", "florida","georgia","hawaii","idaho","illinois","indiana","iowa","kansas","kentucky", "louisiana","maine","maryland","massachusetts","michigan","minnesota","mississippi", "missouri","montana","nebraska","nevada","new hampshire","new jersey","new mexico", "new york","north carolina","north dakota","ohio","oklahoma","oregon","pennsylvania", "rhode island","south carolina","south dakota","tennessee","texas","utah","vermont", "virginia","washington","west virginia","wisconsin","wyoming", } COUNTRIES = { "united states","u.s.","usa","canada","mexico","china","india","japan","germany","france", "united kingdom","u.k.","ireland","brazil","singapore","south korea","korea","taiwan", "vietnam","thailand","malaysia","indonesia","australia","spain","italy","netherlands", "switzerland","sweden","poland","israel","saudi arabia","united arab emirates","uae", "philippines","czech republic","hungary","turkey","egypt","south africa","argentina","chile", } _MONEY_RE = re.compile( r"(?:US)?\$\s?(\d[\d,]*\.?\d*)\s?(billion|bn|million|mm|mn|m\b|thousand|k\b|trillion)?", re.IGNORECASE, ) _MULT = { "trillion": 1e12, "billion": 1e9, "bn": 1e9, "million": 1e6, "mm": 1e6, "mn": 1e6, "m": 1e6, "thousand": 1e3, "k": 1e3, } _YEAR_RE = re.compile(r"\b(?:fiscal\s+|fy\s?)?(20[0-3]\d|19[89]\d)\b", re.IGNORECASE) _QUARTER_RE = re.compile(r"\bQ([1-4])\s?(20\d\d)\b", re.IGNORECASE) _TECH_COMPILED = [(re.compile(p, re.IGNORECASE), tag) for p, tag in TECHNOLOGY_TERMS.items()] _STATUS_COMPILED = [(re.compile(p, re.IGNORECASE), s) for p, s in STATUS_TERMS] _CITY_STATE_RE = re.compile(r"\bin\s+([A-Z][a-zA-Z.]+(?:\s[A-Z][a-zA-Z.]+){0,2}),\s+([A-Z][a-zA-Z]+)") def extract_money(text: str) -> tuple[float | None, str | None]: """Return (usd_amount, raw_span) for the largest explicit dollar figure.""" best_val, best_raw = None, None for m in _MONEY_RE.finditer(text): try: num = float(m.group(1).replace(",", "")) except ValueError: continue unit = (m.group(2) or "").lower().strip() val = num * _MULT.get(unit, 1.0) if best_val is None or val > best_val: best_val, best_raw = val, m.group(0).strip() return best_val, best_raw def extract_locations(text: str) -> list[str]: low = text.lower() found: list[str] = [] for st in US_STATES: if re.search(rf"\b{re.escape(st)}\b", low): found.append(st.title()) for co in COUNTRIES: if re.search(rf"\b{re.escape(co)}\b", low): found.append(co.upper() if len(co) <= 4 else co.title()) for m in _CITY_STATE_RE.finditer(text): found.append(f"{m.group(1)}, {m.group(2)}") # de-dup preserving order seen, out = set(), [] for f in found: k = f.lower() if k not in seen: seen.add(k); out.append(f) return out[:6] def extract_years(text: str) -> list[int]: yrs = {int(y) for y in _YEAR_RE.findall(text)} for m in _QUARTER_RE.finditer(text): yrs.add(int(m.group(2))) return sorted(yrs) def extract_technologies(text: str) -> list[str]: tags = [] for rx, tag in _TECH_COMPILED: if rx.search(text) and tag not in tags: tags.append(tag) return tags def detect_status(text: str) -> str: for rx, status in _STATUS_COMPILED: if rx.search(text): return status return "mentioned" # pre-compile project-type signals TYPE_SIGNALS = { ptype: [re.compile(sig, re.IGNORECASE) for sig in spec["signals"]] for ptype, spec in PROJECT_TYPES.items() } TYPE_LABELS = {k: v["label"] for k, v in PROJECT_TYPES.items()} ALL_TYPES = list(PROJECT_TYPES.keys())