| 1 | """Hiring Signals analysis from normalized jobs SourceItems.""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | import re |
| 6 | from collections import Counter, defaultdict |
| 7 | from typing import Any |
| 8 | |
| 9 | from . import schema |
| 10 | |
| 11 | |
| 12 | THEME_KEYWORDS: dict[str, tuple[str, ...]] = { |
| 13 | "enterprise readiness": ( |
| 14 | "enterprise", "soc 2", "sso", "security", "compliance", "procurement", |
| 15 | "admin", "governance", "audit", |
| 16 | ), |
| 17 | "go-to-market": ( |
| 18 | "sales", "account executive", "customer success", "solutions", "partnership", |
| 19 | "revenue", "demand generation", "marketing", |
| 20 | ), |
| 21 | "ai and machine learning": ( |
| 22 | "machine learning", "ml", "ai", "llm", "model", "research scientist", |
| 23 | "applied scientist", "data scientist", |
| 24 | ), |
| 25 | "infrastructure and reliability": ( |
| 26 | "infrastructure", "platform", "devops", "sre", "reliability", "cloud", |
| 27 | "distributed systems", "backend", |
| 28 | ), |
| 29 | "product expansion": ( |
| 30 | "product manager", "product designer", "growth", "activation", "mobile", |
| 31 | "frontend", "design", |
| 32 | ), |
| 33 | "data and analytics": ( |
| 34 | "data", "analytics", "business intelligence", "warehouse", "etl", |
| 35 | "insights", |
| 36 | ), |
| 37 | } |
| 38 | |
| 39 | SENIORITY_TERMS = ("head of", "director", "vp", "principal", "staff", "lead", "founding") |
| 40 | |
| 41 | # Leadership markers that establish/own a function (a first-of-function hire). |
| 42 | LEADERSHIP_TERMS = ("head of", "chief", "global head", "svp", "vp of", "vp,", "director of") |
| 43 | |
| 44 | # Title qualifiers that are level/logistics noise, not a specialized capability. |
| 45 | _GENERIC_QUALIFIERS = { |
| 46 | "senior", "staff", "principal", "lead", "junior", "mid", "sr", "jr", |
| 47 | "i", "ii", "iii", "iv", "remote", "hybrid", "onsite", "on-site", |
| 48 | "contract", "intern", "full-time", "part-time", "us", "uk", "emea", |
| 49 | } |
| 50 | |
| 51 | |
| 52 | def analyze( |
| 53 | items: list[schema.SourceItem], |
| 54 | *, |
| 55 | explicit: bool, |
| 56 | topic: str = "", |
| 57 | ) -> dict[str, Any]: |
| 58 | """Return a structured Hiring Signals summary for report artifacts.""" |
| 59 | if not items: |
| 60 | return { |
| 61 | "mode": "explicit" if explicit else "standard", |
| 62 | "company_size_tier": "unknown", |
| 63 | "include": False, |
| 64 | "signals": [], |
| 65 | "strategic_candidates": [], |
| 66 | "omitted_reason": "no current public jobs evidence found", |
| 67 | } |
| 68 | |
| 69 | size_tier = infer_company_size(items, topic=topic) |
| 70 | themes = _theme_items(items) |
| 71 | signals = [_build_signal(theme, theme_items, size_tier) for theme, theme_items in themes.items()] |
| 72 | signals = [signal for signal in signals if signal["evidence_count"] > 0] |
| 73 | signals.sort(key=lambda s: (s["confidence_score"], s["evidence_count"]), reverse=True) |
| 74 | |
| 75 | # Strategic single-role signals are NOT count-gated: a founding or |
| 76 | # first-of-function role can outweigh a department's worth of headcount. |
| 77 | # The engine only FLAGS these; the reasoning model judges true novelty |
| 78 | # (e.g. whether "Human Simulation" is a new bet for this company). |
| 79 | strategic_candidates = _strategic_candidates(items) |
| 80 | |
| 81 | include = ( |
| 82 | bool(signals) or bool(strategic_candidates) |
| 83 | ) if explicit else any(_passes_standard_threshold(signal, size_tier) for signal in signals) |
| 84 | if not explicit: |
| 85 | signals = [signal for signal in signals if _passes_standard_threshold(signal, size_tier)] |
| 86 | |
| 87 | return { |
| 88 | "mode": "explicit" if explicit else "standard", |
| 89 | "company_size_tier": size_tier, |
| 90 | "include": include, |
| 91 | "signals": signals, |
| 92 | "strategic_candidates": strategic_candidates, |
| 93 | "omitted_reason": "" if include else _omitted_reason(items, size_tier, signals), |
| 94 | } |
| 95 | |
| 96 | |
| 97 | def infer_company_size(items: list[schema.SourceItem], *, topic: str = "") -> str: |
| 98 | """Infer a coarse company-size tier from jobs evidence.""" |
| 99 | topic_lower = topic.lower() |
| 100 | firmographic_text = " ".join( |
| 101 | " ".join([ |
| 102 | str(item.metadata.get("company_size") or ""), |
| 103 | topic, |
| 104 | ]) |
| 105 | for item in items |
| 106 | ).lower() |
| 107 | text = " ".join( |
| 108 | " ".join([ |
| 109 | item.title, |
| 110 | item.body[:1000], |
| 111 | str(item.metadata.get("company_size") or ""), |
| 112 | str(item.metadata.get("source_domain") or ""), |
| 113 | topic, |
| 114 | ]) |
| 115 | for item in items |
| 116 | ).lower() |
| 117 | count = len(items) |
| 118 | # Brand-name shortcut must match the COMPANY being researched (the topic), |
| 119 | # never the job-description body - JDs list enterprise customers (e.g. |
| 120 | # "trusted by Microsoft, Google"), which would misclassify a startup as |
| 121 | # mega-cap and suppress its real signals. |
| 122 | if re.search(r"\b(apple|uber|google|microsoft|amazon|meta|netflix)\b", topic_lower): |
| 123 | return "mega-cap" |
| 124 | if count >= 200 or re.search(r"\b(fortune 500|thousands of employees)\b", firmographic_text): |
| 125 | return "large-enterprise" |
| 126 | if count >= 35 or re.search(r"\b(series [cd]|public company)\b", text): |
| 127 | return "growth" |
| 128 | if count <= 12 or re.search(r"\b(founding|seed|series a|early[- ]stage|startup)\b", text): |
| 129 | return "startup" |
| 130 | return "mid-market" |
| 131 | |
| 132 | |
| 133 | def _theme_items(items: list[schema.SourceItem]) -> dict[str, list[schema.SourceItem]]: |
| 134 | themed: dict[str, list[schema.SourceItem]] = defaultdict(list) |
| 135 | for item in items: |
| 136 | text = f"{item.title} {item.body}".lower() |
| 137 | matched = False |
| 138 | for theme, keywords in THEME_KEYWORDS.items(): |
| 139 | if any(keyword in text for keyword in keywords): |
| 140 | themed[theme].append(item) |
| 141 | matched = True |
| 142 | if not matched: |
| 143 | dept = str(item.metadata.get("department") or item.container or "").strip().lower() |
| 144 | fallback = dept or "general hiring" |
| 145 | themed[fallback].append(item) |
| 146 | return dict(themed) |
| 147 | |
| 148 | |
| 149 | def _build_signal(theme: str, items: list[schema.SourceItem], size_tier: str) -> dict[str, Any]: |
| 150 | titles = [item.title for item in items if item.title] |
| 151 | departments = [ |
| 152 | str(item.metadata.get("department") or item.container or "").strip() |
| 153 | for item in items |
| 154 | if str(item.metadata.get("department") or item.container or "").strip() |
| 155 | ] |
| 156 | senior_roles = [ |
| 157 | title for title in titles |
| 158 | if any(term in title.lower() for term in SENIORITY_TERMS) |
| 159 | ] |
| 160 | strategic_count = sum(1 for title in titles if _is_strategic_title(title)) |
| 161 | score = _confidence_score( |
| 162 | len(items), len(set(departments)), len(senior_roles), size_tier, |
| 163 | strategic_count=strategic_count, |
| 164 | ) |
| 165 | evidence = [ |
| 166 | { |
| 167 | "title": item.title, |
| 168 | "url": item.url, |
| 169 | "department": item.metadata.get("department") or item.container or "", |
| 170 | "published_at": item.published_at, |
| 171 | } |
| 172 | for item in items[:5] |
| 173 | ] |
| 174 | return { |
| 175 | "theme": theme, |
| 176 | "interpretation": _interpretation(theme), |
| 177 | "confidence": _confidence_label(score), |
| 178 | "confidence_score": score, |
| 179 | "evidence_count": len(items), |
| 180 | "departments": [name for name, _count in Counter(departments).most_common(3)], |
| 181 | "senior_roles": senior_roles[:3], |
| 182 | "evidence": evidence, |
| 183 | } |
| 184 | |
| 185 | |
| 186 | def _confidence_score( |
| 187 | count: int, |
| 188 | department_count: int, |
| 189 | senior_count: int, |
| 190 | size_tier: str, |
| 191 | strategic_count: int = 0, |
| 192 | ) -> int: |
| 193 | # Count no longer dominates: founding/first-of-function and seniority can |
| 194 | # let a small cluster outrank a large generic one (a "new bet" beating |
| 195 | # "doubling down"). The reasoning model still makes the final novelty call. |
| 196 | score = count * 12 + min(department_count, 3) * 6 + senior_count * 10 + strategic_count * 14 |
| 197 | if size_tier == "startup": |
| 198 | score += 20 |
| 199 | elif size_tier == "mid-market": |
| 200 | score += 10 |
| 201 | elif size_tier == "growth": |
| 202 | score -= 5 |
| 203 | elif size_tier == "large-enterprise": |
| 204 | score -= 25 |
| 205 | elif size_tier == "mega-cap": |
| 206 | score -= 40 |
| 207 | return max(0, min(100, score)) |
| 208 | |
| 209 | |
| 210 | def _passes_standard_threshold(signal: dict[str, Any], size_tier: str) -> bool: |
| 211 | thresholds = { |
| 212 | "startup": (2, 50), |
| 213 | "mid-market": (3, 58), |
| 214 | "growth": (4, 65), |
| 215 | "large-enterprise": (6, 78), |
| 216 | "mega-cap": (8, 86), |
| 217 | "unknown": (3, 62), |
| 218 | } |
| 219 | min_count, min_score = thresholds.get(size_tier, thresholds["unknown"]) |
| 220 | return signal["evidence_count"] >= min_count and signal["confidence_score"] >= min_score |
| 221 | |
| 222 | |
| 223 | def _confidence_label(score: int) -> str: |
| 224 | if score >= 75: |
| 225 | return "high" |
| 226 | if score >= 50: |
| 227 | return "medium" |
| 228 | return "low" |
| 229 | |
| 230 | |
| 231 | def _interpretation(theme: str) -> str: |
| 232 | if theme == "general hiring": |
| 233 | return "hiring activity is visible, but the priority signal is diffuse" |
| 234 | return f"appears to be increasing focus on {theme}" |
| 235 | |
| 236 | |
| 237 | def _strategic_candidates(items: list[schema.SourceItem]) -> list[dict[str, Any]]: |
| 238 | """Flag individual roles worth surfacing regardless of how many share a theme. |
| 239 | |
| 240 | Pure structural detection (founding, first-of-function, specialized |
| 241 | qualifier, geographic novelty) - no semantic novelty judgment, which is |
| 242 | left to the reasoning model. Guarantees these roles reach synthesis instead |
| 243 | of being averaged away by count-weighting. |
| 244 | """ |
| 245 | item_locations = [(item, _norm_location(item)) for item in items] |
| 246 | location_counts = Counter(loc for _item, loc in item_locations if loc) |
| 247 | dominant = max(location_counts.values()) if location_counts else 0 |
| 248 | |
| 249 | scored: list[tuple[int, dict[str, Any]]] = [] |
| 250 | for item, location in item_locations: |
| 251 | flags = _title_flags(item.title or "") |
| 252 | if location and location_counts.get(location, 0) == 1 and dominant >= 3: |
| 253 | flags.append("new-geo") |
| 254 | if not flags: |
| 255 | continue |
| 256 | priority = ( |
| 257 | ("founding" in flags) * 4 |
| 258 | + ("new-geo" in flags) * 3 |
| 259 | + ("leadership" in flags) * 2 |
| 260 | + ("specialized" in flags) * 1 |
| 261 | ) |
| 262 | scored.append((priority, { |
| 263 | "title": item.title, |
| 264 | "url": item.url, |
| 265 | "department": str(item.metadata.get("department") or item.container or "").strip(), |
| 266 | "location": location, |
| 267 | "published_at": item.published_at, |
| 268 | "flags": flags, |
| 269 | })) |
| 270 | scored.sort(key=lambda pair: pair[0], reverse=True) |
| 271 | return [candidate for _priority, candidate in scored[:10]] |
| 272 | |
| 273 | |
| 274 | def _title_flags(title: str) -> list[str]: |
| 275 | """Structural strategic flags derivable from a title alone (no geo).""" |
| 276 | lowered = title.lower() |
| 277 | flags: list[str] = [] |
| 278 | if "founding" in lowered or re.search(r"\bfirst\b", lowered): |
| 279 | flags.append("founding") |
| 280 | if any(term in lowered for term in LEADERSHIP_TERMS): |
| 281 | flags.append("leadership") |
| 282 | if _specialization(title): |
| 283 | flags.append("specialized") |
| 284 | return flags |
| 285 | |
| 286 | |
| 287 | def _is_strategic_title(title: str) -> bool: |
| 288 | return bool(_title_flags(title)) |
| 289 | |
| 290 | |
| 291 | def _specialization(title: str) -> str: |
| 292 | """Return a specialized sub-domain qualifier from a title, or ''. |
| 293 | |
| 294 | "Research Scientist, Human Simulation" -> "Human Simulation". |
| 295 | "Engineer (Forward Deployed)" -> "Forward Deployed". |
| 296 | "Engineer, Senior" -> "" (generic level word, not a capability). |
| 297 | """ |
| 298 | tail = "" |
| 299 | paren = re.search(r"\(([^)]+)\)", title) |
| 300 | if paren: |
| 301 | tail = paren.group(1).strip() |
| 302 | elif "," in title: |
| 303 | tail = title.rsplit(",", 1)[1].strip() |
| 304 | if not tail or len(tail) < 4: |
| 305 | return "" |
| 306 | if tail.lower() in _GENERIC_QUALIFIERS: |
| 307 | return "" |
| 308 | return tail |
| 309 | |
| 310 | |
| 311 | def _norm_location(item: schema.SourceItem) -> str: |
| 312 | return str(item.metadata.get("location") or "").strip().lower() |
| 313 | |
| 314 | |
| 315 | def _omitted_reason( |
| 316 | items: list[schema.SourceItem], |
| 317 | size_tier: str, |
| 318 | signals: list[dict[str, Any]], |
| 319 | ) -> str: |
| 320 | if not items: |
| 321 | return "no current public jobs evidence found" |
| 322 | if size_tier in {"large-enterprise", "mega-cap"}: |
| 323 | return "jobs evidence is too diffuse for the inferred company size" |
| 324 | if not signals: |
| 325 | return "jobs evidence did not cluster into a clear signal" |
| 326 | return "jobs evidence is too thin for standard-report inclusion" |
| 327 |