BizPipeline — RevOps End-to-End

1# Full pipeline — scraping → enrichment → scoring → email
2async def run_pipeline(config: PipelineConfig) -> PipelineRun:
3    run = PipelineRun.create(config)
4
5    # Step 1 — Multi-source scraping (parallel)
6    raw_leads = await asyncio.gather(
7        scrape_linkedin(config.filters),
8        scrape_pappers(config.filters),
9        scrape_societe_com(config.filters),
10    )
11    run.update(stage="scraping", count=len(raw_leads))
12
13    # Step 2 — Deduplicate (fuzzy matching on company name)
14    unique_leads = dedup_fuzzy(raw_leads, threshold=0.85)
15
16    # Step 3 — AI enrichment (batch to avoid rate limits)
17    enriched = await enrich_batch(unique_leads, batch_size=20)
18
19    # Step 4 — Scoring against historical deals
20    scored = score_leads(enriched, model=config.scoring_model)
21    hot_leads = [l for l in scored if l.score >= config.threshold]
22
23    # Step 5 — Launch email sequence
24    for lead in hot_leads:
25        await email_sequencer.enqueue(lead, sequence=config.sequence)
26
27    run.finalize(output=hot_leads)
28    return run

1# Multi-touch attribution — links revenue to pipeline touchpoints
2from dataclasses import dataclass
3from enum import Enum
4
5class Model(Enum):
6    LINEAR        = "linear"       # equal credit each touch
7    TIME_DECAY    = "time_decay"   # more credit to recent touches
8    FIRST_TOUCH   = "first_touch"
9    LAST_TOUCH    = "last_touch"
10
11def attribute_revenue(deal: Deal, model: Model) -> list[Attribution]:
12    touches = db.touchpoints.for_lead(deal.lead_id)
13    revenue  = deal.amount
14
15    if model == Model.LINEAR:
16        weight = 1 / len(touches)
17        return [Attribution(t, revenue * weight) for t in touches]
18
19    if model == Model.TIME_DECAY:
20        # Half-life of 7 days — recent touches get more credit
21        now   = deal.closed_at
22        decay = [0.5 ** ((now - t.created_at).days / 7) for t in touches]
23        total = sum(decay)
24        return [Attribution(t, revenue * d / total) for t, d in zip(touches, decay)]
25
26    # first / last touch
27    idx = 0 if model == Model.FIRST_TOUCH else -1
28    return [Attribution(touches[idx], revenue)]