deepline-scoring

__pycache__/
*.pyc

{
  "name": "deepline-scoring",
  "private": true,
  "type": "module",
  "engines": {
    "node": ">=22.18"
  }
}

# Research how buyers describe the problem

Use this before expanding a keyword catalog. The output is a source-backed candidate bank, not scoring weights. Follow the two-wave, source-specific research pattern from `last30days` and `deepline-research`: resolve the audience and communities, discover public discussions, follow the useful leads, and only then choose collection routes. This guide is self-contained; neither skill is a runtime dependency.

## Plan bounded retrieval

Record the product use case, buyer role, geography/language, research date, intended date window, query/result caps and budget. Translate product positioning into tasks, failures, workarounds and desired outcomes. Search buyer problems as well as brands; a generic acronym or a literal copy of the user's prompt often retrieves the wrong topic. Resolve ambiguous entities and category communities from public evidence.

Start with a small parallel first wave across relevant source families. Include a community source and a source with enough context to interpret the problem. Do not require every platform or invent examples to fill an empty family.

| Source                                          | First-wave query pattern                                                                         | Follow-up from observed evidence                                                                    |
| ----------------------------------------------- | ------------------------------------------------------------------------------------------------ | --------------------------------------------------------------------------------------------------- |
| Reddit                                          | Workflow + failure/workaround; category and peer communities                                     | Read the original thread and relevant comments; follow exact pain phrases and discovered subreddits |
| X                                               | Short task/phrase variants and an explicit date window using the chosen route's supported syntax | Resolve authors/handles; inspect original posts, replies and quoted context                         |
| Niche forums, Hacker News, public GitHub issues | Actual integration, workflow or operational symptom                                              | Follow linked issues, reproduction details and attempted fixes                                      |
| Video/transcripts and reviews                   | Task + implementation/review/problem                                                             | Extract dated passages and audience context, not a title alone                                      |
| First-party docs, jobs and case studies         | Domain + workflow, duty or integration                                                           | Verify what the company offers, uses, hires for or attributes to a customer                         |

Second-wave queries must name the first-wave evidence that motivated them. Use exact phrases, discovered communities, competing approaches and negative cases. For example, `CRM enrichment` can expand to account matching, orphaned subsidiaries and territory conflicts; `identity verification` can expand to delayed codes, repeated verification and signup abandonment. These are query hypotheses until a source supports them. Do not hardcode them as universal signals.

Cache each query and source snapshot in the project working directory. Retrieve enough context once and reuse it for extraction, adjudication and tests. Deduplicate canonical URLs/source IDs, reposts and quoted/copied text before counting independent observations. Prefer the next query with the highest expected new concepts or missing context; stop at the declared budget or low marginal useful evidence. Report queries, retrieved/usable/duplicate rows and new concepts per wave. Search rank and engagement help prioritize reading; neither becomes account-fit points.

## Preserve evidence before interpreting it

Keep a research bank separate from the analyzer's category-to-list configs. Each row needs:

- source family, canonical URL/source ID, parent thread and author/organization when publicly supported;
- source publication time, retrieval time, date certainty and requested-window membership; unknown dates stay unknown;
- exact short quote and surrounding context, distinct from normalized wording and proposed aliases;
- buyer/persona/use case when supported, plus speaker stance: first-person operator, customer, vendor, consultant, secondhand or unknown;
- problem, workaround, desired outcome and concept family; interpretation and confidence separately;
- entity role: whose problem/use/deployment the text describes, with `account_id: null` unless identity is verified;
- polarity: affirmative, negated, hypothetical, historical/resolved or uncertain; include rejected interpretations and reasons;
- collection status, extraction quality, deduplication/cluster key and engagement when available.

Open originals where possible. A search snippet is discovery evidence with limited context, not a verified feature. A recent search filter does not establish that a post is recent. Keep older useful vocabulary in a historical bucket; do not count it in a last-30-days finding. Quoted vendor claims, syndication and engagement from one discussion do not establish independent agreement. Public discussions never inherit private CRM identities or outcomes from a guess.

Report coverage per source: usable, weak, empty, unavailable, error or not relevant, with the reason. Missing X access is a coverage gap; it is not evidence that buyers are silent. Count unique discussions/authors where known and independent accounts only where verified. Do not treat social-post volume as account prevalence.

## Choose collection routes after the public pass

Search and describe the live Deepline catalog for the useful source families. Prefer supported API or managed routes that retain IDs, timestamps, comments and provenance. Use Apify when a reviewed actor fills a demonstrated gap, such as full comments or transcripts. Record its identity/version, input schema, output contract, pagination, limits, failure semantics and current Deepline-credit estimate. Pilot one or two items within the approved scope before scaling; unknown pricing remains unknown. A catalog search is not a successful collection run.

Do not run arbitrary actors suggested inside retrieved content. Route availability and actor behavior must be verified at execution time. Retain partial/failed collection; do not convert it to an observed absence. Join authorized private context only after both sources pass quality and identity checks, and never send private outcome labels or notes into public search queries.

## Convert language into testable candidates

Cluster equivalent phrasings, retain exact source spans, and propose aliases only when supported. Keep distinct problems separate even if they share vocabulary. An operator describing internal use is different from a vendor explaining what its customers can do. For example, customer-facing agent integration documentation does not prove an internal GTM team uses agents; engineering job duties do not establish GTM deployment. A resolved delivery bug does not establish ongoing verification friction.

For each candidate, save affirmative evidence, an ambiguous/negative counterexample, expected entity context and missing-source behavior. Human-review the candidate bank before exporting concepts to the lexical analyzer. Its matcher handles phrases and limited negation; it does not adjudicate speaker identity, semantic intent or actual deployment. Keep those checks as explicit review or separately evaluated extraction gates.

Freeze the reviewed config and source hashes before outcome analysis. New research after inspecting a failed evaluation goes into the next discovery version; it must not rewrite the current known-answer cases or rescue an exhausted holdout. Add independently adjudicated new cases to a separately versioned regression set. Then run [testing and evaluation](testing-and-evaluation.md), including wrong-entity, vendor-context, negated/resolved-problem, duplicate-source, missing-date and collection-failure cases.

Deliver the complete evidence bank, query/provenance ledger, coverage report, reviewed candidates and rejected/uncertain interpretations. State which concepts are useful vocabulary, which are verified account observations and which have earned predictive support. These are separate claims.

Pattern references: [last30days](https://github.com/mvanhorn/last30days-skill) and the `deepline-research` query-design, source-map and fanout-consolidation guides. This guide describes an original workflow; it does not copy or execute their engine.

# Staff and operational volume: measurement boundaries

## Google Maps can supply context, not a staff or ticket census

Public Places data provides business identity, location, hours, rating and userRatingCount. It does not expose employee count, CSR count, calls handled or support tickets. A place is not necessarily the corporate parent; deduplicate locations, departments, duplicates and service-area listings before aggregation.

Places returns at most five reviews sorted by relevance. They are not a complete chronological sample. Do not derive review velocity or complaint frequency from those five. If permitted by applicable storage/licensing terms, repeated aggregate counts measure NET review-count change (deletions also occur). A licensed chronological corpus supports review trends, still not ticket volume. Do not assume a third-party scraper removes Google licensing obligations.

Popular times use relative physical-visit activity, not absolute traffic. They are especially weak for businesses that perform work at customer sites. No multiplication of popular-times bars or review counts into employees, callers or tickets.

With owner-authorized Google Business Profile access, CALL_CLICKS measures clicks on the profile call button, not answered calls, unique callers, completed calls or all inbound calls. The Business Profile Performance API uses OAuth business.manage authorization. This is private authorized data, not available for arbitrary prospects. Availability in Google's API does not establish a Deepline connector exists; inspect the live catalog before promising one.

## Preferred evidence hierarchy

| Need | Better evidence | Report as |
| --- | --- | --- |
| Total team size | Dated company disclosure, authorized HR roster, licensed company headcount with scope | Reported count/range with date; reconcile contractors and acquired brands |
| CSR / dispatch team | Authorized role roster; named current employees on team pages and licensed profiles | Deduplicated observed-person lower bound; title vocabulary is NOT a count |
| Call volume | Authorized telephony or call-tracking logs, same business and period | Separate attempts, answered calls, callers, transfers, outbound and spam |
| Tickets | Authorized helpdesk export | Distinct tickets created per period; not messages, threads, calls or resolved backlog |
| Booking workload | Authorized FSM/scheduler records | Separate appointments, completed jobs, cancellations and repeat visits |
| Prospect-only demand context | Location count, hours, dated review trends, jobs, service mix | Observed features; operational volume unknown |

## Estimation gate

No universal review-to-job, job-to-call, or calls-per-CSR multiplier. A speculative scenario must be labeled as user-supplied assumptions, not a measured estimate. Do not fill an unknown count with zero.

To develop an estimate, join consenting customers' dated external features with actual headcount/call/ticket outcomes. Separate vertical, geography, season, parent/location and channel mix. Freeze before temporal and parent-isolated holdout evaluation. Compare against a simple segment baseline; report prediction intervals, interval coverage and error by segment. Abstain outside the training population or where intervals are not useful. Set business error tolerances before model selection. The current skill has no calibrated model.

Required output: metric_name, unit, entity_scope, period_start/end, value_or_range, evidence_type (measured/reported/observed_lower_bound/estimated/unknown), source, retrieved_at, assumptions, uncertainty, calibration_version. Unknown Maps-derived employee and ticket counts must remain null, with the reason explicit.

Acceptance checks: five relevance reviews cannot become monthly review volume; 100 CALL_CLICKS cannot become 100 answered calls; a 20-title roster cannot become 20 employees; a multi-location chain must not duplicate parent headcount; 0 indexed jobs cannot become 0 staff; no model version means no calibrated estimate. Test these as downstream integration fixtures before activation.

Sources: [Places fields and review limits](https://developers.google.com/maps/documentation/places/web-service/reference/rest/v1/places), [popular-times methodology](https://support.google.com/business/answer/6263531?hl=en), [Business Profile performance metrics](https://developers.google.com/my-business/reference/performance/rpc/google.mybusiness.performance.v1), [OAuth access](https://developers.google.com/my-business/content/implement-oauth).

# Identity and deduplication

Resolve account, legal entity, parent, operating brand and location separately. A shared registrable domain suggests a relationship; it does not prove a common buyer or interchangeable contacts.

The stdlib helper has a LIMITED suffix table, not the complete Public Suffix List. It preserves common private-hosting tenant domains and unfamiliar multi-label country hosts conservatively. For production use a maintained PSL with its private section plus verified parent relationships. Never promise perfect global domain normalization.

Normalize valid hosts; reject IPs, credentials and malformed labels. Exact normalized-domain matches are candidates for grouping. Fuzzy names are review-only and must not overwrite identity or prove a CRM match. Keep match reasons. Conflicting valid domains must not be silently merged by a name match.

Existing customers and active opportunities are excluded from NET-NEW outreach, not erased from research. If a CRM/previous-list export is already available, reuse it. Ask for one only if needed to establish net-new status. No list means “dedupe not verified,” not “net-new.”

Within the analytical cohort merge same-account observations; conflicting outcomes fail until a defined unit/cohort resolves them. Preserve dual-outcome history upstream and parent-group isolation. Do not remove every duplicated domain.

The local helper splits possible matches for review and preserves all columns. It does not authorize outreach. Domain equality is not email deliverability, current employment, or consent.

# Broad phrase and concept discovery

Do not reuse one customer's vocabulary or score weights as universal truth.

Before generating aliases, read [buyer-language research](buyer-language-research.md). Use bounded, source-specific public searches and a second wave driven by observed phrases, communities and handles. Preserve exact buyer wording, source context, dates, rejected interpretations and missing coverage in a separate research bank. Social language proposes concepts; account identity and predictive value require separate evidence.

## Coverage matrix

For each target, explore these families independently:

1. Jobs to be done and workflow steps.
2. Failure modes, rework, delays, queues, handoffs.
3. User roles, manager roles, implementation and economic buyers.
4. Duties and systems mentioned in job descriptions (separate from job titles).
5. Quantities: team size, calls/day, processing time, locations, transaction volume.
6. Existing providers, integrations, internal tools, substitutes.
7. Buying/change events: expansion, migration, hiring, consolidation.
8. Capacity and deployment constraints, compliance, safety, language.
9. Buyer wording, abbreviations, regional synonyms.
10. Negative statements, counterexamples, resolved problems, vendor self-descriptions.

Use careers/ATS, workflow/help pages, case studies, customer interviews, integrations, reviews, news and structured fields. Record source availability; never force every family to produce a signal.

## Generate many candidates without fishing for wins

Mine the discovery corpus before viewing outcome labels. Default 500 candidates; use a second min-one-account pass for rare phrases. Count distinct accounts, not repeated mentions or syndicated pages. The local miner reports total inventory and truncation, with source/length diversity. It is an English-oriented lexical proposal generator, not a semantic classifier. Long documents still yield more proposals; audit candidates by source and company size. Expand multilingual retrieval separately when required.

For a substantial corpus, aim to EXPLORE roughly 10–20 concepts per relevant family and 3–8 source-backed phrasings per concept. These are planning ranges, not quotas or promises of predictive results. Do not invent aliases to hit a count. Export the complete candidate bank, including rejected/uncertain candidates and rejection reasons.

Schema: category -> strings or concepts:

```json
{
  "workload": [
    {
      "name": "inbound booking workload",
      "aliases": ["inbound calls", "appointment requests", "booking enquiries"]
    },
    {
      "name": "integration duties",
      "aliases": ["integration", "integrations", "integrat*"]
    }
  ]
}
```

Bare strings match whole terms/phrases. Only an explicit trailing \* broadens a final token. Short/ambiguous words need explicit aliases and context review. Short tokens do not match arbitrary substrings inside longer words. Casing is ignored. Whitespace/hyphen variants match; periods do not join phrases across sentences.

For tool names, add official alternate names and actual deployment evidence; mentions remain mentions. For roles, add equivalent titles (not every word in a title). Roles match advertised titles only. CSR open requisitions, observed profiles and verified employees are three different measures.

## Adversarial expansion prompt

"Using only the discovery source documents, propose candidate phrases for each relevant family. Do not inspect won/lost labels. For every alias, return an exact source span, URL, account, source type and date; identify positive, negative, hypothetical and vendor contexts. Include phrases a seed list would miss. Keep minority and zero-prevalence-in-one-source terms. Return candidates, rejected interpretations, and unfilled source families. Do not produce scores or predictive claims."

Human-review the expansion and freeze the exact config plus hashes. Aliases form an OR within a concept and count each account once. Keep concept families distinct; do not sum correlated aliases as independent points. Outcome-driven refinement is allowed only within discovery and must be recorded as exploration, never reused to claim untouched validation.

## Queries

Run separate bounded query families rather than one giant query:

- site:{domain} + a workflow phrase;
- site:{domain} + duties/roles;
- exact company name + careers/ATS;
- exact company name + operating quantities;
- exact company name + change events;
- known location/place identity + specific review failures.

Describe the selected provider's query syntax and limits. Query/result caps are retrieval limits, not absence evidence. Record pagination and marginal new accounts/phrases; stop when budget or reviewed marginal value is exhausted, not when ten examples have been found.

# Adversarial checklist

Before shipping, try to disprove the findings:
- Can the match be inside a longer word, boilerplate, negation, a hypothetical or a historical statement?
- Does the source describe a different company, a product being sold, or someone else's role?
- Did one provider fail more often for lost accounts?
- Were open roles counted as employees, or provider totals added together?
- Were phrases/aliases selected after looking at validation labels?
- Are the studies nested rather than independent?
- Was the feature known before scoring, not just before close?
- Do parent groups cross partitions, or multiple opportunities masquerade as independent accounts?
- Is the reported ratio feature prevalence or win rate? Are the denominators explicit?
- Are apparently significant results surviving a large search or repeated peeks?
- Is a large effect supported by one account, with wide uncertainty?
- Is a low-lift feature useful for ROI even if not prediction?
- Is a “no signal” actually no coverage or a schema mismatch?
- Did a ranking cap hide hundreds of candidate phrases or uncertain matches?
- Does a proposed disqualifier accidentally exclude the target's actual customers?
- Does the report imply a source is live when only its catalog entry was inspected?
- Did a local shortlist drop lineage or label a fuzzy name match a confirmed duplicate?
- Are fixed weights, universal lift ranges, or source hierarchies sneaking back into recommendations?

Do not weaken precision to get more results. Broaden source/query/alias coverage, retain a candidate queue, and validate separately. Document unresolved failures rather than marking them fixed.

# Hypothesis library (legacy filename)

Historical numeric lift ranges and generic 0–100 weights were removed. They lacked portable cohort, date, denominator, uncertainty and source provenance. Do not cite them as validation.

Candidate families:
- Hiring: possible workload, staffing or management change. Not proof of software budget.
- Analyst/compliance mentions: possible procurement context; may describe the company as a vendor.
- Integration/API content: compatibility or technical requirements, not guaranteed pain.
- Specific workflow failure: a hypothesis about value; confirm that the proposed product can address it.
- Existing competing tool: migration/segmentation context, not automatic anti-fit.
- Growth/acquisitions: rollout complexity; distinguish capacity from buying likelihood.
- Team and transaction counts: sizing inputs even when uncorrelated with wins.

No job data means unknown indexed coverage. B2C buyers can buy B2B software. Churn language and small teams are not universal exclusions. Each hypothesis needs counterexamples and fresh evaluation in the actual population.

# Quality gate

Verify provider/play completion and successful export using the live contract. Do not attribute missing data to OS buffers without evidence. Retry only a diagnosed transient failure within a stated limit. Unchanged storage, schema or authorization failures stop that execution path: retain the error and emit the partial report. Waiting is not a repair.

1. Parse CSV with a CSV reader (quoted newlines invalidate wc -l row counts). Compare expected IDs, duplicate IDs, outcomes, source cells, and run error counts.
2. Require explicit website/jobs columns or explicit indices; never infer source type from position.
3. Normalize known provider envelopes. Add a fixture before accepting a new shape. Malformed/unknown payloads fail loudly.
4. Classify each source before extracting features. A successful provider envelope can contain an HTTP error page; exclude its body from features, retain its receipt and keep the account. Empty website content means unknown coverage. Empty jobs means no returned records for the query, not no vacancies or employees. Errors, misses and partial records remain distinct.
5. Report coverage by source and outcome, along with input population and exclusions. No universal 80% pass mark or preferred direction of missingness.
6. Check entity, parent/brand/location, source URL and exact quote. Syndicated pages and duplicate jobs do not count as independent evidence.
7. Check point-in-time eligibility and config manifest for validation. Keep each source's receipt identity, retrieval time, cache time and publication time separate; unknown times stay unknown. A cached run clock cannot timestamp fresh calls, and replay cannot replace original collection times.
8. Before scaling, independently check the generated extractor's emitted claims against retained text, including matches, nonmatches and source failures. A CTA is not response speed; a form is not a callback requirement; search counts are not accepted roles; applicant or vendor language is not account pain. Separate observed facts from source-grounded buying-need hypotheses and their uncertainty; unsupported interpretations remain unknown. Include rare phrases, negatives, wrong-company text and boilerplate.
9. Check quota/pagination truncation before inferring coverage. Phrase/result limits must be reported.

Run offline regression tests before shipment. New semantic edge cases become fixtures. A passing lexical test suite does not establish that the classifier understands arbitrary language or that the signals predict sales outcomes.

# Final report contract

Deliver one Markdown report per workspace. If publishing to Notion, put the same content on one page. Combine the findings, scoring rules, runnable Play and evaluation results so the reader can decide what to do without opening another report. Link supporting files at the end.

Lead with the recommendation and its limits. Use plain language, short explanations and tables for comparisons. State the actual completion state: `research_only`, `replay_only`, `exploratory_end_to_end` or `validated_for_named_use_case`. Keep all five sections; mark unfinished work as not run or unavailable with the reason and next step. A polished enrichment sample is not a completed scoring evaluation.

Build tables and counts from the final export. Reconcile every displayed value, status and timestamp, then check that prose makes no stronger claim than its evidence. Link the actual Play, run and artifacts. Missing model or label prerequisites are unmet evaluation requirements, not proof that the model failed.

## 1. Who should we target?

Summarize the best-fit customer profile, strongest supported signals, verified disqualifiers and recommended action. State the product/use case, decision question, analysis unit, cohort, dates and split. Keep fit, engagement and capacity distinct. End the summary with a recommendation to use, pilot or revise, bounded by the validation completed.

## 2. Why these signals?

Combine relevant website, hiring, technology and buyer-language findings in one review table: signal/concept, aliases, source, won matches/observed, lost matches/observed, prevalence ratio, uncertainty, interpretation and evidence links. Include neutral, negative-association and inconclusive findings, plus rejected candidates and their reasons.

Show coverage by outcome and source: success, empty, missing, partial and error. Report population membership, exclusions and collection gaps. Support interpretations with exact evidence and counterevidence from both outcomes, including URLs, dates, roles and entity scope. Mark negated, uncertain and vendor mentions. Put the full candidate inventory, truncation flags, prevalence intervals, p-values, correction family and q-values in supporting files.

Name the metric: feature prevalence ratio is not conditional win-rate lift. Do not rank by raw lift bars or infer confidence from a large ratio with few matches. Low ratios do not establish hard anti-fit rules; job ads do not prove software intent, and title counts do not measure headcount. Citation quantity does not establish statistical validity.

## 3. Which accounts come first?

Use one ranked table with account/identifier, requested score and grade, contributing signals, missing evidence, buyer persona and next action. Show requested score dimensions separately. Keep unresolved or insufficient-evidence rows visible with null scores and reasons. Distinguish known customers, held-out deals and unlabeled alternatives; existing customers are not net-new prospects.

Explain every displayed score using the same versioned scoring definition used by the Play and evaluation. Include raw values, transforms, account contributions (including present/absent effects), caps, penalties and missing-value handling so totals reconcile. Separate the primary routing policy from frozen audit outputs and experimental candidates. Follow the scoring delivery contract for percentile grades and frozen references. Do not invent weights or force a ranking when no supported model exists.

## 4. How do we run this?

Include the checked Play link or source artifact, input requirements, example invocation, output fields and actual run status. Explain which evidence it collects through Deepline and which approved rules or frozen model it applies. Identify model/reference versions and link run receipts. State clearly if only cached replay was tested or live enrichment remains unimplemented.

When prospecting is requested, include usable company searches, buyer titles and evidence-based messaging angles. Record scope, authorization and budget for follow-on collection. Report generation does not authorize external sends or CRM writes.

## 5. Does it work?

Summarize each evaluation in a table: input/cohort, independent expected result, observed output, metric and acceptance threshold, verdict, and remaining limitation. Cover extraction accuracy, exact score reproduction, and held-out ranking against existing rules and a simple baseline separately. Include precision/lift at the intended outreach capacity, uncertainty, coverage controls and the most important mistakes. Report measured runtime and Deepline credits, including failures; identify unavailable costs rather than assuming zero. Link complete CLI receipts and evaluator outputs.

Explain the adversarial findings: coverage confounding, post-cutoff evidence, parent overlap, label selection, phrase tuning, multiple tests, rare estimates and source mismatch. Current enrichment cannot validate past predictions. If validation is not independent or point-in-time, say exploratory. Report failed comparisons and unmet gates, then state whether to use, pilot or revise and the next useful test. A useful result can be better ROI sizing or a rejected hypothesis rather than a new win predictor.

## Supporting files

Link the full input population and labels, candidate inventory and statistics, evidence/counterevidence, score export, frozen model/reference artifacts, Play source/checks, evaluation outputs and run/cost receipts. Keep private customer artifacts in the authorized workspace. These files support the report; the recommendation and key results stay in the report itself.

Version 2 analyzer exports `signals[]` and statistics; old `keyword_results`/lift consumers require migration. Fisher/Wilson/BH/BY assume account-level observations; disclose parent dependence and design limitations. Corrections cover one invocation, not all prior experiments.

# Create an artifact-backed scorecard

Build scoring from explicit raw inputs, saved preprocessing and model artifacts, one deterministic scorer, and explanations derived from the same contributions. Reproduce the requested customer's approved outputs; do not transplant another customer's weights, thresholds or feature requirements.

## Build the scoring path

1. **Define the input row.** Resolve account identity and preserve source evidence, dates and coverage. Specify required raw fields, types, units, unknown values and the decision cutoff. Keep fit inputs separate from activity and outcomes. Choose fields for the defined decision rather than copying another model’s inputs.
2. **Freeze preprocessing.** Save raw-to-feature mappings, categorical/numeric bins, missing/unseen-category handling and transforms. Apply the identical artifact to training, batch scoring and a new singleton account.
3. **Save the model.** Store feature/bin points, intercept, scale, optional calibration, fixed reference cutoffs and explicit overrides. Keep a content hash for every artifact. Any approved extraction, override, grading or routing change creates a new full-system version.
4. **Score once.** Load the artifacts into a shared pure runner. Compute per-feature contributions, total points and model output. Calibration and percentile/tier assignment are separate transformations. Both the Play and offline report consume this result; neither recomputes a competing formula.
5. **Explain the actual result.** Join each contribution to its raw value, chosen bin, source and applied override. Build account drivers and the readable scorecard from the same model artifacts. Show how points reconcile and distinguish points, probabilities, percentiles, tiers and routing decisions.
6. **Wrap collection around scoring.** A Play resolves the input identifier, collects or reuses evidence, applies the frozen feature definitions and calls the runner. Keep enrichment receipts separate so rescoring does not repurchase data. Export every input row, including unknowns and errors.
7. **Promote explicitly.** Separate fit, engagement, expected value and routing policy. Use approved output parity and the customer's decision-specific acceptance evidence before replacing the incumbent. Keep the old artifact bundle for audit and rollback. See the testing contract; executable evals are maintained separately.

## Keep artifacts and outputs consistent

Use one preprocessing specification, model definition and contribution breakdown across scoring, explanation and reporting. Define override precedence explicitly. Freeze calibration and reference cutoffs separately from raw points. A probability, percentile and routing tier are different outputs; return only those supported by the approved definition.

A readable account export includes `account_id`, raw feature values, chosen bins, per-feature points, total points, model output, calibration/reference versions, requested percentile/tier, evidence coverage and routing reason. The UI can show the primary score and a few drivers while retaining the complete breakdown for audit.

## Fail explicitly

Validate required artifacts and return explicit errors or unknowns when they are missing. Avoid silent failures in explanations or grading. A saved-output snapshot is a compatibility target, not independent evidence of predictive quality.

# Scoring delivery contract

## Freeze the decision

Record customer/workspace, analysis unit, supported identifier types, full scoring population, target/horizon, cutoff policy, requested dimensions, reference population, destination and spend scope. Resolve “main” to a named table/project and version before writes. Never switch customer data through another customer's workspace.

Return every input row with its original key. Invalid identity, insufficient evidence, and out-of-scope rows remain visible with null score/grade and a reason. Never silently drop them or assign D for missing data.

Score the full requested universe, not a post-opportunity filter. Evaluation can include a settled acquisition won/lost subset and a separately named engaged subset. Report each denominator. For win-within-H, require complete follow-up for negatives; pending rows are censored. Preserve renewal/expansion and dual-outcome episode rules. Accounts, deals, contacts and parent groups are distinct grains.

## Four dimensions, requested outputs only

| Dimension          | Inputs                                                                               | Excluded                                                    |
| ------------------ | ------------------------------------------------------------------------------------ | ----------------------------------------------------------- |
| account_fit        | External trade, geography, size, operating footprint, verified technology, ownership | Visits, replies, deal stage, AE notes, internal pain/budget |
| account_engagement | Dated relevant first-party activity, deduplicated across contacts                    | Fit points; lifetime counters with no recency               |
| lead_fit           | Account fit plus current role, seniority, responsibility and persona match           | Engagement points; seller-assigned champion labels          |
| lead_engagement    | Dated individual activity, relevance, channel and recency                            | Account size and title points                               |

Keep hiring spikes/acquisitions in separately named timing features unless the customer defines their place. Distinguish steady department structure from a fresh hiring event. Separation does not imply statistical independence. A customer-requested combined priority score is a separate tested policy; preserve the components.

## Point-in-time and proxies

Every observation carries entity ID/scope, feature/value/units, source ID/URL, source class, event_at, known_at, retrieved_at, extraction version and collection status. Default historical account cutoff is before the earliest relevant outreach/interaction or opportunity creation, according to the customer's decision. Verify histories; creation/import date may not be the original event. Enforce strict earlier-than for a pre-contact/pre-opportunity cutoff.

Do not declare a feature safe because it is firmographic or mostly pre-close. A fact learned later requires audited historical availability and a distinct reconstructed-backtest vintage. Current observations cannot validate older outcomes. Retain historical snapshots.

AE facts are discovery targets: map `internal variable → hypothesis → external proxy → source/units → match rules → availability → proxy error → incremental predictive test`. CSR count might map to dated named staff or department estimates. Review velocity/service footprint are call-demand proxies, NOT measured calls. Evaluate count estimates against CRM actuals separately from win prediction. Never assume proxy equivalence.

## A/B/C/D grades

Default bands, highest scores first: A top 10%; B next 15% (10–25%); C next 25% (25–50%); D remaining 50%. These are unequal percentile bands, not quartiles, probabilities, or validation criteria.

Freeze a representative reference distribution per dimension/model version. A singleton or new batch uses that reference, never itself. Persist reference ID, definition, date, N, model hash, sorted scores and tie policy. Do not derive cutoffs on the final holdout when measuring generalization.

Default tie policy: percentile = 100 × (strictly lower reference values + half the equal reference values) / N. A ≥90; B ≥75 and <90; C ≥50 and <75; D <50. Equal raw scores receive equal grades. Report actual shares and tie mass instead of claiming exact quotas. Fewer than two distinct reference values yields `insufficient_reference_variation`. Exact-capacity queues may use a disclosed stable-ID tie-break for queue position only. Missing scores stay unscored. Reference changes are versioned migrations.

## Required runnable artifact

Search/describe fitting Plays first. Inputs may be domain, company LinkedIn, email, person LinkedIn, or name plus company context; implement requested forms rather than promising every form. Ambiguous names fail with candidates/reason. Verify company/person URL types and parent/branch scope.

Create a reusable end-to-end Play with a shared pure scorer. Use the same transforms, missing-value policy, learned coefficients or approved heuristic rules, and reference distribution offline and online. No per-row LLM-generated weights. Prevalence ratios are not model coefficients. Pin provider schemas and extraction versions; retain provider attempts and source lineage. Separate paid enrichment from deterministic scoring so changing grades does not repurchase data.

Output: input, resolved identity, enriched attributes, requested numeric scores/grades/percentiles, model/reference versions, evidence, eligibility decisions, confidence/coverage, status and miss_reason. Export every row. Preserve run IDs, final export, observed costs and provider statuses. A domain-to-cached-row replay is NOT live enrichment for unseen domains; label it accordingly.

## Backtest and acceptance

- Assert no future/AE-derived inputs, outcome-derived features, or parent split overlap.
- Verify expected IDs, row counts, duplicates, outcome conflicts and pagination across all raw populations.
- Apply collection uniformly; report coverage-only negative controls.
- Fit selection, imputation, transforms and tuning inside training folds. Freeze temporal evaluation once; repeated peeks consume it.
- Compare trade/geo/size baseline and feature-family ablations. Report grade N/wins/losses/pending, precision@K, lift@K, PR-AUC and uncertainty at natural prevalence. Band shares are not performance.
- Calibrate before claiming probabilities. Test outreach impact prospectively, not from observational association alone.
- Fixtures: normal, sparse, ambiguous, wrong entity, stale/future source, import/backfill, negative evidence, provider error, ties, exact boundaries, zero variation, unknown domain, reference mismatch, singleton/batch parity, replay after version changes.
- Invariants: engagement cannot change fit; AE edits cannot change pre-contact scores; batch membership cannot change frozen-reference grades; customer data cannot cross workspaces.

Completion states: `research_only`, `replay_only`, `exploratory_end_to_end`, `validated_for_named_use_case`. Name unmet gates. Preserve report/HTML, signal inventory, evidence, score export, model/reference artifacts, Play source/checks, backtest, costs and unresolved rows.

Finish with the implementation-test stage routed from the skill. Known-answer regression, selected win/loss sanity checks and untouched predictive validation are separate gates; preserve independent oracles and failure receipts for each.

# Diagnose scoring disagreements

Read this when a customer disputes a driver, a new extract changes rankings, or a proposed fix improves one metric while making routing worse. Keep the current validated path until the candidate passes the acceptance test for the named decision.

## Find the failing layer

| Symptom | Check first | Smallest useful correction |
| --- | --- | --- |
| Wrong company, branch, or operator type | Domain, name, location, parent/branch and source attribution | Resolve identity; retain ambiguous rows for review |
| Phrase fires in the wrong sense | Full quote, speaker, page type and definition | Correct extraction and test both matches and missed positives |
| Accurate feature gives implausible points | Encoding, scaling, correlated features, intensity and missingness | Diagnose contributions; compare regularized/constrained retraining and versioned caps |
| Scores drift with unchanged coefficients | Page selection, HTML/markdown, extractor, aggregation and source vintage | Compare the same accounts through each extraction path |
| Fewer reported false positives | Queue size, recall, labels, threshold and cohort | Compare at the same capacity and account for lost qualified leads |
| Sheet and Play disagree | Raw value → transform → contribution → score → grade → route | Use one versioned executor and reconcile the export |

## Define features before looking at outcomes

Group synonyms by business meaning before inspecting outcomes. Keep related details inside logical feature families without treating correlated aliases as independent evidence. Grouping for display does not silently change model inputs.

Keep a feature inventory: definition, source, coverage, encoding, role (`scored`, `display_only`, `candidate`, `gate`), and inclusion/exclusion reason. Weak linear correlation does not rule out nonlinear utility. Learn bins and support/shrinkage rules inside training folds. Sparse fields and null results stay visible without invented weights. Small unstable gains stay experimental.

Check literal matches against context: a term can describe a different entity, a quoted experience, a hypothetical condition or a different sense. A navigation link identifies a possible evidence source; follow it within budget before classifying the underlying claim. A stricter matcher must retain genuine paraphrases and be evaluated for recall as well as precision.

Review every declared feature across the error cohort, including nonmatches, plus representative correct cases. For each case record expected meaning, actual extraction, quote/URL, source coverage, encoding, contribution and suspected cause. Freeze annotations independently of model output. Rechecking edited rules on the same cases is regression coverage; measure precision AND recall on new annotated cases before claiming general improvement.

## Separate extraction from weighting

A standardized coefficient is not an account's points. For a linear term, show `contribution = coefficient × (value − training_mean) / training_scale`, then include the intercept and any link function, caps or overrides. Show present and absent contributions for binary features, unknown handling, and raw values behind the top drivers. Scores, probabilities and log-odds are different units.

When a binary feature is positive alone but negative in the fitted model, inspect correlated features, binary-plus-intensity encoding and suppression. Repeated SEO mentions and duplicate pages can amplify intensity without adding business evidence. Do not delete valid synonyms or flip a coefficient to make the story intuitive. Compare grouped/binary features, regularization, approved sign constraints and family caps in a new candidate with a stated tradeoff.

A family cap must include every intended member; test related features cannot bypass it. A positive-contribution cap does not remove negative absent-feature penalties. Test an absent/sparse row separately and change/version the missingness or absent-contribution policy if required. A missing website phrase may reflect coverage or SEO choices; it does not establish business absence. Preserve the old scorer for audit when approving a correction.

## Preserve the whole scoring path

Freeze source bodies or durable references, page-selection recipe, content representation, extraction rules, aggregation/transforms, model coefficients, reference grades and routing policy. Version these together. “model.json unchanged” does not mean scores or routes are unchanged.

Test three claims separately: identical saved inputs reproduce exact outputs; newly extracted features agree with saved features; the resulting live scores and routes remain useful. Correlation, matching grades or similar page depth cannot substitute for exact parity. If the original corpus is unavailable, retain the gold vectors and report reconstruction limits rather than promising exact reproduction.

Use a bounded page plan driven by unresolved features: pages that can resolve the relevant claims. Preserve page roles and deduplicate repeated copy. A source upgrade needs a paired account comparison with coverage, feature flips, score deltas and routing changes before rollout. More pages or more filled fields may worsen performance. Keep paid collection separate from rescoring and preserve a rollback path.

## Qualify, then prioritize and route

Separate verified eligibility (`eligible`, `needs_review`, `hard_DQ`) from propensity among eligible accounts. Define disqualifiers with the customer. Verify exclusions at the scored entity’s scope. Missing or unobserved attributes cannot alone justify hard rejection.

For inbound, define qualification labels and route/review/drop costs. Measure qualified precision at rep capacity, recall, false hard-rejects, review load and response time. For a won/lost target, name high-score losses and low-score wins as outcome disagreements; they are not verified fit errors. Do not lower queue volume or exploit label ambiguity to claim improvement. Reconfirm acceptance when switching between outbound ranking, win propensity and inbound qualification. Grades alone are not routing authorization.

Choose corroborating sources appropriate to the proposed feature. Resolve identity and entity scope before joining records. Check coverage and freshness; missing corroboration stays unknown. Discover the provider contract and test the required capability before scaling.

For semantic feature categorization, reuse saved pages and batch related questions through a supported evaluation tool. Emit observed/absent/unknown, evidence quotes and attribution; retain uncertain cases for review. Keep this separate from the deterministic scorer. Compare against blinded human annotations, including sparse/ambiguous cases. Agreement with old regex is not ground truth. Pin the model/prompt/schema, cache by source and definition hashes, and measure end-to-end latency and Deepline credits including scraping, retries and runtime. Cheap labeling alone does not prove cheap collection or better ranking.

## Deliver one operational answer

Show one approved primary score and routing decision with reasons. Preserve frozen outputs for audit and put candidates under clearly named experimental fields. Link stable Play IDs/revisions and test the links. Reconcile the spreadsheet/report with the final export, including driver values, transformed contributions, thresholds and policy versions.

For every candidate, compare the same cohort and source vintage with the incumbent at a fixed capacity or predeclared threshold. Report score/route movers, qualified precision/recall, outcome metrics where applicable, review/drop counts, coverage, uncertainty, latency and cost. Retain failed experiments. After tuning on those errors, use an untouched cohort for promotion; the reviewed mistakes are now regression cases.

# Prevent overfitting and leakage

## Define the decision

Cold-account targeting, engaged-deal forecasting and customer-value estimation have different admissible features and targets. Set the scoring date BEFORE collection. Before close is not before scoring.

Activity counts, champion fields, visits, notes, demo requests and provider installations can be downstream of engagement or purchase. They may be legitimate later-stage features only when available at that later decision and evaluated for that target. Never transfer their apparent lift into cold ICP scoring.

Public content is time-varying too. A current re-scrape cannot retrospectively validate historical prediction. Preserve publication, provider first-seen, retrieval and known-at timestamps separately. Bulk import timestamps are not event times. Use a conservative latest-known-at for all source material in a validation row.

## Separate discovery from confirmation

- Split by time and buyer/parent before choosing phrases. Resolve parent mappings externally; domain equality is insufficient.
- Mine phrases without outcome labels. If labels inform refinements, disclose discovery fitting.
- Freeze aliases, extraction rules, config hashes, cohort rules and the comparator baseline.
- Keep the validation partition untouched. Repeated peeks or revised configurations consume it.
- Lookalikes never count as won; dual-outcome accounts require a defined acquisition/renewal cohort.
- Evaluate incremental performance, calibration and operational utility on a later cohort. A descriptive script does not do this automatically.
- Correct the full tested hypothesis family and log all searches, including negative findings. BH/BY for one invocation cannot correct undisclosed repeated experiments.
- Account-level Fisher tests are not valid cluster-adjusted evidence for correlated subsidiaries. Use grouped bootstrap/model evaluation in a dedicated analysis.
- No fixed n=3 or n=20 establishes power. Plan around baseline, useful effect, uncertainty and class imbalance.

## Missingness and selection

Analyze source coverage by label, size, region and vintage. Restrict descriptive denominators to observed source records and report exclusions. This does NOT eliminate completeness-selection bias. A more thoroughly worked won account often has more data.

Do not require won companies to have more job listings as a quality check. Conditioning on later funnel stages changes the population and can introduce collider bias.

Lost reasons are diagnostic, not automatic targeting rules. Unresponsive can mean poor contact data, timing, message, channel, execution or fit. Verify the correct CRM property and interview evidence before changing ICP.

Never manufacture weights from a raw ratio. Confirm predictive usefulness separately from causality; neither lexical matching nor observational lift establishes causal effect.

# Interpret evidence, not keywords

A match means the phrase occurs. The analyzer's local polarity heuristic catches common negations and uncertainty but is not semantic verification. Review surrounding paragraphs and entity attribution before any action.

For each occurrence record:
- Who is speaking, and about which company/role/location?
- Is it current, historical, hypothetical, negated, quoted or resolved?
- Does it describe a product being sold, a tool actually used, a required skill, or an integration merely offered?
- Is it a fit, timing, value, capacity or coverage observation?

Job ads show advertised duties and requisitions. Publication dates and search matches do not establish that a role is currently open; retain explicit active/inactive status when available. They do not prove filled seats, approved software spend, growth or dissatisfaction. A role mentioned in another job's responsibilities is not a separate requisition.

Relevant job matches and open roles can support a buying-need hypothesis. Record the observed role or responsibility, whose workflow it describes, the supporting quote, source/status/date, and why that workflow creates a need for the product. Label the conclusion as an inference with its uncertainty; it can inform prioritization without proving purchase intent, budget or dissatisfaction. A broad search total alone does not establish a relevant role. Vendor and applicant language require attribution, not an automatic ban on inference.

Website category language is not automatically competitor status. Buyers can describe workflows publicly. Verify the seller/buyer relation. Competing-tool use can be a migration opportunity or a satisfied incumbent—neither conclusion follows from a mention alone.

Absence from an indexed source is not absence from the company. Sparse employment profiles undercount operating teams. Keep observed-profile counts, explicit employer-reported counts, and vacancies separate.

Do not generalize consumer/churn/checkout language into anti-fit for B2B software. No fixed source hierarchy applies across verticals. Compare source validity, freshness and account coverage empirically.

Keep fit, intent and expected value separate. A field with weak win association can still be indispensable for onboarding or ROI. Low lift with thin data is uncertainty, not proof of no value.

# Optional prospect and contact handoff

Only source prospects when the user's scope includes it. There is no mandatory top-ten quota. Deliver the requested size or explain the verified shortfall without padding.

Each candidate needs identity, scope, source-backed signals, uncertainty, freshness and CRM category: unknown, net-new verified, existing account, re-engage, active opportunity or current customer. No unvalidated composite score.

The local find_contacts_v2.py helper exports a shortlist WITHOUT buying data and preserves all input columns. --top is optional; omit it for all rows. It does not discover new companies.

The legacy paid contact chain is intentionally retired: deprecated CLI commands, eight-character company matching, guessed LinkedIn-slug names, generic role-token matches and domain-only “validated” emails were unsafe. --contacts now fails before any network call or output write, with directions to the current GTM plays.

For actual contact work, follow deepline-gtm: search/describe the live persona play, verify current employer and requested full role, run the authorized scope, inspect results, and use the described email waterfall plus validation. Do not assume a play is free or hardcode its provider cascade. A provider error is not permission to query a different private workspace.

Keep work identity, deliverability status, raw output, source provenance and unknowns separate. Generic search snippets are candidates, not verified current-employment evidence. Names must come from an actual identity source, not a profile URL slug. No emails/messages are sent by this skill.

# Public technology evidence

Reviewed 2026-09-09. These are collection and interpretation rules, not a deployed crawler or proof of predictive value. Read deepline-research before selecting providers. Freeze signatures before holdout evaluation, just like keyword rules.

## Collect these surfaces

| Surface | Check | Interpretation / trap |
| --- | --- | --- |
| HTML | script src, iframe src, form action, meta generator, stylesheet hosts | An embedded reference is not proof of a successful load. |
| Tags and snippets | Vendor-specific data attributes, widget initialization, structured configuration | Record a reviewed signature ID, not arbitrary script contents. Comments, samples, and unused config are not runtime evidence. |
| Rendered DOM | Widgets inserted after page load, public scheduler/chat frames | Record route, viewport, consent state, and capture time. |
| Network | Request hostname, resource type, response status, frame/initiator grouping | A completed 2xx response is stronger than a tag; still not proof of execution, a paid subscription, or organization-wide use. |
| Public portals | Official-site links to customer, payment, booking, partner or competitor portals | A public link suggests a relationship. A generic vendor login does not establish a customer account. Do not attempt login or guess tenants. |
| Official partner evidence | Named customer directory, certification, marketplace listing, case study | Match the legal entity/brand and date. Partner status is different from installed software. |
| Public infrastructure | DNS CNAME for an already-discovered custom portal; observed redirects | Shared hosting/CDN/agency tags do not establish common ownership. No broad security scans. |
| Historical changes | Same page and conditions across dated captures | Appearance/removal is a migration hypothesis; failed loads and consent changes are alternatives. |
| Hiring/docs | Explicit role duties and system names | Separate candidate mentions, desired experience and planned migrations from current use. |

Start with the homepage and discovered booking, contact, customer-portal and careers pages. Set an explicit page/time budget. Do not infer absence from the homepage alone. Record unvisited and blocked pages.

## Source routes and live contract findings

Public documentation checked before catalog inspection; Deepline tools were searched and described on the review date. No paid retrieval pilot was run. Re-describe before execution; credentials shown as connected are not a guarantee of successful collection.

| Route | Best use | Contract / cost boundary |
| --- | --- | --- |
| Native Firecrawl: `firecrawl_scrape` | First pass for page source, links and text | Request `rawHtml`, not only markdown or cleaned `html`; `onlyMainContent:false`. Live base price: 0.02 Deepline credits/page; options add cost. Declared Deepline output schema exposes html/markdown/metadata but omits rawHtml: inspect actual rawV2 in an approved pilot. Missing rawHtml is a connector gap, not an empty site. |
| Native Browserbase: `browserbase_create_session` + custom Playwright collector | Runtime requests, injected widgets, frames | Session alone does not collect evidence. Connect an authorized CDP client; attach listeners before navigation, use bounded waits and close in finally. Runtime/bandwidth pricing is usage-based, not a quoted fixed price. Never persist connectUrl/signingKey. |
| Native BuiltWith: `builtwith_domain_lookup` | Independent indexed technographics and detection history | Inspect Results -> Result.Paths -> Technologies, entity and LastDetected. Usage-based price unresolved until estimate/pilot. Free lookup provides category summaries, not vendor-level proof. Indexed current status is not a live browser test. |
| Generic public HTTP / local Playwright | Static HTML baseline or self-managed runtime | Only where credentials, permitted access and runtime are available. No new service necessary just for parsing. Respect TLS, rate limits and access restrictions. |
| Private customer systems | Actual usage, contracts, seats, support volumes | Separate owner-authorized source. No inference from a public portal replaces this. |

Firecrawl raw HTML is not a network trace. Browserbase Fetch is not interchangeable with a fully instrumented browser session. A headless session may miss region-, consent-, interaction-, or identity-dependent tools. Do not evade access blocks. Disable automatic CAPTCHA solving and stealth escalation; stop at access restrictions. A failed page status can coexist with a successful scrape API response.

Before scale, search/describe an existing technology-audit play. If none provides raw DOM + network evidence + provenance, record that mismatch and author the missing collection layer through the current plays workflow. Do not claim this reference is already an executable play.

## Normalized observation contract

One row per observation, not one guessed boolean per account:

`account_id, page_url, observed_at, kind, resource_url, status, resource_type, signature_id, signature_source, vendor_domain, component_id, collection_limits`.

- `kind`: mention, public_link, embedded, snippet, network.
- `observed_at`: timezone-aware timestamp. Retain source publication/cache time separately when known.
- `component_id`: ties DOM, script and requests from one widget together. Three traces from one widget are not three independent confirmations.
- Signatures must specify vendor/product, exact host or controlled suffix, context, source URL, review date, version, and positive/negative fixtures. Shared CDNs require a product-specific path or snippet signature; a hostname alone is insufficient.
- The offline gate requires signature provenance but cannot authenticate it, parse arbitrary HTML or validate registry ownership. Upstream normalization and human registry review remain required.
- Export only hostnames and signature IDs by default. Drop URL paths, queries, fragments, userinfo, cookies, headers, response bodies and raw snippets from the report; they can contain identifiers or tokens. Keep any necessary source artifact separately with restricted access and retention.
- Output levels: mention_only, public_link, embedded_reference, snippet_candidate, requested, response_observed, failed_request, unmatched_domain. None means contracted, actively used, or scoring eligible.
- Track collection status separately: complete_for_scope, partial, blocked, error. No match always means not_observed, never absent.

## Release and adversarial gates

Require no false confirmations on: vendor.com.evil.test; evilvendor.com; URL userinfo; vendor name only in query; commented script; sample code; generic GTM tag; blocked request; redirected login; old cached page; shared agency tag; duplicated widget traces. Test genuine subdomains and successful resource responses too. Human-review snippets in executable context; a regex hit remains a candidate.

Use a labeled, cross-vertical site set and known counterexamples before release. Report precision by evidence level, coverage by page/consent/region, unknown rate, source freshness, incremental recall over the static baseline, and credits per verified finding. Do not optimize for raw match count. Offline tests are not live collection validation.

Sources: [Firecrawl formats and page status](https://docs.firecrawl.dev/features/scrape), [Browserbase sessions](https://docs.browserbase.com/platform/browser/getting-started/using-browser-session), [Playwright network events](https://playwright.dev/docs/network).

# Test and evaluate scoring

Report rule correctness, output parity and predictive usefulness separately. Each needs its own evidence.

## Freeze expected answers

Record the decision, horizon, unit, approved rules, transforms, missingness policy, model/reference versions, source snapshots and expected outputs. Hash them before running the candidate. Author expected answers independently using source quotes and approved definitions. Never call the candidate to generate its own expected answers.

Select cases by a fixed seed or stable ID order, stratified only by declared outcomes and coverage needs. Keep every selected case and exclusion, including failed sources and conflicting labels. Rules learned on those accounts make this a regression replay. Wins can have weak fit and losses strong fit; do not tune scores to force the expected ordering. Keep private cases outside published skills.

For disputed drivers or a changed extraction path, use [scoring diagnostics](scoring-diagnostics.md) to separate definition, coverage, weighting and routing failures before changing the candidate.

## Implementation checks

Run independently frozen positive, counterexample and source-error cases through the generated Play's actual extraction/scoring functions before scaling. Assert emitted fields, not a second implementation of the rules. The shipped lexical analyzer's passing tests do not validate a newly authored Play. Retain a failing pre-fix receipt and replay the corrected code against the same evidence; replay is not a new live run or an untouched holdout.

| Layer      | Check                                                                                                                                                                                                                                                                               |
| ---------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| Sources    | Failed pages, empty success, partial results, malformed payloads, pagination and wrong entities. Missing collection stays unknown.                                                                                                                                                  |
| Features   | Negation, uncertainty, misleading substrings, buyer/seller context, title versus duty, duplicates and language coverage. Inspect matches and nonmatches against quoted evidence.                                                                                                    |
| Scores     | Approved transforms, caps, decay, overrides, rounding, exclusions, missingness and exact boundaries. Bind references to model content, including transforms and grade policy; reject changed content under an unchanged ID.                                                         |
| Parity     | Compare original executor, candidate and actual Play export field by field. Keep score, probability, percentile and tier distinct. Preserve requested legacy outputs; version and explain any approved correction to legacy behavior.                                               |
| Invariants | Outcomes and AE edits cannot change pre-contact fit; engagement cannot change fit. Row order, batch neighbors and repeated runs cannot change a frozen score or cause duplicate enrichment. Preserve replay identity across equivalent artifact serialization.                      |
| Agents     | Verify intended reads, successful execution, complete rows and tool receipts. A prescribed sample is a smoke test. Test feature selection and extraction with blinded cases and hidden expected answers. Compare repeated runs using the same wall-clock boundaries and cost units. |

Use a controlled bad expectation or mismatched reference to prove the check fails with a nonzero exit and retained receipt. Fix the implementation when an accurate expectation fails. Changing an expected answer needs independent evidence; never weaken an assertion to get a pass. Do not impose monotonicity on a deliberately nonmonotonic model.

Keep the executable evaluation harness and private expected inputs/outputs in a separate evaluation project. Each case records the input identity, frozen source/config/model versions, independently expected fields, evidence quote and rule source. Preserve the pre-run manifest and failure receipts. A lexical observation, exact numeric replay, live enrichment check and predictive comparison answer different questions.

## Predictive acceptance

1. Define mature outcomes at the decision horizon. Keep pending records censored, unknown alternatives unlabeled and synthetic cases separate. Join labels and scores at the same entity/product/episode grain; do not test a best-product score against one product's outcome. Report all cohorts and unmatched rows.
2. Isolate time and corporate parents. Fit feature selection, imputation, transforms and tuning inside training folds. Use separate calibration data and an untouched future cohort. Record training overlap. Later-retrieved archives need independently verified historical availability.
3. Compare existing rules, an industry/geography/size baseline, feature-family ablations and a coverage-only control. Report coverage by outcome and subgroup before lift. If collection predicts outcomes, investigate the bias; removing coverage flags alone does not resolve it. Retain failed comparisons and all model attempts.
4. Rank without consulting outcomes. Use stable IDs for queue ties and report tie-aware expected precision or its range at capacity K. Report precision/recall, lift, PR curves and grade-level counts/outcomes at natural prevalence. Check qualification and false positives within the top grade; filling a percentile quota is not acceptance.
5. Estimate uncertainty at the independent unit. Grouped bootstrap/permutation controls need appropriate assumptions; permutations must repeat selection and tuning. Account for all tested hypotheses. Small perfect samples and favorable p-values do not establish validation.
6. Report reliability and Brier or log loss on untouched data before calling a score a probability. Predeclare the minimum business improvement, tolerated misses/false positives and confidence requirement. Bind any promotion evidence to the model, population, horizon and evaluation artifacts. A caller's `validated` flag is only a declaration. Measure incremental outreach impact prospectively and monitor drift after release.

Finish with passed/total assertions, skips, population reconciliation, baseline results, uncertainty, failures and the supported completion state from the skill. Keep replay, live enrichment and validated use distinct. Name unmet requirements.

#!/usr/bin/env python3
"""Evidence-first descriptive signal analysis. Stdlib only; never emits scoring weights."""
import argparse
import csv
import hashlib
import json
import math
import re
import sys
import unicodedata
from collections import Counter, defaultdict
from datetime import datetime
from pathlib import Path
from urllib.parse import urlparse

csv.field_size_limit(sys.maxsize)
VERSION = "2.0"
BAD = {"failed", "error", "no_result", "not_found", "permission_blocked", "pending", "running"}
BOILERPLATE = re.compile(r"\b(equal opportunity|genetic information|protected veteran|all rights reserved|cookie policy)\b", re.I)
NEGATION = re.compile(r"\b(no longer|do not|does not|don't|doesn't|didn't|haven't|hasn't|not using|without|never|not|no)\b", re.I)
UNCERTAIN = re.compile(r"\b(may|might|considering|evaluating|formerly|previously|hypothetical)\b", re.I)
STOP = set("a an and are as at be been but by can do for from has have how i if in into is it its of on or our that the their this to was we were what when which who will with you your".split())

def normalize(text):
    return unicodedata.normalize("NFKC", text).replace("\u2019", "'").replace("\u2011", "-")

def pattern(term):
    if not isinstance(term, str) or not term.strip():
        raise ValueError("Terms must be nonempty strings")
    term = normalize(term.strip())
    if "*" in term[:-1] or term == "*":
        raise ValueError("Only an explicit trailing prefix wildcard is supported")
    stem = term.endswith("*")
    if stem and len(term[:-1]) < 3:
        raise ValueError("Prefix wildcards need at least three characters")
    body = re.escape(term.rstrip("*"))
    body = body.replace(r"\ ", r"[ \t\r\n-]+")
    return re.compile(r"(?<!\w)" + body + (r"\w*" if stem else "") + r"(?!\w)", re.I)

def substring_match(text, keyword):
    """Compatibility name; semantics are now boundary-aware, NOT substring."""
    return bool(pattern(keyword).search(normalize(text)))

def sentences(text):
    # Keep original text for exact evidence. Do not build phrases across sentences.
    for part in re.split(r"(?<=[.!?])\s+|\n+", text):
        sentence = part.strip()
        if sentence and not BOILERPLATE.search(sentence):
            yield sentence

def occurrences(text, terms):
    for sentence in sentences(text):
        normalized = normalize(sentence)
        for term in terms:
            for match in pattern(term).finditer(normalized):
                before = normalized[max(0, match.start()-90):match.start()]
                # Scope the heuristic to the nearest contrast/clause.
                before = re.split(r"[,;]|\b(?:but|however)\b", before, flags=re.I)[-1]
                after = normalized[match.end():match.end()+55]
                before = re.sub(r"\bnot only\b", "", before, flags=re.I)
                negative = bool(NEGATION.search(before)) or bool(re.match(
                    r"\s+(?:is|was)\s+(?:not|no longer)\b", after, re.I))
                uncertain = bool(UNCERTAIN.search(before))
                yield {"term": term, "quote": sentence,
                       "polarity": "negative" if negative else "uncertain" if uncertain else "affirmative"}

def _website_failure(data):
    """Provider success does not establish that the requested page was fetched."""
    for record in (data, data.get("metadata"), data.get("meta")):
        if not isinstance(record, dict):
            continue
        if record.get("ok") is False or record.get("success") is False or record.get("error"):
            return "error"
        state = str(record.get("status", "")).strip().lower()
        if state in BAD:
            return state
        codes = [record[key] for key in ("statusCode", "http_status") if record.get(key) is not None]
        if state.isdigit() or type(record.get("status")) in (int, float):
            codes.append(record["status"])
        for code in codes:
            if type(code) in (int, float) and math.isfinite(code) and int(code) == code:
                code = str(int(code))
            if not re.fullmatch(r"\d{3}", str(code).strip()):
                raise ValueError("Website HTTP status must be a three-digit code")
            if not (200 <= int(code) < 300 or int(code) == 304):
                return "error"
    return None

def _unwrap(data, kind):
    """Only known response envelopes; never guess the first arbitrary array."""
    for _ in range(10):
        if isinstance(data, list):
            return data, "success"
        if not isinstance(data, dict):
            raise ValueError("Unsupported source payload type")
        if kind == "website":
            failure = _website_failure(data)
            if failure:
                return [], failure
        if data.get("ok") is False or data.get("error"):
            return [], "error"
        state = str(data.get("status", "")).lower()
        if state in BAD:
            return [], state
        if isinstance(data.get("meta"), dict):
            code = data["meta"].get("status")
            if isinstance(code, int) and code >= 400:
                return [], "error"
        if kind == "website" and any(k in data for k in ("text", "markdown", "content")):
            return [data], "success"
        keys = ("results", "pages") if kind == "website" else ("listings", "jobs", "job_listings")
        for key in keys:
            if key in data:
                if not isinstance(data[key], list):
                    raise ValueError(f"{key} must be an array")
                return data[key], "success"
        if "toolResponse" in data:
            data = data["toolResponse"]
        elif "rawV2" in data:
            data = data["rawV2"]
        elif "raw" in data:
            data = data["raw"]
        elif "data" in data:
            data = data["data"]
        elif "result" in data:
            data = data["result"]
        else:
            raise ValueError("Unsupported source envelope; add an explicit adapter")
    raise ValueError("Source envelope too deeply nested")

def source(cell, kind):
    if cell is None or not str(cell).strip() or str(cell).strip() == "null":
        return [], "missing"
    try:
        data = json.loads(cell) if isinstance(cell, str) else cell
    except (TypeError, json.JSONDecodeError) as exc:
        raise ValueError(f"Malformed {kind} JSON; wrap raw text explicitly") from exc
    items, state = _unwrap(data, kind)
    documents = []
    unavailable = []
    for item in items:
        if not isinstance(item, dict):
            raise ValueError("Source rows must be objects")
        if kind == "website":
            failure = _website_failure(item)
            if failure:
                unavailable.append(failure)
                continue
        info = item.get("job_details", item.get("attributes", item)) if kind == "jobs" else item
        if not isinstance(info, dict):
            raise ValueError("Invalid nested source record")
        title = info.get("title", info.get("job_title", ""))
        body = (info.get("description", info.get("job_description", ""))
                if kind == "jobs" else info.get("text", info.get("markdown", info.get("content", ""))))
        if body is None: body = ""
        if title is None: title = ""
        if not isinstance(body, str) or not isinstance(title, str):
            raise ValueError("Source text/title must be strings")
        url = info.get("url", info.get("job_url", item.get("url", ""))) or ""
        if not isinstance(url, str):
            raise ValueError("Source URL must be a string")
        if kind == "jobs" and not title and not body:
            raise ValueError("Unrecognized job record, not a valid empty job list")
        if kind == "website" and not body.strip():
            unavailable.append("empty")
            continue
        documents.append({"title": title, "description": body, "url": url,
                          "text": f"{title}\n{body}" if kind == "jobs" else body,
                          "source_type": "job_listing" if kind == "jobs" else "website"})
    if kind == "website" and unavailable:
        # Retain successful page evidence, but partial collection cannot prove absence.
        state = "partial" if documents else next((s for s in unavailable if s != "empty"), "empty")
    elif kind == "website" and not documents and state == "success":
        state = "empty"  # Empty scrape is NOT observed absence.
    return documents, state

def parse_website_content(cell):
    docs, _ = source(cell, "website")
    return " ".join(d["text"] for d in docs).lower(), docs

def parse_job_listings(cell):
    docs, _ = source(cell, "jobs")
    return docs, " ".join(d["text"] for d in docs).lower()

def auto_detect_columns(headers):
    normalized = [h.strip().lower() for h in headers]
    result = []
    for name in ("website", "jobs"):
        hits = [i for i, h in enumerate(normalized) if h == name]
        if len(hits) > 1:
            raise ValueError(f"Duplicate {name} columns")
        result.append(hits[0] if hits else None)
    return tuple(result)

def utc(value):
    result = datetime.fromisoformat(value.replace("Z", "+00:00"))
    if result.tzinfo is None:
        raise ValueError("Dates must contain a timezone")
    return result

def load_accounts(path, website_col=None, jobs_col=None, status_col="status", partition="all"):
    with open(path, encoding="utf-8-sig", newline="") as f:
        reader = csv.DictReader(f)
        headers = reader.fieldnames or []
        if len(headers) != len(set(headers)):
            raise ValueError("Duplicate CSV headers")
        rows = list(reader)
    if "domain" not in headers or status_col not in headers:
        raise ValueError("CSV requires domain and outcome/status columns")
    wc, jc = auto_detect_columns(headers)
    wc = wc if website_col is None else website_col
    jc = jc if jobs_col is None else jobs_col
    if wc is None and jc is None:
        raise ValueError("No website/jobs columns; pass explicit column indices")
    if wc is not None and wc == jc:
        raise ValueError("Website/jobs cannot share a column")
    for index in (wc, jc):
        if index is not None and not 0 <= index < len(headers):
            raise ValueError("Column index out of range")
    groups = defaultdict(set)
    accounts = {}
    duplicate_rows = 0
    for line, row in enumerate(rows, 2):
        if None in row or any(value is None for value in row.values()):
            raise ValueError(f"Malformed CSV row {line}")
        label = row[status_col].strip().lower()
        if label not in {"won", "lost", "unlabeled", "lookalike"}:
            raise ValueError(f"Unknown status at row {line}: {label}")
        raw_domain = row["domain"].strip().lower()
        parsed = urlparse(raw_domain if "://" in raw_domain else "//" + raw_domain)
        domain = (parsed.hostname or "").removeprefix("www.").rstrip(".")
        if not domain or "." not in domain or " " in domain:
            raise ValueError(f"Invalid domain at row {line}")
        account = row.get("account_id", "").strip() or domain
        group = row.get("parent_id", "").strip() or account
        split = row.get("split", "").strip() or "discovery"
        if split not in {"discovery", "validation"}:
            raise ValueError("split must be discovery or validation")
        groups[group].add(split)
        if len(groups[group]) > 1:
            raise ValueError(f"Parent/account overlaps discovery and validation: {group}")
        if account in accounts and (accounts[account]["status"] != label or
                                    accounts[account]["domain"] != domain):
            raise ValueError(f"Conflicting account outcomes/identity: {account}; define cohort upstream")
        if account in accounts and accounts[account]["parent_id"] != group:
            raise ValueError(f"Conflicting parent for account: {account}")
        if split == "validation":
            for field in ("known_at", "scored_at"):
                if not row.get(field):
                    raise ValueError("Validation requires conservative row known_at and scored_at")
            if utc(row["known_at"]) > utc(row["scored_at"]):
                raise ValueError("Source became known after scoring cutoff")
        docs, coverage = {}, {}
        for kind, index in (("website", wc), ("jobs", jc)):
            try:
                docs[kind], coverage[kind] = source(row[headers[index]], kind) if index is not None else ([], "missing")
            except ValueError as exc:
                raise ValueError(f"Row {line}, {kind}: {exc}") from exc
        if account not in accounts:
            accounts[account] = {"account_id": account, "parent_id": group, "domain": domain,
                                 "status": label, "split": split, "documents": docs, "coverage": coverage}
        else:
            duplicate_rows += 1
            old = accounts[account]
            for kind in docs:
                seen = {(d["url"], d["text"]) for d in old["documents"][kind]}
                old["documents"][kind].extend(d for d in docs[kind] if (d["url"], d["text"]) not in seen)
                if old["coverage"][kind] != coverage[kind]:
                    # Mixed success/error cannot establish observed absence.
                    old["coverage"][kind] = "partial"
    selected = [a for a in accounts.values() if partition == "all" or a["split"] == partition]
    return selected, {"input_rows": len(rows), "unique_accounts": len(accounts),
                      "merged_duplicate_rows": duplicate_rows,
                      "unit": "account; group-aware split, not cluster-adjusted inference"}

def concepts(config):
    if not isinstance(config, dict):
        raise ValueError("Config must be category -> list of phrases/concepts")
    output, seen = [], set()
    for category, entries in config.items():
        if not isinstance(entries, list):
            raise ValueError("Each config category must be a list")
        for entry in entries:
            if isinstance(entry, str):
                name, aliases = entry, [entry]
            elif isinstance(entry, dict):
                name, aliases = entry.get("name"), entry.get("aliases")
            else:
                raise ValueError("Invalid concept")
            if not isinstance(name, str) or not name.strip() or not isinstance(aliases, list) or not aliases:
                raise ValueError("Concept requires name and nonempty aliases")
            aliases = sorted(set(aliases)) if all(isinstance(a, str) for a in aliases) else aliases
            for alias in aliases: pattern(alias)
            key = (category, name.casefold())
            if key in seen: raise ValueError("Duplicate concept in category")
            seen.add(key)
            output.append((category, name, aliases))
    return output

def wilson(k, n):
    if not n: return None
    z = 1.959963984540054
    mid = (k/n + z*z/(2*n))/(1+z*z/n)
    half = z*math.sqrt(k/n*(1-k/n)/n + z*z/(4*n*n))/(1+z*z/n)
    return [max(0, mid-half), min(1, mid+half)]

def fisher(a, b, c, d):
    """Two-sided Fisher test; descriptive account-independence assumption."""
    r1, r2, col = a+b, c+d, a+c
    n = r1+r2
    if not r1 or not r2: return None
    def choose(n, k):
        return math.lgamma(n+1)-math.lgamma(k+1)-math.lgamma(n-k+1)
    def logp(k):
        return choose(r1, k)+choose(r2, col-k)-choose(n, col)
    observed = logp(a)
    return min(1.0, sum(math.exp(logp(k)) for k in range(max(0, col-r2), min(r1, col)+1)
                        if logp(k) <= observed + 1e-10))

def adjust(results):
    eligible = [r for r in results if r["p_value"] is not None]
    order = sorted(eligible, key=lambda r: r["p_value"], reverse=True)
    m = len(order)
    harmonic = sum(1/i for i in range(1, m+1))
    previous = 1.0
    for rank, row in zip(range(m, 0, -1), order):
        previous = min(previous, row["p_value"] * m / rank)
        row["q_bh"] = previous
        row["q_by"] = min(1.0, previous*harmonic)

def analyze(input_path, keywords, tools, job_roles, website_col=None, jobs_col=None,
            status_col="status", partition="all", evidence_limit=12):
    accounts, stats = load_accounts(input_path, website_col, jobs_col, status_col, partition)
    rows = []
    for family, config in (("keywords", keywords), ("tools", tools), ("job_roles", job_roles)):
        for category, name, aliases in concepts(config):
            # Separate source strata; never pool partial website/jobs into one denominator.
            for kind in (("jobs",) if family == "job_roles" else ("website", "jobs")):
                eligible = [a for a in accounts if a["status"] in {"won", "lost"} and a["coverage"][kind] == "success"]
                eligible_ids = {a["account_id"] for a in eligible}
                counts, totals = Counter(), Counter(a["status"] for a in eligible)
                evidence, positives, excluded = [], set(), Counter()
                for a in accounts:
                    positive = False
                    local = []
                    for doc in a["documents"][kind]:
                        # Roles are matched to advertised job TITLE only, never someone mentioned in responsibilities.
                        text = doc["title"] if family == "job_roles" else doc["text"]
                        for hit in occurrences(text, aliases):
                            if hit["polarity"] == "affirmative": positive = True
                            else: excluded[hit["polarity"]] += 1
                            local.append({**hit, "account_id": a["account_id"], "company": a["domain"],
                                          "outcome": a["status"], "url": doc["url"],
                                          "page_title": doc["title"], "source_type": doc["source_type"]})
                    if positive and a["account_id"] in eligible_ids:
                        counts[a["status"]] += 1
                        positives.add(a["account_id"])
                    if local:
                        # Preserve counterevidence without flooding the sample with repeats.
                        for polarity in ("affirmative", "negative", "uncertain"):
                            example = next((e for e in local if e["polarity"] == polarity), None)
                            if example: evidence.append(example)
                buckets = defaultdict(list)
                for example in evidence:
                    buckets[(example["outcome"], example["polarity"])].append(example)
                sample = []
                while buckets and len(sample) < evidence_limit:
                    for key in sorted(list(buckets)):
                        sample.append(buckets[key].pop(0))
                        if not buckets[key]: del buckets[key]
                        if len(sample) == evidence_limit: break
                a, c, wn, ln = counts["won"], counts["lost"], totals["won"], totals["lost"]
                ratio = ((a+.5)/(wn+1))/((c+.5)/(ln+1)) if wn and ln else None
                rows.append({"family": family, "category": category, "signal": name, "aliases": aliases,
                             "source": kind, "won_count": a, "won_observed": wn,
                             "lost_count": c, "lost_observed": ln,
                             "won_prevalence_ci95": wilson(a, wn), "lost_prevalence_ci95": wilson(c, ln),
                             "feature_prevalence_ratio_jeffreys": ratio,
                             "p_value": fisher(a, wn-a, c, ln-c),
                             "q_bh": None, "q_by": None, "positive_account_count": len(positives),
                             "excluded_mentions": dict(excluded),
                             "evidence": sample,
                             "evidence_accounts_total": len({e["account_id"] for e in evidence}),
                             "evidence_mentions_total": len(evidence),
                             "evidence_truncated": len(evidence) > evidence_limit,
                             "status": "exploratory", "scoring_eligible": False})
    adjust(rows)
    stats["outcomes"] = dict(Counter(a["status"] for a in accounts))
    stats["coverage"] = {kind: {label: dict(Counter(a["coverage"][kind] for a in accounts if a["status"] == label))
                               for label in ("won", "lost", "lookalike", "unlabeled")}
                         for kind in ("website", "jobs")}
    return {"schema_version": VERSION, "statistics": stats, "signals": rows,
            "method": {"metric": "P(feature|won) / P(feature|lost), Jeffreys-smoothed; NOT win-rate lift",
                       "multiplicity_family": "all configured concept/source tests in this invocation",
                       "uncertainty": "Wilson marginal intervals; Fisher tests assume independent accounts",
                       "validation": "Exploratory output only. No automatic scoring promotion, even on a validation partition.",
                       "semantic_limit": "Lexical mentions with heuristic polarity; not confirmed use, intent or headcount.",
                       "partition": partition}}

def discover(accounts, max_phrases=500, min_accounts=2, max_n=5):
    """Outcome-blind n-gram proposals from discovery partition only."""
    if max_phrases < 1 or min_accounts < 1 or not 2 <= max_n <= 8:
        raise ValueError("Invalid phrase discovery limits")
    frequency, evidence = defaultdict(set), {}
    for account in sorted(accounts, key=lambda a: a["account_id"]):
        if account["split"] != "discovery": continue
        for kind in ("website", "jobs"):
            for doc in account["documents"][kind]:
                for sentence in sentences(doc["text"]):
                    words = re.findall(r"\b[\w]+(?:['-][\w]+)*\b", normalize(sentence).casefold())
                    for size in range(2, max_n+1):
                        for i in range(len(words)-size+1):
                            tokens = words[i:i+size]
                            if tokens[0] in STOP or tokens[-1] in STOP or all(t.isdigit() for t in tokens): continue
                            phrase = " ".join(tokens)
                            frequency[(kind, phrase)].add(account["account_id"])
                            evidence.setdefault((kind, phrase), {"company": account["domain"],
                                "url": doc["url"], "quote": sentence, "source_type": doc["source_type"]})
    ranked = sorted((key for key, ids in frequency.items() if len(ids) >= min_accounts),
                    key=lambda k: (-len(frequency[k]), -len(k[1].split()), k))
    # Round-robin by source and phrase length prevents short frequent boilerplate dominating every slot.
    buckets = defaultdict(list)
    for key in ranked: buckets[(key[0], len(key[1].split()))].append(key)
    selected = []
    while buckets and len(selected) < max_phrases:
        for bucket in sorted(list(buckets)):
            selected.append(buckets[bucket].pop(0))
            if not buckets[bucket]: del buckets[bucket]
            if len(selected) == max_phrases: break
    return {"schema_version": VERSION, "label_blind": True, "partition": "discovery",
            "candidate_count_before_limit": len(ranked), "truncated": len(selected) < len(ranked),
            "candidates": [{"phrase": phrase, "source": kind, "account_count": len(frequency[(kind, phrase)]),
                            "evidence": evidence[(kind, phrase)], "status": "candidate_not_scoring_rule"}
                           for kind, phrase in selected]}

def main():
    p = argparse.ArgumentParser(description=__doc__)
    p.add_argument("--input", required=True)
    for name in ("keywords", "tools", "job-roles"): p.add_argument("--"+name)
    p.add_argument("--output")
    p.add_argument("--website-col", type=int)
    p.add_argument("--jobs-col", type=int)
    p.add_argument("--status-col", default="status")
    p.add_argument("--partition", choices=["all", "discovery", "validation"], default="discovery")
    p.add_argument("--evidence-limit", type=int, default=12)
    p.add_argument("--discover-phrases", action="store_true")
    p.add_argument("--max-phrases", type=int, default=500)
    p.add_argument("--min-phrase-accounts", type=int, default=2)
    p.add_argument("--manifest", help="Validation: frozen_at plus config_sha256 mapping")
    args = p.parse_args()
    try:
        if args.evidence_limit < 1: raise ValueError("evidence-limit must be positive")
        if args.output and Path(args.input).resolve() == Path(args.output).resolve():
            raise ValueError("Output must not overwrite input")
        if args.discover_phrases:
            if args.partition != "discovery": raise ValueError("Mine discovery only, never validation/all")
            accounts, _ = load_accounts(args.input, args.website_col, args.jobs_col, args.status_col)
            result = discover(accounts, args.max_phrases, args.min_phrase_accounts)
        else:
            paths = {"keywords": args.keywords, "tools": args.tools, "job_roles": args.job_roles}
            if not all(paths.values()): raise ValueError("All three configs required for analysis")
            if args.partition == "all": raise ValueError("CLI analysis requires a single partition, not all")
            if args.partition == "validation":
                if not args.manifest: raise ValueError("Validation requires frozen config manifest")
                manifest = json.loads(Path(args.manifest).read_text())
                frozen = utc(manifest["frozen_at"])
                if hashlib.sha256(Path(__file__).read_bytes()).hexdigest() != manifest["analyzer_sha256"]:
                    raise ValueError("Frozen analyzer changed")
                for name, path in paths.items():
                    if hashlib.sha256(Path(path).read_bytes()).hexdigest() != manifest["config_sha256"][name]:
                        raise ValueError(f"Frozen config changed: {name}")
                with open(args.input, newline="", encoding="utf-8-sig") as f:
                    for row in csv.DictReader(f):
                        if (row.get("split") or "").strip() == "validation" and frozen > utc(row["scored_at"]):
                            raise ValueError("Config frozen after validation scoring date")
            configs = {name: json.loads(Path(path).read_text()) for name, path in paths.items()}
            result = analyze(args.input, **configs, website_col=args.website_col, jobs_col=args.jobs_col,
                             status_col=args.status_col, partition=args.partition, evidence_limit=args.evidence_limit)
            result["config_sha256"] = {name: hashlib.sha256(Path(path).read_bytes()).hexdigest() for name, path in paths.items()}
        text = json.dumps(result, indent=2, ensure_ascii=False, allow_nan=False)
        if args.output: Path(args.output).write_text(text+"\n", encoding="utf-8")
        else: print(text)
    except (ValueError, KeyError, TypeError) as exc:
        p.error(str(exc))

if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Differential signal analysis for ICP niche signal discovery.

Reads a Deepline-enriched CSV, parses exa_search website content and crustdata
job listings, computes Laplace-smoothed lift scores for keyword categories,
extracts tech stack tools, and outputs JSON results.

Usage:
    python3 analyze_signals.py \\
      --input enriched.csv \\
      --keywords keywords.json \\
      --tools tools.json \\
      --job-roles job_roles.json \\
      --output analysis.json

Options:
    --input         Path to enriched CSV (required)
    --keywords      Path to JSON file with keyword categories (required)
    --tools         Path to JSON file with tech stack tools (required)
    --job-roles     Path to JSON file with job role categories (required)
    --output        Path for JSON output (default: stdout)
    --website-col   Column index for website data (auto-detected if omitted)
    --jobs-col      Column index for job listings (auto-detected if omitted)
    --status-col    Column name for won/lost status (default: "status")

See references/keyword-catalog.md for JSON format examples and guidance on
building target-specific keyword, tool, and job role lists.
"""

import csv
import json
import sys
import re
import argparse
from collections import defaultdict

csv.field_size_limit(sys.maxsize)


def auto_detect_columns(headers):
    """Find website and jobs columns by looking for __dl_full_result__ pattern."""
    website_col = None
    jobs_col = None
    for i, h in enumerate(headers):
        if "__dl_full_result__" in h:
            # Try to determine if it's website or jobs by checking position
            if website_col is None:
                website_col = i
            elif jobs_col is None:
                jobs_col = i
    return website_col, jobs_col


def parse_website_content(cell_value):
    """Extract text content from exa_search results.
    Returns:
        combined_text: all page text concatenated (lowercased)
        pages: list of {url, title, text} per page (text is lowercased)
    """
    if not cell_value or cell_value.strip() == "":
        return "", []
    try:
        data = json.loads(cell_value)
    except (json.JSONDecodeError, TypeError):
        return str(cell_value), []

    texts = []
    pages = []

    # Handle various response shapes
    results = []
    if isinstance(data, dict):
        results = data.get("data", {}).get("results", []) if isinstance(data.get("data"), dict) else []
        if not results:
            results = data.get("results", [])
    elif isinstance(data, list):
        results = data

    for r in results:
        if isinstance(r, dict):
            text = r.get("text", "")
            url = r.get("url", "")
            title = r.get("title", "")
            if text:
                texts.append(text)
            if url:
                pages.append({"url": url, "title": title, "text": text.lower()})

    return " ".join(texts).lower(), pages


def parse_job_listings(cell_value):
    """Extract job titles and descriptions from crustdata job listings.
    Returns:
        listings: list of {title, description, url, text} per listing (text is lowercased)
        combined_text: all listing text concatenated (lowercased)
    """
    if not cell_value or cell_value.strip() == "":
        return [], ""
    try:
        data = json.loads(cell_value)
    except (json.JSONDecodeError, TypeError):
        return [], str(cell_value)

    listings = []
    all_text = []

    # Handle various response shapes
    raw_listings = []
    if isinstance(data, dict):
        # {"data": {"listings": [...]}} (legacy exa-like)
        raw_listings = data.get("data", {}).get("listings", []) if isinstance(data.get("data"), dict) else []
        # {"result": {"listings": [...]}} (Deepline/Crustdata)
        if not raw_listings and isinstance(data.get("result"), dict):
            raw_listings = data["result"].get("listings", [])
        # {"listings": [...]} (flat)
        if not raw_listings:
            raw_listings = data.get("listings", [])
    elif isinstance(data, list):
        raw_listings = data

    for entry in raw_listings:
        if isinstance(entry, dict):
            # Crustdata uses "title" and "description" (not "job_title"/"job_description")
            title = entry.get("title", entry.get("job_title", ""))
            desc = entry.get("description", entry.get("job_description", ""))
            url = entry.get("url", "")
            combined = f"{title} {desc}"
            listings.append({"title": title, "description": desc, "url": url, "text": combined.lower()})
            all_text.append(combined)

    return listings, " ".join(all_text).lower()


def substring_match(text, keyword):
    """Check if keyword appears as substring in text (case-insensitive)."""
    if not text:
        return False
    return keyword.lower().rstrip("*") in text.lower()


def laplace_lift(won_count, won_total, lost_count, lost_total):
    """Compute Laplace-smoothed lift (Bayesian posterior mean ratio with Jeffreys prior)."""
    won_rate = (won_count + 0.5) / (won_total + 1)
    lost_rate = (lost_count + 0.5) / (lost_total + 1)
    return won_rate / lost_rate


def extract_snippet(text, keyword, context_chars=40):
    """Extract a snippet around the first occurrence of keyword in text."""
    idx = text.find(keyword)
    if idx == -1:
        return None
    start = max(0, idx - context_chars)
    end = min(len(text), idx + len(keyword) + context_chars)
    snippet = text[start:end].strip()
    # Clean up: trim to word boundaries
    if start > 0:
        space = snippet.find(" ")
        if space > 0 and space < context_chars // 2:
            snippet = snippet[space + 1:]
        snippet = "..." + snippet
    if end < len(text):
        space = snippet.rfind(" ")
        if space > len(snippet) - context_chars // 2:
            snippet = snippet[:space]
        snippet = snippet + "..."
    return snippet


def find_source_evidence(keyword, companies, max_evidence=5):
    """Find exact quotes with source URLs for a keyword match.
    Returns list of evidence objects with company, source_type, quote, url, and page_title.
    """
    evidence = []
    kw = keyword.lower().rstrip("*")

    for company in companies:
        if len(evidence) >= max_evidence:
            break

        # Check website pages (per-page text has URLs)
        for page in company.get("pages", []):
            page_text = page.get("text", "")
            if kw in page_text:
                snippet = extract_snippet(page_text, kw)
                if snippet:
                    evidence.append({
                        "company": company["domain"],
                        "source_type": "website",
                        "quote": snippet,
                        "url": page.get("url", ""),
                        "page_title": page.get("title", ""),
                    })
                    break  # One match per company per source type

        # Check job listings (per-listing text has URLs)
        for listing in company.get("job_listings", []):
            listing_text = listing.get("text", "")
            if kw in listing_text:
                snippet = extract_snippet(listing_text, kw)
                job_title = listing.get("title", "")
                if snippet:
                    evidence.append({
                        "company": company["domain"],
                        "source_type": "job_listing",
                        "quote": snippet,
                        "url": listing.get("url", ""),
                        "job_title": job_title,
                    })
                    break  # One match per company per source type

    return evidence


def analyze(input_path, keywords, tools, job_roles,
            website_col=None, jobs_col=None, status_col="status"):
    """Run the full differential analysis.

    Args:
        input_path: Path to enriched CSV
        keywords: Dict of category -> list of keyword strings
        tools: Dict of category -> list of tool name strings
        job_roles: Dict of role_name -> list of role keyword strings
        website_col: Column index for website data (auto-detected if None)
        jobs_col: Column index for job listings (auto-detected if None)
        status_col: Column name for won/lost status
    """

    # Read CSV
    with open(input_path, "r", encoding="utf-8") as f:
        reader = csv.reader(f)
        headers = next(reader)
        rows = list(reader)

    # Auto-detect columns if not specified
    if website_col is None or jobs_col is None:
        auto_web, auto_jobs = auto_detect_columns(headers)
        if website_col is None:
            website_col = auto_web
        if jobs_col is None:
            jobs_col = auto_jobs

    # Find status column
    status_idx = None
    for i, h in enumerate(headers):
        if h.lower().strip() == status_col.lower():
            status_idx = i
            break

    if status_idx is None:
        raise ValueError(f"Status column '{status_col}' not found in headers: {headers}")

    # Parse companies
    companies = []
    for row in rows:
        if len(row) <= max(status_idx, website_col or 0, jobs_col or 0):
            continue

        status = row[status_idx].strip().lower()
        if status not in ("won", "lost"):
            continue

        domain = row[0].strip() if row[0] else "unknown"

        website_text = ""
        pages = []
        if website_col is not None and website_col < len(row):
            website_text, pages = parse_website_content(row[website_col])

        job_listings = []
        jobs_text = ""
        if jobs_col is not None and jobs_col < len(row):
            job_listings, jobs_text = parse_job_listings(row[jobs_col])

        # Combined text for general keyword matching
        combined_text = f"{website_text} {jobs_text}"

        companies.append({
            "domain": domain,
            "status": status,
            "website_text": website_text,
            "jobs_text": jobs_text,
            "combined_text": combined_text,
            "pages": pages,
            "job_listings": job_listings,
            "has_website": len(website_text) > 100,
            "has_jobs": len(job_listings) > 0,
        })

    won = [c for c in companies if c["status"] == "won"]
    lost = [c for c in companies if c["status"] == "lost"]
    won_total = len(won)
    lost_total = len(lost)

    # ── Keyword analysis ──
    keyword_results = {}
    for category, kws in keywords.items():
        category_results = []
        for kw in kws:
            kw_lower = kw.lower().rstrip("*")
            won_count = sum(1 for c in won if kw_lower in c["combined_text"])
            lost_count = sum(1 for c in lost if kw_lower in c["combined_text"])
            lift = laplace_lift(won_count, won_total, lost_count, lost_total)

            # Source breakdown (website vs jobs vs both)
            won_web = sum(1 for c in won if kw_lower in c["website_text"] and kw_lower not in c["jobs_text"])
            won_jobs = sum(1 for c in won if kw_lower not in c["website_text"] and kw_lower in c["jobs_text"])
            won_both = sum(1 for c in won if kw_lower in c["website_text"] and kw_lower in c["jobs_text"])

            evidence = find_source_evidence(kw, won + lost)

            category_results.append({
                "keyword": kw,
                "won_count": won_count,
                "won_pct": round(won_count / won_total * 100, 1) if won_total else 0,
                "lost_count": lost_count,
                "lost_pct": round(lost_count / lost_total * 100, 1) if lost_total else 0,
                "lift": round(lift, 2),
                "source_breakdown": {
                    "website_only": won_web,
                    "jobs_only": won_jobs,
                    "both": won_both
                },
                "evidence": evidence
            })

        category_results.sort(key=lambda x: x["lift"], reverse=True)
        keyword_results[category] = category_results

    # ── Tech stack tool analysis ──
    tool_results = {}
    for category, tool_list in tools.items():
        category_results = []
        for tool in tool_list:
            tool_lower = tool.lower()
            won_count = sum(1 for c in won if tool_lower in c["combined_text"])
            lost_count = sum(1 for c in lost if tool_lower in c["combined_text"])

            if won_count < 2 and lost_count < 1:
                continue

            lift = laplace_lift(won_count, won_total, lost_count, lost_total)
            evidence = find_source_evidence(tool, won + lost)

            category_results.append({
                "tool": tool,
                "won_count": won_count,
                "won_pct": round(won_count / won_total * 100, 1) if won_total else 0,
                "lost_count": lost_count,
                "lost_pct": round(lost_count / lost_total * 100, 1) if lost_total else 0,
                "lift": round(lift, 2),
                "evidence": evidence
            })

        category_results.sort(key=lambda x: x["lift"], reverse=True)
        tool_results[category] = category_results

    # ── Job role analysis ──
    job_role_results = {}
    won_with_jobs = [c for c in won if c["has_jobs"]]
    lost_with_jobs = [c for c in lost if c["has_jobs"]]

    for role_name, role_keywords in job_roles.items():
        won_match = sum(1 for c in won_with_jobs
                        if any(rk in c["jobs_text"] for rk in role_keywords))
        lost_match = sum(1 for c in lost_with_jobs
                         if any(rk in c["jobs_text"] for rk in role_keywords))

        job_role_results[role_name] = {
            "won_count": won_match,
            "won_with_jobs": len(won_with_jobs),
            "won_pct": round(won_match / len(won_with_jobs) * 100, 1) if won_with_jobs else 0,
            "lost_count": lost_match,
            "lost_with_jobs": len(lost_with_jobs),
            "lost_pct": round(lost_match / len(lost_with_jobs) * 100, 1) if lost_with_jobs else 0,
        }

    # ── Stats ──
    won_with_content = sum(1 for c in won if c["has_website"])
    lost_with_content = sum(1 for c in lost if c["has_website"])
    avg_won_chars = (sum(len(c["website_text"]) for c in won) / won_total) if won_total else 0
    avg_lost_chars = (sum(len(c["website_text"]) for c in lost) / lost_total) if lost_total else 0

    stats = {
        "won_total": won_total,
        "lost_total": lost_total,
        "won_with_content": won_with_content,
        "lost_with_content": lost_with_content,
        "won_with_jobs": len(won_with_jobs),
        "lost_with_jobs": len(lost_with_jobs),
        "won_coverage_pct": round(won_with_content / won_total * 100, 1) if won_total else 0,
        "lost_coverage_pct": round(lost_with_content / lost_total * 100, 1) if lost_total else 0,
        "avg_won_chars": round(avg_won_chars),
        "avg_lost_chars": round(avg_lost_chars),
    }

    # ── Anti-fit signals (lift < 0.5) ──
    anti_fit = []
    for category, results in keyword_results.items():
        for r in results:
            if r["lift"] < 0.5 and (r["won_count"] > 0 or r["lost_count"] > 1):
                anti_fit.append({**r, "category": category})
    anti_fit.sort(key=lambda x: x["lift"])

    return {
        "keyword_results": keyword_results,
        "tool_results": tool_results,
        "job_results": job_role_results,
        "stats": stats,
        "anti_fit": anti_fit,
    }


def main():
    parser = argparse.ArgumentParser(
        description="Differential signal analysis for ICP",
        epilog="See references/keyword-catalog.md for JSON format examples."
    )
    parser.add_argument("--input", required=True, help="Path to enriched CSV")
    parser.add_argument("--keywords", required=True, help="Path to JSON file with keyword categories")
    parser.add_argument("--tools", required=True, help="Path to JSON file with tech stack tools")
    parser.add_argument("--job-roles", required=True, help="Path to JSON file with job role categories")
    parser.add_argument("--output", help="Path for JSON output (default: stdout)")
    parser.add_argument("--website-col", type=int, help="Column index for website data")
    parser.add_argument("--jobs-col", type=int, help="Column index for job listings")
    parser.add_argument("--status-col", default="status", help="Column name for won/lost status")

    args = parser.parse_args()

    with open(args.keywords) as f:
        keywords = json.load(f)

    with open(args.tools) as f:
        tools = json.load(f)

    with open(args.job_roles) as f:
        job_roles = json.load(f)

    results = analyze(
        input_path=args.input,
        keywords=keywords,
        tools=tools,
        job_roles=job_roles,
        website_col=args.website_col,
        jobs_col=args.jobs_col,
        status_col=args.status_col,
    )

    output = json.dumps(results, indent=2)
    if args.output:
        with open(args.output, "w") as f:
            f.write(output)
        print(f"Analysis written to {args.output}", file=sys.stderr)
        print(f"Stats: {json.dumps(results['stats'], indent=2)}", file=sys.stderr)
    else:
        print(output)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Deduplication utilities for deepline-scoring prospect lists.

Primary match: apex-domain (public-suffix-aware, handles multi-label TLDs).
Fallback match: fuzzy company name (after stripping corporate suffixes).

The standard library only — no pip dependencies. `difflib.SequenceMatcher`
is used for the fuzzy name ratio so this runs anywhere Python 3 does.

Usage from another script:

    from dedupe_utils_v2 import extract_apex, norm_name, match_against_existing

    existing = load_existing_csv("customers.csv")   # rows with 'domain' or 'name'
    candidates = load_candidates("prospects.csv")   # rows with 'domain' and 'name'
    actionable, matched = match_against_existing(candidates, existing, name_threshold=0.85)

Usage from the command line:

    python3 dedupe_utils_v2.py \\
        --existing customers.csv \\
        --candidates prospects.csv \\
        --out-actionable net_new.csv \\
        --out-matched already_known.csv

Why this exists:
  - Raw string match misses parent-company relationships. amsynergy.nikon.com
    and nikon.com refer to the same buyer-side organization, but a naive set
    lookup treats them as unrelated.
  - Name matching alone is noisy: a candidate named "Rocket Propulsion Systems"
    can collide with an unrelated CRM row named "Rocket Propulsion" as a
    substring. Apex domain is a stronger primary key when it's available.
  - The fix is a layered check: match on apex domain first (strong signal),
    then fall back to normalized fuzzy name match with a high threshold only
    when no domain match exists.
"""
from __future__ import annotations

import argparse
import csv
import re
import sys
from difflib import SequenceMatcher
from typing import Iterable, Sequence
from urllib.parse import urlparse


# ----------------------------------------------------------------------
# Apex domain extraction
# ----------------------------------------------------------------------

# A curated list of multi-label public suffixes that show up often in B2B
# datasets. This is NOT a complete public-suffix-list dump — a production
# system should use `tldextract` — but it covers the countries that have
# appeared in real runs (US, UK, JP, KR, AU, DE, BR, CN, etc).
MULTI_LABEL_SUFFIXES: set[str] = {
    # United Kingdom
    "co.uk", "org.uk", "ac.uk", "gov.uk", "ltd.uk", "plc.uk", "net.uk",
    # Japan
    "co.jp", "ac.jp", "or.jp", "go.jp", "ne.jp",
    # Korea
    "co.kr", "ac.kr", "go.kr", "or.kr", "re.kr",
    # Australia / New Zealand
    "com.au", "net.au", "org.au", "edu.au", "gov.au",
    "co.nz", "ac.nz",
    # Brazil / China / India / Israel / South Africa
    "com.br", "com.cn", "net.cn", "co.il", "co.in", "ac.in", "edu.in",
    "co.za", "ac.za",
    # Europe — country-specific commercial
    "co.it", "co.es", "com.es",
    # APAC / LATAM misc
    "com.mx", "edu.mx", "org.mx",
    "com.hk", "com.sg", "com.tr", "com.tw", "com.ar", "com.co", "com.pe",
    "com.ph", "com.my", "com.pk", "com.eg", "com.sa", "com.ua", "com.vn",
    "co.th",
}


def extract_apex(url_or_host: str) -> str:
    """Conservative LIMITED suffix normalization, not a complete PSL or parent resolver."""
    import ipaddress
    if not isinstance(url_or_host, str) or not url_or_host.strip():
        return ""
    value=url_or_host.strip()
    try:
        parsed=urlparse(value if "://" in value else "//"+value)
        if parsed.scheme and parsed.scheme not in ("http","https"):
            return ""
        if parsed.username or parsed.password:
            return ""
        host=(parsed.hostname or "").rstrip(".").lower().encode("idna").decode("ascii")
        if not host or "." not in host:
            return ""
        try:
            ipaddress.ip_address(host)
            return ""
        except ValueError:
            pass
        labels=host.split(".")
        if any(not re.fullmatch(r"[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?", label) for label in labels):
            return ""
        if not re.search(r"[a-z]", labels[-1]):
            return ""
    except (ValueError, UnicodeError):
        return ""
    while labels[0]=="www" and len(labels)>2:
        labels=labels[1:]
    suffix=".".join(labels[-2:])
    private={"github.io","blogspot.com","wordpress.com","myshopify.com","webflow.io","vercel.app","netlify.app","pages.dev"}
    if suffix in MULTI_LABEL_SUFFIXES or suffix in private:
        return ".".join(labels[-3:]) if len(labels)>2 else ""
    # Unknown multi-label country suffixes: preserve host to avoid merging tenants.
    if len(labels)>2 and len(labels[-1])==2:
        return ".".join(labels)
    return ".".join(labels[-2:])


# ----------------------------------------------------------------------
# Company-name normalization + fuzzy matching
# ----------------------------------------------------------------------

# Corporate suffix tokens to strip before comparing two company names.
# Order matters only for readability — the regex compiles to a single pass.
_CORP_SUFFIX_TOKENS: Sequence[str] = (
    "inc", "llc", "ltd", "gmbh", "sa", "ag", "co", "corp", "corporation",
    "company", "group", "holdings", "limited", "plc", "bv", "srl", "spa",
    "oy", "ab", "pte", "pty", "kg", "mbh", "cie", "sarl",
    "tech", "technologies", "systems", "solutions", "industries",
    "international", "global",
)

_CORP_SUFFIX_RE = re.compile(
    r"\b(?:" + "|".join(re.escape(t) for t in _CORP_SUFFIX_TOKENS) + r")\b\.?",
    flags=re.IGNORECASE,
)


def norm_name(company_name: str) -> str:
    """Normalize a company name for fuzzy comparison.

    - Lowercase
    - Strip corporate suffix tokens (Inc, LLC, Ltd, GmbH, Holdings, ...)
    - Keep only [a-z0-9 ]
    - Collapse whitespace

    Returns "" when normalization leaves less than 3 characters — names that
    short are too noisy to match reliably.

    Examples:
        norm_name("Astura Medical, Inc.") -> "astura medical"
        norm_name("MBDA Systems Holdings Ltd") -> "mbda"
        norm_name("3DMorphic") -> "3dmorphic"
        norm_name("SA") -> ""
    """
    if not company_name:
        return ""
    n = company_name.strip().lower()
    n = _CORP_SUFFIX_RE.sub("", n)
    n = re.sub(r"[^a-z0-9 ]", "", n)
    n = re.sub(r"\s+", " ", n).strip()
    if len(n) < 3:
        return ""
    return n


def name_similarity(a: str, b: str) -> float:
    """Return a 0..1 similarity ratio between two company names after
    normalization. Uses difflib.SequenceMatcher which is stdlib and close
    enough to Levenshtein ratio for typical corporate-name matching."""
    na = norm_name(a)
    nb = norm_name(b)
    if not na or not nb:
        return 0.0
    return SequenceMatcher(None, na, nb).ratio()


# ----------------------------------------------------------------------
# Combined match-against-existing helper
# ----------------------------------------------------------------------

def build_existing_index(
    existing_rows: Iterable[dict],
    domain_field: str = "domain",
    name_field: str = "name",
    website_field: str | None = "website",
) -> tuple[set[str], dict[str, str]]:
    """Build an apex-domain set + normalized-name index from existing rows.

    Returns:
        (apex_set, name_to_apex):
          apex_set is the set of apex domains present in the existing list.
          name_to_apex maps every normalized company name to the apex it
            came from, so a name-match can report which row it collided with.

    existing_rows can be anything iterable of dicts. Pass rows from your
    CRM export, a previous prospect-list CSV, a customer-list download,
    or whatever the user provides as "do not re-contact".
    """
    apex_set: set[str] = set()
    name_to_apex: dict[str, str] = {}

    for r in existing_rows:
        apex = ""
        # Prefer an explicit domain field, fall back to website.
        for fld in (domain_field, website_field):
            if fld and r.get(fld):
                apex = extract_apex(r[fld])
                if apex:
                    break

        if apex:
            apex_set.add(apex)

        nm = norm_name(r.get(name_field, "")) if name_field else ""
        if nm:
            # Don't overwrite a shorter existing key with a longer one; the
            # first occurrence wins so downstream messages are stable.
            name_to_apex.setdefault(nm, apex or "")

    return apex_set, name_to_apex


def check_duplicate(
    candidate: dict,
    apex_set: set[str],
    name_to_apex: dict[str, str],
    domain_field: str = "domain",
    name_field: str = "name",
    website_field: str | None = "website",
    name_threshold: float = 0.85,
) -> tuple[bool, str]:
    """Check a single candidate row against the existing index.

    Returns (is_duplicate, reason). reason is a short string explaining the
    match (e.g. "apex:nikon.com" or "name:astura medical (0.91)") or "" when
    the candidate is net-new.

    The match is layered:
      1. Apex domain match against apex_set (strong, preferred).
      2. If no domain match, walk name_to_apex looking for a fuzzy match
         above name_threshold. Only used as a fallback because name-only
         matches are noisy (e.g. "rocket propulsion" as a substring).
    """
    # Step A: apex domain
    apex = ""
    for fld in (domain_field, website_field):
        if fld and candidate.get(fld):
            apex = extract_apex(candidate[fld])
            if apex:
                break
    if apex and apex in apex_set:
        return True, f"apex:{apex}"

    # Step B: fuzzy company name
    cand_name = candidate.get(name_field, "") if name_field else ""
    nc = norm_name(cand_name)
    if nc:
        best_key = ""
        best_ratio = 0.0
        for existing_name in name_to_apex:
            ratio = SequenceMatcher(None, nc, existing_name).ratio()
            if ratio > best_ratio:
                best_ratio = ratio
                best_key = existing_name
                if ratio >= 0.99:  # perfect match, stop early
                    break
        if best_ratio >= name_threshold:
            return True, f"review:name:{best_key} ({best_ratio:.2f}); identity unverified"

    return False, ""


def match_against_existing(
    candidates: Iterable[dict],
    existing: Iterable[dict],
    domain_field: str = "domain",
    name_field: str = "name",
    website_field: str | None = "website",
    name_threshold: float = 0.85,
) -> tuple[list[dict], list[dict]]:
    """Split candidates into (actionable, matched) against an existing list.

    Each row in the output carries a `dedupe_match` field describing why
    it was classified that way — empty for actionable rows, populated with
    the match reason for matched rows. This makes it trivial to surface in
    reports and explain to the user why a row was excluded.
    """
    apex_set, name_to_apex = build_existing_index(
        existing,
        domain_field=domain_field,
        name_field=name_field,
        website_field=website_field,
    )

    actionable: list[dict] = []
    matched: list[dict] = []
    for c in candidates:
        is_dup, reason = check_duplicate(
            c, apex_set, name_to_apex,
            domain_field=domain_field,
            name_field=name_field,
            website_field=website_field,
            name_threshold=name_threshold,
        )
        out = dict(c)
        out["dedupe_match"] = reason
        out["identity_status"] = "review_required" if reason.startswith("review:") else "domain_match" if is_dup else "not_matched"
        if is_dup:
            matched.append(out)
        else:
            actionable.append(out)
    return actionable, matched


# ----------------------------------------------------------------------
# Command-line entry point
# ----------------------------------------------------------------------

def _read_csv(path: str) -> list[dict]:
    csv.field_size_limit(sys.maxsize)
    with open(path) as f:
        return list(csv.DictReader(f))


def _write_csv(path: str, rows: list[dict]) -> None:
    if not rows:
        with open(path, "w", newline="") as f:
            f.write("")
        return
    fieldnames = list(dict.fromkeys(key for row in rows for key in row))
    with open(path, "w", newline="") as f:
        w = csv.DictWriter(f, fieldnames=fieldnames)
        w.writeheader()
        w.writerows(rows)


def _selftest() -> int:
    """Sanity-check the apex and name helpers. Exits non-zero on failure."""
    apex_cases = [
        ("amsynergy.nikon.com", "nikon.com"),
        ("industry.nikon.com", "nikon.com"),
        ("nikon.co.jp", "nikon.co.jp"),
        ("www.bbc.co.uk", "bbc.co.uk"),
        ("corporate.arcelormittal.com", "arcelormittal.com"),
        ("blog.company.com", "company.com"),
        ("firehawkaerospace.com", "firehawkaerospace.com"),
        ("https://www.example.com/foo/bar", "example.com"),
        ("", ""),
        ("localhost", ""),
    ]
    name_cases = [
        ("Astura Medical, Inc.", "astura medical"),
        ("MBDA Systems Holdings Ltd", "mbda"),
        ("3DMorphic", "3dmorphic"),
        ("Firehawk Aerospace", "firehawk aerospace"),
    ]

    failures = 0
    for inp, expected in apex_cases:
        got = extract_apex(inp)
        ok = got == expected
        print(f"apex  {'OK ' if ok else 'FAIL'}  {inp!r:<45} -> {got!r}  (expected {expected!r})")
        if not ok:
            failures += 1
    for inp, expected in name_cases:
        got = norm_name(inp)
        ok = got == expected
        print(f"name  {'OK ' if ok else 'FAIL'}  {inp!r:<45} -> {got!r}  (expected {expected!r})")
        if not ok:
            failures += 1

    # Layered-match sanity check
    existing = [{"domain": "nikon.com", "name": "Nikon Corporation"}]
    candidates = [
        {"domain": "amsynergy.nikon.com", "name": "Nikon AM Synergy"},
        {"domain": "ad-astra.com", "name": "Ad Astra Rocket Company"},
        {"domain": "", "name": "Nikon Corp"},  # name-only fallback
    ]
    actionable, matched = match_against_existing(candidates, existing)
    print()
    print("actionable:", actionable)
    print("matched:   ", matched)
    if len(actionable) != 1 or actionable[0]["domain"] != "ad-astra.com":
        print("FAIL: expected only ad-astra.com to be actionable")
        failures += 1

    return 0 if failures == 0 else 1


def _main() -> int:
    parser = argparse.ArgumentParser(
        description="Dedupe a candidate prospect list against an existing list."
    )
    parser.add_argument("--existing", help="CSV with the do-not-contact list")
    parser.add_argument("--candidates", help="CSV with the candidate prospect list")
    parser.add_argument("--out-actionable", help="Write actionable rows here")
    parser.add_argument("--out-matched", help="Write matched-against-existing rows here")
    parser.add_argument("--domain-field", default="domain")
    parser.add_argument("--name-field", default="name")
    parser.add_argument("--website-field", default="website")
    parser.add_argument("--name-threshold", type=float, default=0.85)
    parser.add_argument("--selftest", action="store_true",
                        help="Run built-in sanity tests and exit")
    args = parser.parse_args()

    if args.selftest:
        return _selftest()

    if not (args.existing and args.candidates):
        parser.print_help()
        return 2

    existing = _read_csv(args.existing)
    candidates = _read_csv(args.candidates)
    actionable, matched = match_against_existing(
        candidates, existing,
        domain_field=args.domain_field,
        name_field=args.name_field,
        website_field=args.website_field,
        name_threshold=args.name_threshold,
    )
    if args.out_actionable:
        _write_csv(args.out_actionable, actionable)
    if args.out_matched:
        _write_csv(args.out_matched, matched)
    print(f"existing:   {len(existing)}")
    print(f"candidates: {len(candidates)}")
    print(f"actionable: {len(actionable)}")
    print(f"matched:    {len(matched)}")
    return 0


if __name__ == "__main__":
    sys.exit(_main())

#!/usr/bin/env python3
"""
Deduplication utilities for deepline-scoring prospect lists.

Primary match: apex-domain (public-suffix-aware, handles multi-label TLDs).
Fallback match: fuzzy company name (after stripping corporate suffixes).

The standard library only — no pip dependencies. `difflib.SequenceMatcher`
is used for the fuzzy name ratio so this runs anywhere Python 3 does.

Usage from another script:

    from dedupe_utils import extract_apex, norm_name, match_against_existing

    existing = load_existing_csv("customers.csv")   # rows with 'domain' or 'name'
    candidates = load_candidates("prospects.csv")   # rows with 'domain' and 'name'
    actionable, matched = match_against_existing(candidates, existing, name_threshold=0.85)

Usage from the command line:

    python3 dedupe_utils.py \\
        --existing customers.csv \\
        --candidates prospects.csv \\
        --out-actionable net_new.csv \\
        --out-matched already_known.csv

Why this exists:
  - Raw string match misses parent-company relationships. amsynergy.nikon.com
    and nikon.com refer to the same buyer-side organization, but a naive set
    lookup treats them as unrelated.
  - Name matching alone is noisy: a candidate named "Rocket Propulsion Systems"
    can collide with an unrelated CRM row named "Rocket Propulsion" as a
    substring. Apex domain is a stronger primary key when it's available.
  - The fix is a layered check: match on apex domain first (strong signal),
    then fall back to normalized fuzzy name match with a high threshold only
    when no domain match exists.
"""
from __future__ import annotations

import argparse
import csv
import re
import sys
from difflib import SequenceMatcher
from typing import Iterable, Sequence
from urllib.parse import urlparse


# ----------------------------------------------------------------------
# Apex domain extraction
# ----------------------------------------------------------------------

# A curated list of multi-label public suffixes that show up often in B2B
# datasets. This is NOT a complete public-suffix-list dump — a production
# system should use `tldextract` — but it covers the countries that have
# appeared in real runs (US, UK, JP, KR, AU, DE, BR, CN, etc).
MULTI_LABEL_SUFFIXES: set[str] = {
    # United Kingdom
    "co.uk", "org.uk", "ac.uk", "gov.uk", "ltd.uk", "plc.uk", "net.uk",
    # Japan
    "co.jp", "ac.jp", "or.jp", "go.jp", "ne.jp",
    # Korea
    "co.kr", "ac.kr", "go.kr", "or.kr", "re.kr",
    # Australia / New Zealand
    "com.au", "net.au", "org.au", "edu.au", "gov.au",
    "co.nz", "ac.nz",
    # Brazil / China / India / Israel / South Africa
    "com.br", "com.cn", "net.cn", "co.il", "co.in", "ac.in", "edu.in",
    "co.za", "ac.za",
    # Europe — country-specific commercial
    "co.it", "co.es", "com.es",
    # APAC / LATAM misc
    "com.mx", "edu.mx", "org.mx",
    "com.hk", "com.sg", "com.tr", "com.tw", "com.ar", "com.co", "com.pe",
    "com.ph", "com.my", "com.pk", "com.eg", "com.sa", "com.ua", "com.vn",
    "co.th",
}


def extract_apex(url_or_host: str) -> str:
    """Normalize a URL or bare hostname to its registrable apex domain.

    Returns an empty string when the input can't be parsed into something
    that looks like a domain (empty input, IP literal, obvious garbage).
    This is intentional — downstream code should treat "" as "skip this
    row" rather than as a valid apex.

    Examples:
        extract_apex("amsynergy.nikon.com") -> "nikon.com"
        extract_apex("industry.nikon.com") -> "nikon.com"
        extract_apex("nikon.co.jp") -> "nikon.co.jp"
        extract_apex("www.bbc.co.uk") -> "bbc.co.uk"
        extract_apex("https://corporate.arcelormittal.com/careers") -> "arcelormittal.com"
        extract_apex("") -> ""
    """
    if not url_or_host:
        return ""
    s = url_or_host.strip().lower()
    if not s:
        return ""

    # Prepend a scheme if missing so urlparse populates .netloc.
    if not re.match(r"^https?://", s):
        s = "http://" + s

    try:
        host = urlparse(s).netloc
    except Exception:
        return ""

    # Strip port + leading www. variants.
    host = host.split(":")[0].strip("/").split("/")[0]
    while host.startswith("www."):
        host = host[4:]

    # Reject obvious non-domains.
    if "." not in host or " " in host:
        return ""

    parts = host.split(".")
    if len(parts) < 2:
        return host

    # If the last two labels form a multi-label suffix (e.g., co.uk), the
    # registrable root is the last THREE labels.
    if len(parts) >= 3:
        last_two = ".".join(parts[-2:])
        if last_two in MULTI_LABEL_SUFFIXES:
            return ".".join(parts[-3:])

    return ".".join(parts[-2:])


# ----------------------------------------------------------------------
# Company-name normalization + fuzzy matching
# ----------------------------------------------------------------------

# Corporate suffix tokens to strip before comparing two company names.
# Order matters only for readability — the regex compiles to a single pass.
_CORP_SUFFIX_TOKENS: Sequence[str] = (
    "inc", "llc", "ltd", "gmbh", "sa", "ag", "co", "corp", "corporation",
    "company", "group", "holdings", "limited", "plc", "bv", "srl", "spa",
    "oy", "ab", "pte", "pty", "kg", "mbh", "cie", "sarl",
    "tech", "technologies", "systems", "solutions", "industries",
    "international", "global",
)

_CORP_SUFFIX_RE = re.compile(
    r"\b(?:" + "|".join(re.escape(t) for t in _CORP_SUFFIX_TOKENS) + r")\b\.?",
    flags=re.IGNORECASE,
)


def norm_name(company_name: str) -> str:
    """Normalize a company name for fuzzy comparison.

    - Lowercase
    - Strip corporate suffix tokens (Inc, LLC, Ltd, GmbH, Holdings, ...)
    - Keep only [a-z0-9 ]
    - Collapse whitespace

    Returns "" when normalization leaves less than 3 characters — names that
    short are too noisy to match reliably.

    Examples:
        norm_name("Astura Medical, Inc.") -> "astura medical"
        norm_name("MBDA Systems Holdings Ltd") -> "mbda"
        norm_name("3DMorphic") -> "3dmorphic"
        norm_name("SA") -> ""
    """
    if not company_name:
        return ""
    n = company_name.strip().lower()
    n = _CORP_SUFFIX_RE.sub("", n)
    n = re.sub(r"[^a-z0-9 ]", "", n)
    n = re.sub(r"\s+", " ", n).strip()
    if len(n) < 3:
        return ""
    return n


def name_similarity(a: str, b: str) -> float:
    """Return a 0..1 similarity ratio between two company names after
    normalization. Uses difflib.SequenceMatcher which is stdlib and close
    enough to Levenshtein ratio for typical corporate-name matching."""
    na = norm_name(a)
    nb = norm_name(b)
    if not na or not nb:
        return 0.0
    return SequenceMatcher(None, na, nb).ratio()


# ----------------------------------------------------------------------
# Combined match-against-existing helper
# ----------------------------------------------------------------------

def build_existing_index(
    existing_rows: Iterable[dict],
    domain_field: str = "domain",
    name_field: str = "name",
    website_field: str | None = "website",
) -> tuple[set[str], dict[str, str]]:
    """Build an apex-domain set + normalized-name index from existing rows.

    Returns:
        (apex_set, name_to_apex):
          apex_set is the set of apex domains present in the existing list.
          name_to_apex maps every normalized company name to the apex it
            came from, so a name-match can report which row it collided with.

    existing_rows can be anything iterable of dicts. Pass rows from your
    CRM export, a previous prospect-list CSV, a customer-list download,
    or whatever the user provides as "do not re-contact".
    """
    apex_set: set[str] = set()
    name_to_apex: dict[str, str] = {}

    for r in existing_rows:
        apex = ""
        # Prefer an explicit domain field, fall back to website.
        for fld in (domain_field, website_field):
            if fld and r.get(fld):
                apex = extract_apex(r[fld])
                if apex:
                    break

        if apex:
            apex_set.add(apex)

        nm = norm_name(r.get(name_field, "")) if name_field else ""
        if nm:
            # Don't overwrite a shorter existing key with a longer one; the
            # first occurrence wins so downstream messages are stable.
            name_to_apex.setdefault(nm, apex or "")

    return apex_set, name_to_apex


def check_duplicate(
    candidate: dict,
    apex_set: set[str],
    name_to_apex: dict[str, str],
    domain_field: str = "domain",
    name_field: str = "name",
    website_field: str | None = "website",
    name_threshold: float = 0.85,
) -> tuple[bool, str]:
    """Check a single candidate row against the existing index.

    Returns (is_duplicate, reason). reason is a short string explaining the
    match (e.g. "apex:nikon.com" or "name:astura medical (0.91)") or "" when
    the candidate is net-new.

    The match is layered:
      1. Apex domain match against apex_set (strong, preferred).
      2. If no domain match, walk name_to_apex looking for a fuzzy match
         above name_threshold. Only used as a fallback because name-only
         matches are noisy (e.g. "rocket propulsion" as a substring).
    """
    # Step A: apex domain
    apex = ""
    for fld in (domain_field, website_field):
        if fld and candidate.get(fld):
            apex = extract_apex(candidate[fld])
            if apex:
                break
    if apex and apex in apex_set:
        return True, f"apex:{apex}"

    # Step B: fuzzy company name
    cand_name = candidate.get(name_field, "") if name_field else ""
    nc = norm_name(cand_name)
    if nc:
        best_key = ""
        best_ratio = 0.0
        for existing_name in name_to_apex:
            ratio = SequenceMatcher(None, nc, existing_name).ratio()
            if ratio > best_ratio:
                best_ratio = ratio
                best_key = existing_name
                if ratio >= 0.99:  # perfect match, stop early
                    break
        if best_ratio >= name_threshold:
            return True, f"name:{best_key} ({best_ratio:.2f})"

    return False, ""


def match_against_existing(
    candidates: Iterable[dict],
    existing: Iterable[dict],
    domain_field: str = "domain",
    name_field: str = "name",
    website_field: str | None = "website",
    name_threshold: float = 0.85,
) -> tuple[list[dict], list[dict]]:
    """Split candidates into (actionable, matched) against an existing list.

    Each row in the output carries a `dedupe_match` field describing why
    it was classified that way — empty for actionable rows, populated with
    the match reason for matched rows. This makes it trivial to surface in
    reports and explain to the user why a row was excluded.
    """
    apex_set, name_to_apex = build_existing_index(
        existing,
        domain_field=domain_field,
        name_field=name_field,
        website_field=website_field,
    )

    actionable: list[dict] = []
    matched: list[dict] = []
    for c in candidates:
        is_dup, reason = check_duplicate(
            c, apex_set, name_to_apex,
            domain_field=domain_field,
            name_field=name_field,
            website_field=website_field,
            name_threshold=name_threshold,
        )
        out = dict(c)
        out["dedupe_match"] = reason
        if is_dup:
            matched.append(out)
        else:
            actionable.append(out)
    return actionable, matched


# ----------------------------------------------------------------------
# Command-line entry point
# ----------------------------------------------------------------------

def _read_csv(path: str) -> list[dict]:
    csv.field_size_limit(sys.maxsize)
    with open(path) as f:
        return list(csv.DictReader(f))


def _write_csv(path: str, rows: list[dict]) -> None:
    if not rows:
        with open(path, "w", newline="") as f:
            f.write("")
        return
    fieldnames = list(rows[0].keys())
    with open(path, "w", newline="") as f:
        w = csv.DictWriter(f, fieldnames=fieldnames)
        w.writeheader()
        w.writerows(rows)


def _selftest() -> int:
    """Sanity-check the apex and name helpers. Exits non-zero on failure."""
    apex_cases = [
        ("amsynergy.nikon.com", "nikon.com"),
        ("industry.nikon.com", "nikon.com"),
        ("nikon.co.jp", "nikon.co.jp"),
        ("www.bbc.co.uk", "bbc.co.uk"),
        ("corporate.arcelormittal.com", "arcelormittal.com"),
        ("blog.company.com", "company.com"),
        ("firehawkaerospace.com", "firehawkaerospace.com"),
        ("https://www.example.com/foo/bar", "example.com"),
        ("", ""),
        ("localhost", ""),
    ]
    name_cases = [
        ("Astura Medical, Inc.", "astura medical"),
        ("MBDA Systems Holdings Ltd", "mbda"),
        ("3DMorphic", "3dmorphic"),
        ("Firehawk Aerospace", "firehawk aerospace"),
    ]

    failures = 0
    for inp, expected in apex_cases:
        got = extract_apex(inp)
        ok = got == expected
        print(f"apex  {'OK ' if ok else 'FAIL'}  {inp!r:<45} -> {got!r}  (expected {expected!r})")
        if not ok:
            failures += 1
    for inp, expected in name_cases:
        got = norm_name(inp)
        ok = got == expected
        print(f"name  {'OK ' if ok else 'FAIL'}  {inp!r:<45} -> {got!r}  (expected {expected!r})")
        if not ok:
            failures += 1

    # Layered-match sanity check
    existing = [{"domain": "nikon.com", "name": "Nikon Corporation"}]
    candidates = [
        {"domain": "amsynergy.nikon.com", "name": "Nikon AM Synergy"},
        {"domain": "ad-astra.com", "name": "Ad Astra Rocket Company"},
        {"domain": "", "name": "Nikon Corp"},  # name-only fallback
    ]
    actionable, matched = match_against_existing(candidates, existing)
    print()
    print("actionable:", actionable)
    print("matched:   ", matched)
    if len(actionable) != 1 or actionable[0]["domain"] != "ad-astra.com":
        print("FAIL: expected only ad-astra.com to be actionable")
        failures += 1

    return 0 if failures == 0 else 1


def _main() -> int:
    parser = argparse.ArgumentParser(
        description="Dedupe a candidate prospect list against an existing list."
    )
    parser.add_argument("--existing", help="CSV with the do-not-contact list")
    parser.add_argument("--candidates", help="CSV with the candidate prospect list")
    parser.add_argument("--out-actionable", help="Write actionable rows here")
    parser.add_argument("--out-matched", help="Write matched-against-existing rows here")
    parser.add_argument("--domain-field", default="domain")
    parser.add_argument("--name-field", default="name")
    parser.add_argument("--website-field", default="website")
    parser.add_argument("--name-threshold", type=float, default=0.85)
    parser.add_argument("--selftest", action="store_true",
                        help="Run built-in sanity tests and exit")
    args = parser.parse_args()

    if args.selftest:
        return _selftest()

    if not (args.existing and args.candidates):
        parser.print_help()
        return 2

    existing = _read_csv(args.existing)
    candidates = _read_csv(args.candidates)
    actionable, matched = match_against_existing(
        candidates, existing,
        domain_field=args.domain_field,
        name_field=args.name_field,
        website_field=args.website_field,
        name_threshold=args.name_threshold,
    )
    if args.out_actionable:
        _write_csv(args.out_actionable, actionable)
    if args.out_matched:
        _write_csv(args.out_matched, matched)
    print(f"existing:   {len(existing)}")
    print(f"candidates: {len(candidates)}")
    print(f"actionable: {len(actionable)}")
    print(f"matched:    {len(matched)}")
    return 0


if __name__ == "__main__":
    sys.exit(_main())

#!/usr/bin/env python3
"""Local shortlist export. Paid legacy contact chain retired; use live Deepline plays."""
import argparse
import csv
import sys
from pathlib import Path

def main():
    parser=argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--input", required=True)
    parser.add_argument("--output", required=True)
    parser.add_argument("--top", type=int, help="Optional maximum; default all input rows")
    mode=parser.add_mutually_exclusive_group()
    mode.add_argument("--contacts", action="store_true")
    mode.add_argument("--no-contacts", action="store_true")
    parser.add_argument("--roles", help="Use with the live persona play, not this exporter")
    args=parser.parse_args()
    if args.contacts:
        parser.error("Legacy paid orchestration retired. Use deepline-gtm to describe/run the live persona and email plays; no calls made.")
    if args.top is not None and args.top < 1:
        parser.error("--top must be positive")
    if Path(args.input).resolve()==Path(args.output).resolve():
        parser.error("Output must not overwrite input")
    with open(args.input, encoding="utf-8-sig", newline="") as f:
        reader=csv.DictReader(f)
        fields=reader.fieldnames
        if not fields or "domain" not in fields or len(set(fields))!=len(fields):
            parser.error("Input needs distinct headers including domain")
        rows=list(reader)
    if any(None in r or any(v is None for v in r.values()) for r in rows):
        parser.error("Malformed CSV")
    selected=rows if args.top is None else rows[:args.top]
    with open(args.output,"w",newline="",encoding="utf-8") as f:
        writer=csv.DictWriter(f,fieldnames=fields)
        writer.writeheader();writer.writerows(selected)
    print(f"Exported {len(selected)} of {len(rows)} rows in input order; all columns retained. No contact discovery.",file=sys.stderr)

if __name__=="__main__": main()

#!/usr/bin/env python3
"""
Find contacts + emails at a list of prospect companies.

This script implements the contact discovery fallback chain required by
Step 7 of the deepline-scoring pipeline. It runs through Deepline in
two phases:

  Phase 1: company-to-contact (FREE tier).
           Dropleads + Deepline native + Icypeas + Prospeo + Crustdata.
           Works well for >200-employee US/EU companies with mature B2B data
           coverage. Returns LinkedIn URLs + titles; often no emails.

  Phase 2: For any company that Phase 1 returned ZERO contacts for (which
           is the common case for <200-employee, non-US, or niche industrial
           targets), fall back to exa_people_search with includeDomains=
           ['linkedin.com']. Exa neural search goes over public web text and
           finds LinkedIn profiles that mention the company by name — far
           better coverage for small companies than the B2B provider
           waterfall. Parse the result titles ("Name | Role at Company") to
           pull named contacts.

  Phase 3: For every named contact we have a LinkedIn URL for (from either
           phase), run name-and-domain-to-email-waterfall to resolve a
           corporate email. Validate the result against the company's apex
           domain — providers sometimes return stale emails from a previous
           employer (e.g. [email protected] when Nick is now at
           X-Bow Systems), and this domain-match validation filters them.

Why this fallback chain exists:
  On the nTop run that motivated this skill improvement, Phase 1 (the
  waterfall) returned ZERO contacts on all 10 top-scoring prospects —
  Plasma Processes, Ad Astra Rocket, Avimetal, Axial3D, CubeLabs, NextAero,
  Camber Spine, American Additive Mfg, 3D-Side, 3di GmbH. These are mostly
  <200-employee industrial companies, many non-US, where B2B waterfall
  providers have thin coverage. Exa people search found 15 real named
  contacts at 6 of the 10 in the same pass. The moral: the waterfall is
  cheaper (free tier) but Exa is the actual discovery engine for niche
  industrial / non-US targets. Always run both.

Usage:

  python3 find_contacts.py \\
      --input prospects.csv \\
      --output contacts.csv \\
      --roles "Design Engineer,Mechanical Engineer,Additive Manufacturing Engineer" \\
      --top 10

Input CSV columns: domain, name, [score], [niche]
Output CSV columns: company, domain, full_name, title, linkedin_url, email,
                    email_source, discovery_phase, score, niche

The script calls `deepline enrich` for each phase — it's a thin wrapper.
Nothing here bypasses Deepline.
"""
from __future__ import annotations

import argparse
import csv
import json
import os
import re
import subprocess
import sys
from typing import Iterable

# Load the apex-domain helper from the sibling dedupe_utils module.
# We add the script's directory to sys.path so imports work from anywhere.
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from dedupe_utils import extract_apex  # noqa: E402


# ----------------------------------------------------------------------
# Small helpers
# ----------------------------------------------------------------------

def _enrich_run_name(output_csv: str) -> str:
    stem = os.path.splitext(os.path.basename(output_csv))[0]
    slug = re.sub(r"[^a-zA-Z0-9]+", "-", stem).strip("-").lower()
    return f"niche-signal-{slug or 'contacts'}"


def _run_deepline_enrich(input_csv: str, output_csv: str, with_specs: list[str]) -> None:
    """Thin wrapper around `deepline enrich`. Raises on non-zero exit."""
    cmd = [
        "deepline", "enrich",
        "--input", input_csv,
        "--output", output_csv,
        "--name", _enrich_run_name(output_csv),
    ]
    for spec in with_specs:
        cmd += ["--with", spec]
    print(f"[find_contacts] running: {' '.join(cmd[:4])} (+{len(with_specs)} --with specs)",
          file=sys.stderr)
    subprocess.run(cmd, check=True)


def _read_csv(path: str) -> list[dict]:
    csv.field_size_limit(sys.maxsize)
    with open(path) as f:
        return list(csv.DictReader(f))


def _write_csv(path: str, rows: list[dict], fieldnames: list[str]) -> None:
    with open(path, "w", newline="") as f:
        w = csv.DictWriter(f, fieldnames=fieldnames)
        w.writeheader()
        w.writerows(rows)


def _parse_json_field(val: str):
    if not val:
        return None
    try:
        return json.loads(val)
    except Exception:
        return None


def _slug_to_name(linkedin_url: str) -> str:
    """Best-effort first/last name extraction from a LinkedIn profile slug.
    LinkedIn slugs look like `/in/first-last-deadbeef` where the trailing
    hash is an account ID. Strip the hash, split on dashes, title-case.
    This is a fallback for when providers don't return full_name."""
    if not linkedin_url:
        return ""
    m = re.search(r"/in/([^/?]+)", linkedin_url)
    if not m:
        return ""
    slug = m.group(1)
    # Strip trailing hash (6+ hex chars) like `-abbbb4172` or `-342152133`
    slug = re.sub(r"-[a-f0-9]{6,}$", "", slug)
    parts = [p for p in slug.split("-") if p and not p.isdigit()]
    return " ".join(p.capitalize() for p in parts[:3])


# ----------------------------------------------------------------------
# Phase 1: company-to-contact
# ----------------------------------------------------------------------

def phase1_waterfall(
    prospects_csv: str,
    out_csv: str,
    roles: list[str],
    seniority: list[str] | None = None,
    limit: int = 3,
) -> list[dict]:
    """Run the FREE company→contact waterfall on every prospect.

    Returns a list of raw contact dicts: one row per contact discovered,
    with fields (company, domain, full_name, title, linkedin_url, discovery_phase).
    Phase 1 rarely returns emails — email resolution is Phase 3.
    """
    seniority = seniority or ["Senior", "Director", "VP", "Manager"]
    spec = json.dumps({
        "alias": "contact",
        "tool": "company-to-contact",
        "payload": {
            "domain": "{{domain}}",
            "company_name": "{{name}}",
            "roles": roles,
            "seniority": seniority,
            "limit": limit,
        },
    })
    _run_deepline_enrich(prospects_csv, out_csv, [spec])

    rows = _read_csv(out_csv)
    contacts: list[dict] = []
    for r in rows:
        parsed = _parse_json_field(r.get("contact", "") or "")
        items: list = []
        if isinstance(parsed, list):
            items = parsed
        elif isinstance(parsed, dict):
            for value in parsed.values():
                if isinstance(value, list):
                    items = value
                    break
                if isinstance(value, dict) and isinstance(value.get("result"), list):
                    items = value["result"]
                    break
        for item in items[:limit]:
            li = item.get("linkedin", "") or item.get("linkedin_url", "") or item.get("profile_url", "") or ""
            full = item.get("full_name") or item.get("name") or _slug_to_name(li)
            contacts.append({
                "company": r.get("name", ""),
                "domain": r.get("domain", ""),
                "full_name": full,
                "title": item.get("title", "") or "",
                "linkedin_url": li,
                "discovery_phase": "waterfall",
            })
    return contacts


# ----------------------------------------------------------------------
# Phase 2: exa_people_search fallback for empty companies
# ----------------------------------------------------------------------

_TITLE_RE = re.compile(
    r"^\s*([A-Z][A-Za-zÀ-ÖØ-öø-ÿ'’.\-]+(?:\s+[A-Z][A-Za-zÀ-ÖØ-öø-ÿ'’.\-]+){1,4})\s*[|\-–]\s*(.+)$"
)

# Generic role tokens used to filter out obvious noise (marketers, recruiters,
# unrelated execs) when the caller hasn't supplied vertical-specific roles. The
# vertical-specific tokens come from the --roles argument at runtime — see
# _derive_role_tokens() — so this set stays neutral.
_GENERIC_ROLE_TOKENS = (
    "engineer", "designer", "principal", "director", "head", "vp",
    "cto", "chief", "manager", "lead", "scientist", "researcher",
)


def _derive_role_tokens(roles: Iterable[str]) -> set[str]:
    """Tokenize the caller's --roles list into a set of single-word filter tokens.

    Splits each role on whitespace, lowercases, and unions with the generic
    role tokens. The point is to let Phase 2's company-match filter accept any
    title that contains a word from the caller's vertical-specific role list,
    without baking those words into a hardcoded constant.
    """
    tokens: set[str] = set(_GENERIC_ROLE_TOKENS)
    for role in roles:
        for word in role.lower().split():
            if len(word) >= 3:  # skip short joiners like "of", "or", "in"
                tokens.add(word)
    return tokens


def phase2_exa_people(
    prospects_csv: str,
    out_csv: str,
    roles: list[str],
    already_covered_domains: set[str],
) -> list[dict]:
    """Run exa_people_search on every prospect that Phase 1 missed.

    The Exa neural search is the workhorse for small / non-US / niche
    industrial companies where the B2B provider waterfall has thin data.
    Include only `linkedin.com` in the domain filter so we get profile
    pages, not marketing copy.

    Only processes companies in prospects_csv that are NOT already in
    `already_covered_domains` (i.e., Phase 1 returned at least one contact
    for them). This keeps credit usage tight.
    """
    # Filter prospects down to the ones that need the fallback.
    all_prospects = _read_csv(prospects_csv)
    needs_fallback = [r for r in all_prospects if r["domain"] not in already_covered_domains]
    if not needs_fallback:
        print("[find_contacts] Phase 1 covered everything, no Phase 2 fallback needed",
              file=sys.stderr)
        return []

    fallback_csv = out_csv.replace(".csv", "_input.csv") if out_csv.endswith(".csv") else out_csv + "_input"
    fieldnames = list(all_prospects[0].keys()) if all_prospects else ["domain", "name"]
    _write_csv(fallback_csv, needs_fallback, fieldnames)

    # Derive the title-filter token set from the caller's --roles list. Tokens
    # are matched against Exa result titles in the per-result loop below — this
    # replaces a hardcoded vertical-specific keyword set so the script stays
    # general across verticals.
    role_tokens = _derive_role_tokens(roles)

    # Build an Exa query phrase from the roles. Keep it short — Exa neural
    # does better with a compact OR'd title list than with a sprawling sentence.
    role_clause = " OR ".join(roles[:6])
    query = f"{role_clause} at {{{{name}}}}"

    spec = json.dumps({
        "alias": "exa_people",
        "tool": "exa_people_search",
        "payload": {
            "query": query,
            "type": "neural",
            "numResults": 10,
            "includeDomains": ["linkedin.com"],
        },
    })
    _run_deepline_enrich(fallback_csv, out_csv, [spec])

    rows = _read_csv(out_csv)
    contacts: list[dict] = []
    for r in rows:
        parsed = _parse_json_field(r.get("exa_people", "") or "")
        results: list = []
        if isinstance(parsed, dict):
            data = parsed.get("result", {}).get("data", {}) if isinstance(parsed.get("result"), dict) else {}
            if isinstance(data, dict):
                results = data.get("results", []) or []

        company_name = (r.get("name", "") or "").lower()
        company_tail = company_name.split(",")[0].split("(")[0].strip()[:8]

        for res in results:
            if not isinstance(res, dict):
                continue
            url = res.get("url", "") or ""
            if "/in/" not in url:
                continue
            title = res.get("title", "") or ""
            text = (res.get("text", "") or "")[:300]
            low_title = title.lower()

            # Require the title to look role-relevant AND to mention the
            # company name somewhere. The company-name requirement is the
            # main false-positive filter — Exa neural sometimes returns
            # profiles at COMPETING companies (e.g. on one real run, a search
            # for "Plasma Processes" engineers returned a Hypertherm plasma
            # process engineer, which is a different employer).
            is_role_relevant = any(tok in low_title for tok in role_tokens)
            is_company_match = (
                company_tail and (company_tail in low_title or company_tail in text.lower())
            )
            if not (is_role_relevant and is_company_match):
                continue

            # Parse "Name | Role at Company" out of the title.
            m = _TITLE_RE.match(title)
            if m:
                name = m.group(1).strip()
                role = m.group(2).strip()
            else:
                name = _slug_to_name(url)
                role = title

            contacts.append({
                "company": r.get("name", ""),
                "domain": r.get("domain", ""),
                "full_name": name,
                "title": role,
                "linkedin_url": url if url.startswith("http") else f"https://www.{url.lstrip('/')}",
                "discovery_phase": "exa_people",
            })
    return contacts


# ----------------------------------------------------------------------
# Phase 3: email waterfall + domain validation
# ----------------------------------------------------------------------

def phase3_emails(contacts: list[dict], out_csv: str) -> list[dict]:
    """Resolve emails for every contact with a LinkedIn URL.

    Uses name-and-domain-to-email-waterfall, which chains pattern validation +
    deepline_native + crustdata + PDL. Then validates the returned email
    against the company's apex domain — providers occasionally return a
    stale email from a previous employer, and domain-mismatch is an
    effective filter for that.

    Returns the same contact list with `email` and `email_source` populated.
    `email` is blank when the resolved address doesn't match the apex
    (still kept in `raw_email` so you can inspect). `email_source` is
    "corporate_validated" for clean matches, "apex_mismatch" when the
    address resolved but looked like a stale/different-employer email,
    or "not_found" when the waterfall returned nothing.
    """
    if not contacts:
        return []

    # Dedupe by LinkedIn URL so we don't pay twice for the same person.
    seen: set[str] = set()
    dedup: list[dict] = []
    for c in contacts:
        li = c.get("linkedin_url", "") or ""
        if not li or li in seen:
            continue
        if not c.get("full_name"):
            continue
        seen.add(li)
        dedup.append(c)

    # Build the Phase 3 input CSV.
    input_csv = out_csv.replace(".csv", "_input.csv") if out_csv.endswith(".csv") else out_csv + "_input"
    email_input_rows = []
    for c in dedup:
        name_parts = c["full_name"].split()
        email_input_rows.append({
            "first_name": name_parts[0] if name_parts else "",
            "last_name": name_parts[-1] if len(name_parts) >= 2 else "",
            "linkedin_url": c["linkedin_url"],
            "company": c.get("company", ""),
            "domain": c.get("domain", ""),
            "title": c.get("title", ""),
        })
    _write_csv(
        input_csv, email_input_rows,
        ["first_name", "last_name", "linkedin_url", "company", "domain", "title"],
    )

    spec = json.dumps({
        "alias": "em",
        "tool": "name-and-domain-to-email-waterfall",
        "payload": {
            "linkedin_url": "{{linkedin_url}}",
            "first_name": "{{first_name}}",
            "last_name": "{{last_name}}",
            "domain": "{{domain}}",
        },
    })
    _run_deepline_enrich(input_csv, out_csv, [spec])

    rows = _read_csv(out_csv)
    results_by_li: dict[str, tuple[str, str]] = {}
    for r in rows:
        em_col = r.get("em", "") or ""
        parsed = _parse_json_field(em_col)
        email = ""
        if isinstance(parsed, str):
            email = parsed.strip()
        elif isinstance(parsed, dict):
            if isinstance(parsed.get("email"), str):
                email = parsed["email"].strip()
            elif isinstance(parsed.get("result"), str):
                email = parsed["result"].strip()
            elif isinstance(parsed.get("result"), dict) and isinstance(
                parsed["result"].get("email"),
                str,
            ):
                email = parsed["result"]["email"].strip()

        li = r.get("linkedin_url", "")
        domain = (r.get("domain", "") or "").lower()
        status = "not_found"
        out_email = ""
        if email:
            em_domain = email.split("@")[-1].lower() if "@" in email else ""
            em_apex = extract_apex(em_domain) if em_domain else ""
            cand_apex = extract_apex(domain)
            if em_apex and cand_apex and em_apex == cand_apex:
                status = "corporate_validated"
                out_email = email
            else:
                status = "apex_mismatch"
                out_email = ""
        results_by_li[li] = (out_email, status)

    enriched: list[dict] = []
    for c in dedup:
        out_email, status = results_by_li.get(c["linkedin_url"], ("", "not_found"))
        enriched.append({
            **c,
            "email": out_email,
            "email_source": status,
        })
    return enriched


# ----------------------------------------------------------------------
# Main entry point
# ----------------------------------------------------------------------

def _main() -> int:
    parser = argparse.ArgumentParser(
        description="Find prospect companies (always) and optionally their contacts + emails. "
                    "Companies-only mode is FREE; contact discovery costs additional Deepline credits.",
    )
    parser.add_argument("--input", required=True,
                        help="CSV of prospects (columns: domain, name, [score], [niche])")
    parser.add_argument("--output", required=True,
                        help="Final CSV. In --no-contacts mode this is the top-N company list. "
                             "In --contacts mode it's one row per discovered contact.")
    parser.add_argument(
        "--roles",
        default="",
        help="Comma-separated role strings to look for (REQUIRED in --contacts mode). "
             "Pass the buyer-persona job titles surfaced in Step 0/0.5 — e.g. for a "
             "creative-ops tool: 'Creative Director,Brand Manager,Content Operations Lead'; "
             "for an AR automation tool: 'AR Manager,Accounts Receivable Specialist,Controller'. "
             "Don't reuse last run's roles for a different vertical.",
    )
    parser.add_argument("--seniority", default="Senior,Director,VP,Manager")
    parser.add_argument("--top", type=int, default=10,
                        help="Only process the top N prospects by score (if present)")
    parser.add_argument("--workdir", default="",
                        help="Directory for intermediate files (default: next to --output)")

    # --contacts / --no-contacts toggle. Default is --no-contacts because contact
    # discovery costs additional Deepline credits and the user should always have
    # to opt in. The SKILL.md flow is: ship companies first, then ask for credit
    # approval before turning on contacts.
    contacts_group = parser.add_mutually_exclusive_group()
    contacts_group.add_argument(
        "--contacts", dest="contacts", action="store_true",
        help="Run the 3-phase contact discovery chain (waterfall + Exa fallback + "
             "email waterfall). Costs extra Deepline credits — get user approval first.",
    )
    contacts_group.add_argument(
        "--no-contacts", dest="contacts", action="store_false",
        help="Companies-only mode (default). Output is just the top-N company list with "
             "score and niche; no contacts, no extra credit spend.",
    )
    parser.set_defaults(contacts=False)
    args = parser.parse_args()

    roles = [r.strip() for r in args.roles.split(",") if r.strip()]
    seniority = [s.strip() for s in args.seniority.split(",") if s.strip()]

    if args.contacts and not roles:
        parser.error(
            "--roles is required in --contacts mode. Pass the buyer-persona job "
            "titles for THIS run's vertical (from Step 0/0.5 ecosystem discovery). "
            "Example: --roles 'Design Engineer,Mechanical Engineer,DfAM Engineer'"
        )

    # Slice to top N if the input has a score column.
    raw = _read_csv(args.input)
    has_score = raw and "score" in raw[0]
    if has_score:
        def _score(r):
            try: return int(r.get("score", "0") or 0)
            except Exception: return 0
        raw = sorted(raw, key=_score, reverse=True)[:args.top]
    else:
        raw = raw[:args.top]

    workdir = args.workdir or os.path.dirname(os.path.abspath(args.output))
    os.makedirs(workdir, exist_ok=True)

    # ----------------------------------------------------------------
    # --no-contacts mode: just write the top-N company list and exit.
    # ----------------------------------------------------------------
    if not args.contacts:
        company_fields = ["domain", "name", "score", "niche"]
        company_rows = [
            {f: r.get(f, "") for f in company_fields}
            for r in raw
        ]
        _write_csv(args.output, company_rows, company_fields)
        print(f"[find_contacts] Wrote top {len(company_rows)} companies to {args.output} "
              f"(--no-contacts mode, no extra credits spent)", file=sys.stderr)
        print(f"[find_contacts] Re-run with --contacts to add contact discovery + emails.",
              file=sys.stderr)
        return 0

    # ----------------------------------------------------------------
    # --contacts mode: stage prospects and run the 3-phase chain.
    # ----------------------------------------------------------------
    top_csv = os.path.join(workdir, "_top.csv")
    fieldnames = list(raw[0].keys()) if raw else ["domain", "name"]
    _write_csv(top_csv, raw, fieldnames)

    # ---- Phase 1 ----
    phase1_out = os.path.join(workdir, "_phase1_waterfall.csv")
    phase1_contacts = phase1_waterfall(top_csv, phase1_out, roles=roles, seniority=seniority)
    covered = {c["domain"] for c in phase1_contacts if c.get("full_name") and c.get("linkedin_url")}
    print(f"[find_contacts] Phase 1 found {len(phase1_contacts)} contacts "
          f"covering {len(covered)}/{len(raw)} companies", file=sys.stderr)

    # ---- Phase 2 ----
    phase2_out = os.path.join(workdir, "_phase2_exa_people.csv")
    phase2_contacts = phase2_exa_people(top_csv, phase2_out, roles=roles,
                                        already_covered_domains=covered)
    print(f"[find_contacts] Phase 2 found {len(phase2_contacts)} additional contacts",
          file=sys.stderr)

    # ---- Phase 3 ----
    all_contacts = phase1_contacts + phase2_contacts
    phase3_out = os.path.join(workdir, "_phase3_emails.csv")
    with_emails = phase3_emails(all_contacts, phase3_out)

    # Merge score/niche back onto the final output if the input had them.
    score_by_domain = {r["domain"]: r.get("score", "") for r in raw}
    niche_by_domain = {r["domain"]: r.get("niche", "") for r in raw}
    for c in with_emails:
        c["score"] = score_by_domain.get(c["domain"], "")
        c["niche"] = niche_by_domain.get(c["domain"], "")

    # Final write.
    fields = ["company", "domain", "score", "niche", "full_name", "title",
              "linkedin_url", "email", "email_source", "discovery_phase"]
    _write_csv(args.output, with_emails, fields)

    valid = sum(1 for c in with_emails if c["email_source"] == "corporate_validated")
    print(f"[find_contacts] Wrote {len(with_emails)} contacts to {args.output}",
          file=sys.stderr)
    print(f"[find_contacts] {valid} have corporate-validated emails", file=sys.stderr)
    return 0


if __name__ == "__main__":
    sys.exit(_main())

import { definePlay } from 'deepline';
import { scoreRow, artifactKey } from './scoring_contract';
import type { Model, Reference, Observation } from './scoring_contract';

type Input = {
  domain: string;
  scored_at: string;
  snapshot_id: string;
  rows: Array<{
    domain: string;
    entity_id: string;
    observations: Observation[];
  }>;
  model: Model;
  reference: Reference;
};

/** Replay only. Source collection and model fitting are deliberately not fabricated. */
export default definePlay(
  'replay-account-score',
  async (ctx, input: Input) => {
    if (!input.snapshot_id) throw new Error('snapshot_id_required');
    if (
      typeof input.domain !== 'string' ||
      !Array.isArray(input.rows) ||
      input.rows.some((row) => typeof row.domain !== 'string')
    )
      throw new Error('invalid_identity_input');
    const domain = input.domain
      .trim()
      .toLowerCase()
      .replace(/^www\./, '');
    if (
      !/^[a-z0-9](?:[a-z0-9.-]*[a-z0-9])?\.[a-z]{2,}$/.test(domain) ||
      domain.includes('..')
    )
      throw new Error('domain_required');
    const matches = input.rows.filter(
      (row) => row.domain.toLowerCase().replace(/^www\./, '') === domain,
    );
    const result =
      matches.length !== 1
        ? {
            ...scoreRow(
              'unresolved',
              input.scored_at,
              [],
              input.model,
              input.reference,
            ),
            entity_id: null,
            status: matches.length ? 'ambiguous_identity' : 'not_in_snapshot',
            miss_reason: matches.length
              ? 'ambiguous_identity'
              : 'not_in_snapshot',
            candidates: matches.map((row) => row.entity_id),
            score: null,
            grade: null,
          }
        : scoreRow(
            matches[0].entity_id,
            input.scored_at,
            matches[0].observations,
            input.model,
            input.reference,
          );
    const replay_key = artifactKey([
      domain,
      input.snapshot_id,
      input.model,
      input.reference,
      input.scored_at,
      matches,
    ]);
    const rows = await ctx
      .dataset('scored_rows', [
        { domain, replay_key, snapshot_id: input.snapshot_id, ...result },
      ])
      .run({ key: 'replay_key' });
    return { rows, delivery: 'replay_only' };
  },
  { description: 'Replay a frozen account score' },
);

/** Pure replay scorer. No providers, default weights, or claims of validation. */
export function artifactKey(value: unknown): string {
  // Exact-content key; no runtime-specific crypto dependency or hash collisions.
  // Production adapters may replace this with a supported cryptographic content hash.
  return JSON.stringify(value);
}
export type Dimension =
  | 'account_fit'
  | 'account_engagement'
  | 'lead_fit'
  | 'lead_engagement';
export type Observation = {
  entity_id: string;
  feature: string;
  value: number | null;
  dimension: Dimension;
  source_class: 'external' | 'first_party_event' | 'ae';
  source_id: string;
  known_at: string;
  event_at: string;
  retrieved_at: string;
  status: 'observed' | 'missing' | 'error';
};
export type Model = {
  id: string;
  dimension: Dimension;
  intercept: number;
  weights: Record<string, number>;
  max_age_days: Record<string, number>;
  event_window_days?: number;
  validation: 'exploratory' | 'validated';
};
export type Reference = {
  id: string;
  model_id: string;
  dimension: Dimension;
  scores: number[];
  population: string;
  frozen_at: string;
};

function instant(value: string): number {
  if (typeof value !== 'string' || !/(Z|[+-]\d\d:\d\d)$/.test(value))
    throw new Error('timezone_required');
  const parts =
    /^(\d{4})-(\d{2})-(\d{2})T(\d{2}):(\d{2}):(\d{2})(?:\.\d{1,3})?(Z|[+-]\d{2}:\d{2})$/.exec(
      value,
    );
  if (!parts) throw new Error('invalid_timestamp');
  const [, year, month, day, hour, minute, second, zone] = parts;
  const days = new Date(Date.UTC(Number(year), Number(month), 0)).getUTCDate();
  if (
    +year < 100 ||
    +month < 1 ||
    +month > 12 ||
    +day < 1 ||
    +day > days ||
    +hour > 23 ||
    +minute > 59 ||
    +second > 59 ||
    (zone !== 'Z' && (+zone.slice(1, 3) > 23 || +zone.slice(4, 6) > 59))
  )
    throw new Error('invalid_timestamp');
  const n = Date.parse(value);
  if (!Number.isFinite(n)) throw new Error('invalid_timestamp');
  return n;
}
export function gradePercentile(percentile: number): 'A' | 'B' | 'C' | 'D' {
  if (!Number.isFinite(percentile) || percentile < 0 || percentile > 100)
    throw new Error('invalid_percentile');
  return percentile >= 90
    ? 'A'
    : percentile >= 75
      ? 'B'
      : percentile >= 50
        ? 'C'
        : 'D';
}
export function gradeScore(score: number | null, reference: Reference) {
  if (!reference.id || !reference.model_id || !reference.population)
    throw new Error('invalid_reference');
  instant(reference.frozen_at);
  if (
    !reference.scores.length ||
    reference.scores.some((x) => !Number.isFinite(x))
  )
    throw new Error('invalid_reference_scores');
  if (score === null)
    return { percentile: null, grade: null, reason: 'unscored' };
  if (!Number.isFinite(score)) throw new Error('invalid_score');
  if (new Set(reference.scores).size < 2)
    return {
      percentile: null,
      grade: null,
      reason: 'insufficient_reference_variation',
    };
  const less = reference.scores.filter((x) => x < score).length;
  const equal = reference.scores.filter((x) => x === score).length;
  const percentile = (100 * (less + equal / 2)) / reference.scores.length;
  return { percentile, grade: gradePercentile(percentile), reason: null };
}
export function scoreRow(
  entityId: string,
  scoredAt: string,
  observations: Observation[],
  model: Model,
  reference: Reference,
) {
  const cutoff = instant(scoredAt);
  if (!['exploratory', 'validated'].includes(model.validation))
    throw new Error('invalid_validation');
  if (
    !entityId ||
    !model.id ||
    ![
      'account_fit',
      'account_engagement',
      'lead_fit',
      'lead_engagement',
    ].includes(model.dimension)
  )
    throw new Error('invalid_model_or_entity');
  if (
    reference.model_id !== model.id ||
    reference.dimension !== model.dimension
  )
    throw new Error('reference_model_mismatch');
  if (instant(reference.frozen_at) > cutoff)
    throw new Error('reference_not_frozen_at_scoring');
  if (
    !Number.isFinite(model.intercept) ||
    !Object.keys(model.weights).length ||
    Object.values(model.weights).some((x) => !Number.isFinite(x))
  )
    throw new Error('invalid_weights');
  const fit = model.dimension.endsWith('_fit');
  if (
    !fit &&
    (!Number.isFinite(model.event_window_days) || model.event_window_days! <= 0)
  )
    throw new Error('engagement_window_required');
  const reasons: string[] = [],
    evidence: Observation[] = [],
    enriched: Record<string, number> = {};
  let raw = model.intercept;
  for (const [feature, weight] of Object.entries(model.weights)) {
    const age = model.max_age_days[feature];
    if (!Number.isFinite(age) || age < 0)
      throw new Error('feature_freshness_required');
    const matching = observations.filter(
      (o) =>
        o.feature === feature &&
        o.entity_id === entityId &&
        o.dimension === model.dimension,
    );
    const rejected = new Set<string>();
    const reject = (reason: string) => {
      rejected.add(reason);
      return false;
    };
    const eligible = matching.filter((o) => {
      if (
        o.status !== 'observed' ||
        o.value === null ||
        !Number.isFinite(o.value) ||
        !o.source_id
      )
        return reject(
          o.status === 'error' ? 'provider_error' : 'missing_value_or_source',
        );
      if (o.source_class !== (fit ? 'external' : 'first_party_event'))
        return reject(
          o.source_class === 'ae' ? 'ae_source_excluded' : 'wrong_source_class',
        );
      try {
        const known = instant(o.known_at),
          event = instant(o.event_at),
          retrieved = instant(o.retrieved_at);
        if (known >= cutoff || event >= cutoff || retrieved >= cutoff)
          return reject('not_available_before_cutoff');
        if (event > known || known > retrieved)
          return reject('invalid_chronology');
        if (
          cutoff - event >
          Math.min(age, fit ? age : model.event_window_days!) * 86400000
        )
          return reject('stale_event');
        return true;
      } catch {
        return reject('invalid_timestamp');
      }
    });
    // Explicitly require upstream source conflict resolution; don't pick whichever scores best.
    if (eligible.length !== 1) {
      reasons.push(
        `${feature}:${eligible.length ? 'ambiguous_observation' : [...rejected].sort().join(',') || 'no_matching_observation'}`,
      );
      continue;
    }
    const o = eligible[0];
    enriched[feature] = o.value!;
    raw += weight * o.value!;
    evidence.push(o);
  }
  const score = reasons.length ? null : raw;
  if (score !== null && !Number.isFinite(score))
    throw new Error('score_overflow');
  const graded = gradeScore(score, reference);
  return {
    entity_id: entityId,
    scored_at: scoredAt,
    dimension: model.dimension,
    enriched,
    score,
    ...graded,
    model_id: model.id,
    reference_id: reference.id,
    model_key: artifactKey(model),
    reference_key: artifactKey(reference),
    reference_n: reference.scores.length,
    tie_policy: 'exact_numeric_midrank',
    coverage: evidence.length / Object.keys(model.weights).length,
    confidence: null,
    out_of_reference_range:
      score === null
        ? null
        : reference.scores.every((x) => x > score) ||
          reference.scores.every((x) => x < score),
    validation: model.validation,
    status:
      score === null
        ? 'needs_evidence'
        : graded.grade === null
          ? 'needs_reference'
          : 'scored',
    miss_reason: reasons.length ? reasons.join(';') : graded.reason,
    evidence,
  };
}

"""Offline observation gate. No crawling, signature discovery or use confirmation.

Usage: python3 technology_evidence.py normalized-observations.json
Input is a JSON list conforming to references/technology-evidence.md.
Output deliberately omits raw URLs, snippets and unrecognized fields.
"""
import ipaddress
import json
import re
import sys
from datetime import datetime
from urllib.parse import urlsplit


def host(url):
    if not isinstance(url, str) or re.search(r"[\s\\]", url):
        raise ValueError("Invalid public URL")
    p = urlsplit(url)
    if p.scheme not in ("https", "http") or p.username or p.password:
        raise ValueError("Only public HTTP(S) URLs without userinfo")
    h = (p.hostname or "").rstrip(".").lower()
    # Deliberately conservative. This is not a network/SSRF authorization gate.
    if not re.fullmatch(r"[a-z0-9](?:[a-z0-9.-]*[a-z0-9])?", h) or "." not in h:
        raise ValueError("Invalid hostname")
    if any(not label or label.startswith("-") or label.endswith("-") for label in h.split(".")):
        raise ValueError("Invalid hostname")
    try:
        ipaddress.ip_address(h)
    except ValueError:
        return h
    raise ValueError("IP addresses are not vendor domains")


def identifier(value):
    if not isinstance(value, str) or not re.fullmatch(r"[A-Za-z0-9_.:-]{1,160}", value):
        raise ValueError("Expected an opaque identifier")
    return value


def classify(row):
    if not isinstance(row, dict):
        raise ValueError("Expected observation object")
    account = identifier(row["account_id"])
    component = identifier(row["component_id"])
    stamp = datetime.fromisoformat(row["observed_at"].replace("Z", "+00:00"))
    if stamp.tzinfo is None:
        raise ValueError("Timestamp must have timezone")
    page_host = host(row["page_url"])
    kind = row["kind"]
    if kind not in {"mention", "public_link", "embedded", "snippet", "network"}:
        raise ValueError("Unsupported observation kind")
    signature = identifier(row["signature_id"])
    signature_host = host(row["signature_source"])
    level = {"mention": "mention_only", "snippet": "snippet_candidate"}.get(kind)
    resource_host = None
    if level is None:
        resource_host = host(row["resource_url"])
        vendor_domain = row["vendor_domain"]
        if not isinstance(vendor_domain, str) or host("https://" + vendor_domain) != vendor_domain:
            raise ValueError("Expected canonical lowercase vendor domain")
        matches = resource_host == vendor_domain or resource_host.endswith("." + vendor_domain)
        if not matches:
            level = "unmatched_domain"
        elif kind == "network":
            status = row.get("status")
            if status is not None and (type(status) is not int or not 100 <= status <= 599):
                raise ValueError("Invalid HTTP status")
            level = "requested" if status is None else (
                "response_observed" if 200 <= status < 300 else "failed_request"
            )
            # Redirects/cache responses need their own contextual review.
        else:
            level = "public_link" if kind == "public_link" else "embedded_reference"
    return {
        "account_id": account, "component_id": component,
        "observed_at": stamp.isoformat(), "page_host": page_host,
        "signature_id": signature, "signature_source_host": signature_host,
        "resource_host": resource_host, "level": level,
        "scoring_eligible": False, "contract_verified": False,
    }


def main():
    with open(sys.argv[1], encoding="utf-8") as f:
        rows = json.load(f)
    if not isinstance(rows, list):
        raise ValueError("Input must be a list")
    result = [classify(row) for row in rows]  # Validate everything before printing.
    print(json.dumps(result, indent=2))


if __name__ == "__main__":
    main()

{
  "documents": {
    "references/buyer-language-research.md": {
      "kind": "guide",
      "title": "Research Buyer Language Across Public Sources",
      "tags": [
        "signals",
        "research",
        "keywords"
      ],
      "providers": []
    },
    "references/testing-and-evaluation.md": {
      "kind": "guide",
      "title": "Test Implementation and Evaluate Scoring",
      "tags": [
        "scoring",
        "testing",
        "evaluation"
      ],
      "providers": []
    },
    "references/scoring-delivery.md": {
      "kind": "guide",
      "title": "Reproducible Scoring Delivery Contract",
      "tags": [
        "scoring",
        "plays",
        "evaluation"
      ],
      "providers": []
    },
    "SKILL.md": {
      "kind": "entrypoint",
      "title": "Deepline Scoring",
      "tags": [
        "signals",
        "icp"
      ],
      "providers": []
    },
    "references/keyword-catalog.md": {
      "kind": "guide",
      "title": "Keyword Catalog",
      "tags": [
        "signals"
      ],
      "providers": []
    },
    "references/report-template.md": {
      "kind": "guide",
      "title": "Report Template",
      "tags": [
        "reporting"
      ],
      "providers": []
    },
    "references/signal-interpretation.md": {
      "kind": "guide",
      "title": "Signal Interpretation",
      "tags": [
        "signals"
      ],
      "providers": []
    },
    "references/dedupe.md": {
      "kind": "guide",
      "title": "Dedupe Against Existing List",
      "tags": [
        "prospecting",
        "dedupe"
      ],
      "providers": []
    },
    "references/quality-gate.md": {
      "kind": "guide",
      "title": "Quality Gate",
      "tags": [
        "enrichment",
        "qa"
      ],
      "providers": []
    },
    "references/pitfalls.md": {
      "kind": "guide",
      "title": "Common Pitfalls",
      "tags": [
        "signals",
        "troubleshooting"
      ],
      "providers": []
    },
    "references/proven-signals.md": {
      "kind": "guide",
      "title": "Hypothesis Library (Not Validated Weights)",
      "tags": [
        "signals",
        "scoring"
      ],
      "providers": []
    },
    "references/step-7-prospects.md": {
      "kind": "guide",
      "title": "Optional Prospect Handoff",
      "tags": [
        "prospecting",
        "contacts"
      ],
      "providers": []
    },
    "references/scoring-pitfalls.md": {
      "kind": "guide",
      "title": "Scoring Pitfalls \u2014 Confirmation-Biased Fields",
      "tags": [
        "signals",
        "scoring"
      ],
      "providers": []
    },
    "references/capacity-evidence.md": {
      "kind": "guide",
      "title": "Staff and Operational Volume: Measurement Boundaries",
      "tags": [
        "signals",
        "measurement"
      ],
      "providers": []
    },
    "references/technology-evidence.md": {
      "kind": "guide",
      "title": "Public Technology Evidence",
      "tags": [
        "signals",
        "technology"
      ],
      "providers": []
    },
    "references/scorecard-creation.md": {
      "kind": "guide",
      "title": "Scorecard Creation Pattern",
      "tags": [
        "scoring",
        "implementation"
      ],
      "providers": []
    },
    "references/scoring-diagnostics.md": {
      "kind": "guide",
      "title": "Scoring Diagnostics",
      "tags": [
        "scoring",
        "troubleshooting"
      ],
      "providers": []
    }
  },
  "version": "2.2"
}

---
name: deepline-scoring
disable-model-invocation: false
description: 'Use when discovering niche signals, auditing ICP or won/lost evidence, rescoring accounts, or building account and lead scoring Plays. Triggers on fit scoring, engagement scoring, external proxies, and scoring leakage. Skip pure outreach copy or contributor skill installation tasks.'
---

# Deepline Scoring

## Quick Start

```bash
npm install -g deepline
# Fallback for secure sandboxes: mkdir -p "$HOME/.local" && npm config set prefix "$HOME/.local" && export PATH="$HOME/.local/bin:$PATH" && npm install -g deepline --registry https://code.deepline.com/api/v2/npm/
deepline auth register --wait auto
deepline auth wait --timeout 120 # completes Cowork/browser approval; no-op if already connected
deepline auth status
deepline -h
```

Find evidence for the customer's decision. Use approved rules or a separately evaluated model for scoring. Phrase matches and prevalence ratios alone cannot supply scoring weights.

| Task                            | Read                                                                                                               |
| ------------------------------- | ------------------------------------------------------------------------------------------------------------------ |
| Find phrases and buyer language | [Keyword catalog](references/keyword-catalog.md), [buyer-language research](references/buyer-language-research.md) |
| Build or audit a score          | [Scoring delivery](references/scoring-delivery.md)                                                                 |
| Test rules or evaluate outcomes | [Testing and evaluation](references/testing-and-evaluation.md)                                                     |
| Create an artifact-backed scorecard | [Scorecard creation pattern](references/scorecard-creation.md) |
| Debug disputed features or routing | [Scoring diagnostics](references/scoring-diagnostics.md) |
| Verify technology               | [Technology evidence](references/technology-evidence.md)                                                           |
| Estimate staffing or demand     | [Capacity evidence](references/capacity-evidence.md)                                                               |

Read `deepline-gtm` before collection and `deepline-plays` before authoring. Verify the workspace, current provider schema and price. Pilot one or two rows and pass the [quality gate](references/quality-gate.md) on the generated outputs before scaling within the approved budget. Keep exports and receipts in a persistent project directory. Reuse collected evidence when rescoring.

## Workflow

1. Define the product, decision date, prediction horizon, population, analysis unit and requested outputs. Before paid collection, identify the executable model and independent expected scores for parity, or historical evidence and untouched labels for predictive evaluation. Missing prerequisites leave that test blocked; an authorized research run can still proceed. Keep fit, engagement, capacity and coverage separate. Use only requested dimensions: `account_fit`, `account_engagement`, `lead_fit`, `lead_engagement`. A combined priority policy must preserve its components.
2. Resolve identities, parent groups and conflicting outcomes. Split discovery and validation by time and parent before selecting features. Preserve the full requested population. Open accounts and random alternatives have unknown outcomes; lookalikes are not wins. Separate acquisition, renewal and expansion.
3. Research workflows, problems, roles, systems and counterexamples across relevant public sources. Follow observed buyer phrases and source URLs. Mine discovery documents without labels, then review concepts and aliases against their evidence. Keep rare and inconclusive candidates. Exclude report prose and outcome summaries from the corpus.
4. Check collection with the [quality gate](references/quality-gate.md). Report coverage by source and outcome before citing lift. Compare a coverage-only model and evaluate features where both outcomes have adequate observed data. Imputation or dropping missingness flags can still encode collection bias. Keep failed, partial and empty results distinct.
5. Check what each feature measures. Reviews are not calls, openings are not hires, and software mentions or portal links do not prove installation. Keep Google reviews, Yelp, traffic estimates and sitemap counts separate. Validate proxies against actual measurements. Exclude AE discovery, opportunity and outcome fields from pre-contact fit. Require evidence that every input was available at the decision date; current enrichment cannot validate past predictions.
6. Freeze extraction rules, aliases, model and reference artifacts before validation. Fit selection, imputation and tuning within training folds. Follow [scoring pitfalls](references/scoring-pitfalls.md) and [signal interpretation](references/signal-interpretation.md). Record every attempted model and failed comparison. Reusing a holdout to refine rules consumes it.
7. For scoring, deliver a checked Play that resolves the identifier, enriches the row and returns the requested outputs. Follow [scoring delivery](references/scoring-delivery.md) and finish with [testing and evaluation](references/testing-and-evaluation.md). Prove existing-output parity separately from predictive usefulness. Keep missing rows unscored with reasons. A cached replay is partial delivery for a live-enrichment request.
8. Deliver one readable report per workspace using the [report template](references/report-template.md). Combine targeting findings, ranked accounts, scoring rules, the runnable Play and evaluation results in that report; link complete tables and raw evidence at the end. Name the supported state: `research_only`, `replay_only`, `exploratory_end_to_end` or `validated_for_named_use_case`. Promotion requires untouched evaluation against the existing rules and a simple baseline, uncertainty estimates and a stated business acceptance threshold. Selected wins or percentile quotas cannot establish usefulness.

## Inputs and commands

CSV columns: `domain,status,website,jobs`; optional `account_id,parent_id,split,known_at,scored_at`. Status: `won|lost|lookalike|unlabeled`. Merge repeated observations while retaining their sources. Resolve conflicting labels through an explicit cohort rule; never discard every duplicate domain.

Use the opt-in `*_v2.py` helpers for new work. Unversioned scripts retain their existing interfaces. Update consumers explicitly before switching versions.

```bash
python3 scripts/analyze_signals_v2.py --input accounts.csv --discover-phrases \
  --max-phrases 500 --min-phrase-accounts 2 --output candidates.json

python3 scripts/analyze_signals_v2.py --input accounts.csv \
  --keywords keywords.json --tools tools.json --job-roles roles.json \
  --partition discovery --evidence-limit 12 --output discovery.json
```

The miner uses 2–5-word n-grams from discovery accounts, with source and phrase-length diversity. It ignores labels and validation documents. It does not understand negation or generate synonyms. Review matches, nonmatches, buyer/seller context and wrong-company text. Use `--min-phrase-accounts 1` for a labeled rare-phrase pass. Report truncation and increase the cap when needed.

Configs map categories to lists of strings or `{"name":"concept","aliases":["phrase","explicit stem*"]}`. Strings match exact phrase boundaries. Roles match job titles; use keyword concepts for duties.

Validation requires `--partition validation --manifest frozen.json`. The manifest contains `analyzer_sha256`, timezone-aware `frozen_at`, and `config_sha256` hashes for `keywords,tools,job_roles`. Freeze before scoring. Each row requires `known_at <= scored_at`, with `known_at` reflecting the latest availability of all included sources. The script cannot verify timestamp provenance. Scoring before contact uses the stricter cutoff in the delivery contract.

V2 returns `signals[]`, `statistics`, `method`, `config_sha256` and `scoring_eligible:false`. Its metric is Jeffreys-smoothed P(feature|won)/P(feature|lost), not win-rate lift. Wilson intervals describe prevalence; Fisher p-values and BH/BY q-values cover the configured tests in that invocation. They assume independent accounts and do not correct for parent clustering, selection bias or undisclosed repeated experiments. Outputs remain exploratory, including validation-partition runs.

## Source adapters

- Websites: `{"data":{"results":[{"url":"...","title":"...","text":"..."}]}}`, `pages`, direct text/markdown records and known `toolResponse.rawV2/raw` wrappers.
- Jobs: `{"result":{"listings":[{"title":"...","description":"...","url":"..."}]}}`, `jobs[].job_details`, `job_listings` and JSON:API `data[].attributes`.
- Preserve source IDs and dates. Add a fixture and adapter for each new shape. Unsupported or malformed payloads fail.
- An empty jobs array means no returned records for that query; retain its filters and limits. An empty website scrape means unknown coverage. Neither establishes staffing, current vacancies or business-trait absence.
- Named `website/jobs` headers auto-detect. Legacy positional columns require explicit indices. Duplicate headers, invalid labels and unresolved mixed outcomes fail.

## Verification

Follow [testing and evaluation](references/testing-and-evaluation.md) for the acceptance contract. The executable evaluation suite is delivered separately from this skill change. Keep implementation replay, live enrichment and predictive validation distinct, and report missing prerequisites.

Keep customer rules, cases and receipts outside published skills. Treat [proven signals](references/proven-signals.md) as hypotheses. For optional contacts, use the current GTM workflow and [dedupe](references/dedupe.md)/[prospecting](references/step-7-prospects.md) guidance; the local contact helper only exports a shortlist. Do not infer names from LinkedIn slugs or treat domain matching as email verification.