#!/usr/bin/env python3 """ Daily Outreach Script — Curious Endeavor Signal-enriched, tier-prioritised LinkedIn contact surfacing. Posts 5 contacts to #outreach each morning with signals + approach. Tiers (in order): 1. Lukas High Priority (enriched, CE pitches written) 2. Lukas Medium Priority 3. Overlap contacts (both Lukas + Assaf) → warm intro flag 4. Assaf unique contacts (ICP-scored, network density) v2: Company fit gate added (2026-03-23) Before surfacing any contact, research their company to check: - NOT an agency / creative studio / design firm (peer, not buyer) - NOT post-acquisition (budget frozen, decisions up the chain) - NOT in a recent rebrand (12-month cool-off — already bought) - HAS a real public presence (ghost companies = skip) - Fits CE's scale target (not a 1-2 person boutique) Contacts that fail the gate are skipped + logged to DISQUALIFIED_LOG. """ import csv import json import os import sys import subprocess import time import urllib.request import urllib.error from datetime import datetime, timezone from pathlib import Path # ── Paths & Config ──────────────────────────────────────────────────────────── WORKSPACE = Path("/root/.openclaw/workspace") ASSAF_CSV = WORKSPACE / "work/internal-ce/operations/linkedin/Linkedin part1/Connections.csv" SENT_LOG = WORKSPACE / "work/internal-ce/operations/linkedin/sent-contacts.json" OUTREACH_CHANNEL = "1483736700969816165" CRM_SHEET_ID = "1P39N6Tt_OeMWP0q2MMxooJOhm3E8g5pzT085XyK7X4w" LUKAS_SHEET_ID = "10f5TSoBcqXrgiGXYIQQAZ5-XN27pGiTZKo16T0d9Wrs" BATCH_SIZE = 5 # contacts per owner per day (5 Lukas + 5 Assaf = 10 total) SIGNAL_POOL = 10 # qualified candidates to signal-scan per owner SEARCH_DELAY = 0.8 # seconds between web searches (rate limit courtesy) # ── Company Fit Gate ────────────────────────────────────────────────────────── # Companies that run creative/brand/design work themselves = peers, not buyers. # Contacting them as if they're a prospect is a waste and looks naive. PEER_COMPANY_SIGNALS = [ "agency", "creative agency", "design agency", "brand agency", "marketing agency", "studio", "design studio", "creative studio", "brand studio", "motion studio", "ad agency", "advertising agency", "digital agency", "communications agency", "consultancy", "branding consultancy", "strategy consultancy", "production company", "production house", "content studio", "freelance", "self-employed", "independent", ] # Exceptions: agencies at scale that BUY external brand work for their clients AGENCY_EXCEPTIONS = [ "holding company", "wpp", "publicis", "omnicom", "ipr", "interpublic", "dentsu", "havas", "accenture song", "deloitte digital", "bcg platinion", ] # Post-acquisition signals in company name or job title ACQUISITION_SIGNALS = [ "acquired by", "a [company] company", "part of ", "now part of", "a division of", "business unit", ] # Company name patterns that suggest very small boutique (1-3 person shops) BOUTIQUE_PATTERNS = [ "by [name]", "studio [name]", "[name] studio", "[name] creative", "[name] design", ] DISQUALIFIED_LOG = Path("/root/.openclaw/workspace/work/internal-ce/operations/linkedin/disqualified-companies.json") # ── ICP Scoring ─────────────────────────────────────────────────────────────── # Tier A: Direct marketing buyers — always qualify regardless of company MARKETING_TITLES = [ "cmo", "chief marketing", "vp marketing", "svp marketing", "head of marketing", "director of marketing", "marketing director", "marketing lead", "marketing manager", "head of brand", "brand director", "brand lead", "brand strategy", "brand manager", "chief brand", "vp brand", "head of campaigns", "campaign director", "campaign manager", "head of communications", "communications director", "vp communications", "chief communications", "chief creative", "creative director", "vp design", "head of design", "design director", "head of content", "content director", "brand experience", "vp campaigns", "head of growth", "vp growth", "growth director", "chief growth", ] # Tier B: Founders/CEOs qualify ONLY if company is brand-relevant # (consumer, scale-up with market presence, creative/agency, media, consumer tech) FOUNDER_TITLES = [ "founder", "co-founder", "ceo", "chief executive", "managing director", ] BRAND_RELEVANT_COMPANY_SIGNALS = [ # Industries where brand is central to the business "consumer", "retail", "fashion", "food", "beverage", "fmcg", "media", "entertainment", "hospitality", "travel", "health", "wellness", "beauty", "sport", "gaming", "e-commerce", "ecommerce", "marketplace", # Agency / creative / marketing services "agency", "studio", "creative", "design", "brand", "marketing", "content", "advertising", "communications", "pr ", " pr", "production", # Fintech / scale-ups with consumer surface area "fintech", "neobank", "payments", "insurtech", # Larger companies (not micro-SaaS) "ventures", "capital", "group", "global", "international", ] BRAND_IRRELEVANT_COMPANY_SIGNALS = [ # B2B micro-SaaS / internal tooling — founders here are NOT CE's buyer "workflow", "automation", "knowledge management", "hr tech", "hrtech", "payroll", "erp", "crm software", "data management", "compliance", "cybersecurity", "infrastructure", "devops", "monitoring", "logging", ] # Companies in these spaces disqualify the contact regardless of their title SKIP_COMPANY_SIGNALS = [ "employer brand", "talent branding", "talent acquisition", "recruiting", "recruitment", "staffing", "headhunting", "labour market", "labor market", "workforce", "hr consulting", "learning resources", "e-learning platform", "lms ", "training solutions", "people analytics", "hr software", "hris", ] # Legacy combined list for Assaf raw contact scoring ICP_TITLE_SIGNALS = MARKETING_TITLES + FOUNDER_TITLES ICP_COMPANY_SIGNALS = BRAND_RELEVANT_COMPANY_SIGNALS SKIP_TITLES = [ "intern", "trainee", "student", "assistant", "coordinator", "junior", "analyst", "freelance designer", # HR / People / Recruiting — not CE's buyer "employer brand", "talent acquisition", "talent brand", "recruiting", "recruitment", "people & culture", "hr manager", "human resources", "people operations", "people partner", "chro", "chief human resources", "chief people", "chief hr", "head of hr", "vp hr", "vp people", "director of people", "head of people", "hr business partner", "hrbp", "organizational development", "employee development", "total rewards", "labor relations", "global mobility", "hr digitalization", "people analytics", ] # ── Signal Scoring Bonuses ──────────────────────────────────────────────────── SIGNAL_SCORES = { "funding": 30, "new_cmo": 25, "rebrand": 25, "launch": 20, "expansion": 20, "hiring": 15, "press": 10, } SIGNAL_KEYWORDS = { "funding": ["raised", "funding", "series a", "series b", "seed round", "million", "investment", "backed", "valuation"], "new_cmo": ["new cmo", "new chief marketing", "new vp marketing", "new head of marketing", "new head of brand", "joins as cmo", "appointed", "hires", "new hire", "new brand"], "rebrand": ["rebrand", "rebranding", "new identity", "new logo", "brand refresh", "visual identity", "brand launch"], "launch": ["launches", "launch", "announces", "new product", "goes live", "ships", "released", "introducing"], "expansion": ["expands", "expansion", "enters", "new market", "international", "europe", "us launch", "global"], "hiring": ["hiring", "we're hiring", "open role", "brand manager", "creative director", "marketing manager", "job posting"], "press": ["featured in", "award", "named", "recognized", "best", "top 10", "coverage"], } # ── Google Sheets Access ────────────────────────────────────────────────────── def get_sheets_client(): from google.oauth2.credentials import Credentials from googleapiclient.discovery import build creds = Credentials.from_authorized_user_file( str(WORKSPACE / "google-auth/token.json") ) return build("sheets", "v4", credentials=creds) # ── Load Contacts ───────────────────────────────────────────────────────────── def load_lukas_contacts(sheets): """Load Lukas's enriched contacts from CRM (LinkedIn - Lukas tab).""" contacts = [] result = sheets.spreadsheets().values().get( spreadsheetId=CRM_SHEET_ID, range="LinkedIn - Lukas!A2:K800" ).execute() for r in result.get("values", []): contacts.append({ "first": r[0] if len(r) > 0 else "", "last": r[1] if len(r) > 1 else "", "title": r[2] if len(r) > 2 else "", "company": r[3] if len(r) > 3 else "", "url": r[4] if len(r) > 4 else "", "email": r[5] if len(r) > 5 else "", "connected": r[6] if len(r) > 6 else "", "owner": "Lukas", "priority": r[8] if len(r) > 8 else "Medium", "status": r[9] if len(r) > 9 else "Not Contacted", "why_ce": r[10] if len(r) > 10 else "", "tier": 1 if (r[8] if len(r) > 8 else "Medium") == "High" else 2, }) return contacts def load_assaf_contacts(): """Load Assaf's raw LinkedIn connections.""" contacts = [] lines = open(ASSAF_CSV, encoding="utf-8").readlines() header_idx = next(i for i, l in enumerate(lines) if l.startswith("First Name")) import csv as csvmod reader = csvmod.DictReader(lines[header_idx:]) for row in reader: url = (row.get("URL") or "").strip() if not url: continue contacts.append({ "first": (row.get("First Name") or "").strip(), "last": (row.get("Last Name") or "").strip(), "title": (row.get("Position") or "").strip(), "company": (row.get("Company") or "").strip(), "url": url, "email": (row.get("Email Address") or "").strip(), "connected": (row.get("Connected On") or "").strip(), "owner": "Assaf", "priority": "Raw", "tier": 4, "why_ce": "", }) return contacts def build_company_density(contacts): from collections import Counter companies = [c["company"].lower() for c in contacts if c["company"]] return Counter(companies) # ── ICP Scoring ─────────────────────────────────────────────────────────────── def icp_score(contact, company_density=None): title = contact["title"].lower() company = contact["company"].lower() if any(s in title for s in SKIP_TITLES): return 0 score = 0 # Marketing titles score highest — they ARE the buyer if any(t in title for t in MARKETING_TITLES): score += 20 # Founders score lower — need company relevance to qualify elif any(t in title for t in FOUNDER_TITLES): score += 8 if any(sig in company for sig in BRAND_RELEVANT_COMPANY_SIGNALS): score += 10 if any(sig in company for sig in BRAND_IRRELEVANT_COMPANY_SIGNALS): score -= 20 # penalise hard for s in ICP_COMPANY_SIGNALS: if s in company: score += 3 # Hard penalty: agency/studio/freelance company names → push to bottom of pool # These will still get caught by the fit gate, but this keeps real buyers at top AGENCY_COMPANY_PATTERNS = [ " agency", "agency ", "creative studio", " studio", "studio ", " creative", "design studio", "branding studio", "advertising", " communications", "& associates", "productions", " group", "saatchi", "wieden", "bbdo", "tbwa", "jwt", "grey", "ddb", "ogilvy", "leo burnett", "razorfish", "digitas", "possible", "mullen", "havas", "publicis", "dentsu", "interpublic", "r/ga", "akqa", "porto rocha", "koto", "betc", "collins", "monks", "buzzman", ] FREELANCE_SIGNALS = ["freelance", "self-employed", "independent", "indépendant"] if any(p in company for p in AGENCY_COMPANY_PATTERNS): score -= 30 if any(p in company for p in FREELANCE_SIGNALS): score -= 50 if contact.get("email"): score += 5 connected = contact.get("connected", "") if "2025" in connected or "2026" in connected: score += 3 if company_density: density = company_density.get(company, 1) if density >= 5: score += 20 elif density >= 3: score += 12 elif density >= 2: score += 6 return score # ── Company Fit Research ────────────────────────────────────────────────────── def load_disqualified() -> dict: """Load cached company disqualification results.""" if DISQUALIFIED_LOG.exists(): with open(DISQUALIFIED_LOG) as f: return json.load(f) return {} def save_disqualified(data: dict): DISQUALIFIED_LOG.parent.mkdir(parents=True, exist_ok=True) with open(DISQUALIFIED_LOG, "w") as f: json.dump(data, f, indent=2) def research_company_fit(contact: dict, disqualified_cache: dict) -> tuple[bool, str]: """ Research a company to determine if they're a valid CE outreach target. Returns: (is_fit: bool, reason: str) Disqualify if: - They ARE an agency / creative studio / design firm (peer, not buyer) - Post-acquisition (budget frozen, no autonomy) - Recent rebrand in last 12 months (already bought, wrong timing) - Company has no real public presence (ghost / too micro) - Company is a 1-2 person boutique (wrong scale) Qualify with flag if: - Fractional / interim CMO (different pitch angle needed) """ import urllib.request company = contact.get("company", "").strip() title = contact.get("title", "").lower() if not company or company.lower() in ("freelance", "self-employed", ""): return False, "No company listed" # Check cache first cache_key = company.lower() if cache_key in disqualified_cache: cached = disqualified_cache[cache_key] return cached["fit"], cached["reason"] # ── Quick title-level checks (no API call needed) ────────────────────── # Fractional/interim CMO: valid but needs a different pitch angle is_fractional = any(x in title for x in ["fractional", "interim", "part-time"]) # ── Company name keyword check (fast, no API) ────────────────────────── company_lower = company.lower() # Check if it's a peer agency/studio by name alone is_peer_by_name = ( any(sig in company_lower for sig in PEER_COMPANY_SIGNALS) and not any(exc in company_lower for exc in AGENCY_EXCEPTIONS) ) if is_peer_by_name: result = (False, f"Agency/studio by name: '{company}' — peer, not buyer") disqualified_cache[cache_key] = {"fit": False, "reason": result[1]} save_disqualified(disqualified_cache) return result # ── Gemini research check ────────────────────────────────────────────── api_key = os.environ.get("GEMINI_API_KEY", "AIzaSyDXqYZInk83iVV4mD29pSuHQKbgkiI1x9Q") url = f"https://generativelanguage.googleapis.com/v1beta/models/gemini-2.5-flash:generateContent?key={api_key}" prompt = ( f"Company: {company}\n" f"Return JSON only. No text before or after. Schema:\n" f'{{\"what_it_does\":\"<5 words>\",\"is_agency\":false,\"is_acquired\":false,\"acquired_by\":null,' f'\"recent_rebrand\":false,\"has_presence\":true,\"is_boutique\":false,\"size\":\"large\"}}\n' f"Rules:\n" f"- is_agency: true if creative/brand/marketing agency or studio\n" f"- is_acquired: true if acquired in last 2 years\n" f"- recent_rebrand: true if rebranded in last 12 months\n" f"- size: micro(<10)/small(10-50)/mid(50-500)/large(500+)\n" f'- if company unknown: {{\"not_found\": true}}' ) # No googleSearch grounding — it limits JSON response length and causes truncation. # Model knowledge is sufficient for company type/size classification. body = json.dumps({ "contents": [{"parts": [{"text": prompt}]}], "generationConfig": {"maxOutputTokens": 800, "temperature": 0.1, "responseMimeType": "application/json"} }).encode() research = None last_err = None for attempt in range(3): try: req = urllib.request.Request(url, data=body, headers={"Content-Type": "application/json"}) with urllib.request.urlopen(req, timeout=20) as resp: data = json.loads(resp.read()) raw_text = data["candidates"][0]["content"]["parts"][0]["text"].strip() if not raw_text: raise ValueError("Empty response from Gemini") # Strip code fences if model ignores responseMimeType if raw_text.startswith("```"): raw_text = raw_text.split("\n", 1)[1].rsplit("```", 1)[0].strip() try: research = json.loads(raw_text) except json.JSONDecodeError: # Truncated/malformed — extract what we can via regex import re as _re m = _re.search(r'"is_agency"\s*:\s*(true|false)', raw_text) is_ag = m.group(1) == "true" if m else False m2 = _re.search(r'"size"\s*:\s*"(\w+)"', raw_text) sz = m2.group(1) if m2 else "unknown" research = {"is_agency": is_ag, "size": sz, "is_acquired": False, "recent_rebrand": False, "has_presence": True, "is_boutique": False, "what_it_does": "unknown", "_partial": True} break except Exception as e: last_err = e time.sleep(2 ** attempt) # 1s, 2s, 4s backoff if research is None: # If all retries failed, don't disqualify — flag as unverified return True, f"⚠️ Research failed ({last_err}) — unverified fit" # ── Apply disqualification rules ─────────────────────────────────────── if research.get("not_found"): result = (False, f"Company not found publicly — ghost or too micro") disqualified_cache[cache_key] = {"fit": False, "reason": result[1]} save_disqualified(disqualified_cache) return result if research.get("is_agency") is True: result = (False, f"Agency/studio: '{company}' does creative work itself — peer, not buyer") disqualified_cache[cache_key] = {"fit": False, "reason": result[1]} save_disqualified(disqualified_cache) return result if research.get("is_acquired") is True: by_whom = research.get("acquired_by", "unknown acquirer") result = (False, f"Post-acquisition: '{company}' acquired by {by_whom} — budget frozen, no autonomy") disqualified_cache[cache_key] = {"fit": False, "reason": result[1]} save_disqualified(disqualified_cache) return result if research.get("recent_rebrand") is True: result = (False, f"Recent rebrand: '{company}' refreshed in last 12 months — wrong timing, come back in 6-9mo") disqualified_cache[cache_key] = {"fit": False, "reason": result[1]} save_disqualified(disqualified_cache) return result if not research.get("has_presence"): result = (False, f"No real public presence: '{company}' — can't assess fit, skip") disqualified_cache[cache_key] = {"fit": False, "reason": result[1]} save_disqualified(disqualified_cache) return result if research.get("is_boutique") is True: result = (False, f"1-2 person boutique: '{company}' — wrong scale for CE") disqualified_cache[cache_key] = {"fit": False, "reason": result[1]} save_disqualified(disqualified_cache) return result # ── Build fit context note ───────────────────────────────────────────── size = research.get("size", "unknown") what = research.get("what_it_does", "") notes = [] if is_fractional: notes.append("⚠️ Fractional/interim CMO — pitch as force multiplier across all clients") if size in ("mid", "large"): notes.append(f"✅ Company size: {size}") elif size == "small": notes.append(f"🟡 Company size: small — check budget capacity") if what: notes.append(f"📌 {what}") fit_reason = " · ".join(notes) if notes else "✅ Qualified" # Cache the positive result too (with TTL via date) disqualified_cache[cache_key] = { "fit": True, "reason": fit_reason, "checked_on": datetime.now(timezone.utc).strftime("%Y-%m-%d"), } save_disqualified(disqualified_cache) return True, fit_reason # ── Signal Search ───────────────────────────────────────────────────────────── def search_signals(contact): """ Search for fresh signals via Gemini + Google Search grounding. Returns dict: {signal_type: snippet, ...} """ import urllib.request import urllib.error company = contact["company"] if not company or company.lower() in ("freelance", "self-employed", ""): return {} api_key = os.environ.get("GEMINI_API_KEY", "AIzaSyDXqYZInk83iVV4mD29pSuHQKbgkiI1x9Q") url = f"https://generativelanguage.googleapis.com/v1beta/models/gemini-2.5-flash:generateContent?key={api_key}" prompt = ( f"Search for very recent news (2025-2026) about the company '{company}'. " f"Look specifically for: funding rounds, new CMO or marketing leadership hires, " f"rebrands, product launches, market expansion, open brand/marketing roles, press coverage. " f"Return only a short factual summary of what you find, max 3 sentences. " f"If nothing relevant found, reply: NO_SIGNAL" ) body = json.dumps({ "contents": [{"parts": [{"text": prompt}]}], "tools": [{"googleSearch": {}}], "generationConfig": {"maxOutputTokens": 200, "temperature": 0.1} }).encode() try: req = urllib.request.Request(url, data=body, headers={"Content-Type": "application/json"}) with urllib.request.urlopen(req, timeout=8) as resp: data = json.loads(resp.read()) text = data["candidates"][0]["content"]["parts"][0]["text"].lower() except Exception: return {} if "no_signal" in text or "no recent" in text or "couldn't find" in text: return {} found = {} for sig_type, keywords in SIGNAL_KEYWORDS.items(): if any(kw in text for kw in keywords): for kw in keywords: idx = text.find(kw) if idx != -1: start = max(0, idx - 40) end = min(len(text), idx + 100) found[sig_type] = text[start:end].strip(" .,\n") break return found def signal_score(signals): return sum(SIGNAL_SCORES.get(s, 0) for s in signals) # ── Sent Log ────────────────────────────────────────────────────────────────── def load_sent(): if SENT_LOG.exists(): with open(SENT_LOG) as f: return json.load(f) return {"sent_urls": [], "total_sent": 0} def save_sent(data): SENT_LOG.parent.mkdir(parents=True, exist_ok=True) with open(SENT_LOG, "w") as f: json.dump(data, f, indent=2) # ── Mark CRM as Outreached ──────────────────────────────────────────────────── def mark_outreached_in_crm(sheets, contacts_sent): """Update Status = Outreached for sent contacts in LinkedIn - Lukas tab.""" result = sheets.spreadsheets().values().get( spreadsheetId=CRM_SHEET_ID, range="LinkedIn - Lukas!E2:J800" # E=URL, J=Status ).execute() rows = result.get("values", []) sent_urls = {c["url"] for c in contacts_sent if c["owner"] == "Lukas"} updates = [] for i, row in enumerate(rows, start=2): url = row[0] if row else "" if url in sent_urls: updates.append({ "range": f"LinkedIn - Lukas!J{i}", "values": [["Outreached"]] }) if updates: sheets.spreadsheets().values().batchUpdate( spreadsheetId=CRM_SHEET_ID, body={"valueInputOption": "RAW", "data": updates} ).execute() # ── Overlap Detection ───────────────────────────────────────────────────────── def find_overlaps(lukas_contacts, assaf_contacts): lukas_urls = {c["url"] for c in lukas_contacts if c["url"]} return {c["url"] for c in assaf_contacts if c["url"] in lukas_urls} # ── Approach Copy ───────────────────────────────────────────────────────────── def build_approach(contact, signals, is_warm_intro=False): title = contact["title"].lower() company = contact["company"] why_ce = contact.get("why_ce", "") # Lead with signal if present signal_lead = "" if "funding" in signals: signal_lead = f"🚀 *{company} just raised — brand budget is incoming. First mover wins.*" elif "new_cmo" in signals: signal_lead = f"👤 *New marketing leadership at {company} — new leaders redefine the brand. Perfect timing.*" elif "rebrand" in signals: signal_lead = f"🔄 *{company} is in a rebrand moment — they're already thinking brand. Walk in with a POV.*" elif "launch" in signals: signal_lead = f"📢 *{company} just launched something — they need creative to support it at scale.*" elif "expansion" in signals: signal_lead = f"🌍 *{company} is expanding — brand needs to travel. CE builds brands that scale.*" elif "hiring" in signals: signal_lead = f"📋 *{company} is building a brand/marketing team — budget confirmed, moment is now.*" elif "press" in signals: signal_lead = f"🏆 *{company} just got coverage — warm moment to reach out while they're visible.*" # Warm intro flag warm = "\n🤝 **Warm intro** — both Assaf + Lukas connected. Assaf drops intro note first, then Lukas follows." if is_warm_intro else "" # Use existing Why CE if available, else generate if why_ce and len(why_ce) > 30: body = why_ce[:300] + ("..." if len(why_ce) > 300 else "") elif any(x in title for x in ["ceo", "founder", "co-founder"]): body = f"*\"Every founder in your space tells the same story. The ones who break through aren't smarter — they're clearer.\"* CE finds the lane nobody else is claiming and builds the creative to own it." elif any(x in title for x in ["cmo", "vp marketing", "head of marketing", "marketing director"]): body = f"*\"What does {company}'s competitive landscape look like right now?\"* CE delivers full competitive intelligence + positioning white space in 5 days. The brief your team needs before they brief the work." elif any(x in title for x in ["creative director", "head of brand", "design director", "vp design"]): body = f"*\"What's your biggest production bottleneck?\"* CE is the autonomous creative production layer — you set the direction, we ship at speed. No headcount, no lag." else: body = f"*\"There's a positioning gap in {company}'s category nobody's claiming yet.\"* CE maps it and builds the creative to own it." template = "Signal-Led" if signal_lead else ("Warm Intro" if is_warm_intro else "White Space") approach = "\n".join(filter(None, [signal_lead, body, warm])) return template, approach # ── Build Message ───────────────────────────────────────────────────────────── def _render_picks(picks, start_index=1): lines = [] for i, item in enumerate(picks, start_index): c = item["contact"] signals = item["signals"] is_warm = item.get("warm_intro", False) name = f"{c['first']} {c['last']}".strip() email_line = f"\n**Email:** {c['email']} ✅" if c.get("email") else "" owner_emoji = "🔵" if c["owner"] == "Lukas" else "🟣" priority_tag = f" · {c.get('priority','')}" if c.get("priority") not in ("Raw", "") else "" fit_reason = item.get("fit_reason", "✅ Qualified") fit_line = f"**Company fit:** {fit_reason}" # Signal summary line if signals: sig_tags = " · ".join( {"funding": "🚀 Funding", "new_cmo": "👤 New CMO", "rebrand": "🔄 Rebrand", "launch": "📢 Launch", "expansion": "🌍 Expansion", "hiring": "📋 Hiring", "press": "🏆 Press"}.get(s, s) for s in signals ) signal_line = f"**Signals:** {sig_tags}" else: signal_line = "**Signals:** none detected" template, approach = build_approach(c, signals, is_warm) lines.append( f"### {i}. {name} {owner_emoji}\n" f"**Title:** {c['title']} @ {c['company']}\n" f"**LinkedIn:** {c['url']}{email_line}\n" f"**Connected:** {c['connected']} · **Owner:** {c['owner']}{priority_tag}\n" f"{fit_line}\n" f"{signal_line}\n\n" f"**Template:** {template}\n" f"**Approach:** {approach}\n\n" f"---\n" ) return "\n".join(lines) def build_message(lukas_picks, assaf_picks, date_str): parts = [f"## 📬 Daily Outreach — {date_str}\n"] parts.append("*Ranked by: signal activity → ICP role → network density*\n") if lukas_picks: parts.append("### 🔵 Lukas — Today's 5\n---\n") parts.append(_render_picks(lukas_picks, start_index=1)) if assaf_picks: parts.append("\n\n### 🟣 Assaf — Today's 5\n---\n") parts.append(_render_picks(assaf_picks, start_index=1)) parts.append( "\n*🔵 Lukas owns outreach · 🟣 Assaf owns outreach · " "🤝 Warm intro = Assaf introduces first*\n" "*Next batch tomorrow 8:00 UTC*" ) return "\n".join(parts) # ── Discord Posting ─────────────────────────────────────────────────────────── def get_discord_token() -> str: """Get Kitt bot token from OpenClaw config.""" token = os.environ.get("DISCORD_BOT_TOKEN", "") if token: return token try: cfg_path = os.path.expanduser("~/.openclaw/openclaw.json") with open(cfg_path) as f: cfg = json.load(f) return cfg["channels"]["discord"]["accounts"]["kitt"]["token"] except Exception: return "" def post_to_discord(message: str, dry_run: bool = False) -> bool: if dry_run: print("── DRY RUN ──────────────────────────────────") print(message) print("─────────────────────────────────────────────") return True token = get_discord_token() if not token: print("ERROR: Discord bot token not found", file=sys.stderr) return False # Discord has a 2000 char limit — split into chunks if needed chunks = [] if len(message) <= 2000: chunks = [message] else: # Split on contact dividers (---) parts = message.split("\n---\n") current = "" for part in parts: segment = part + "\n---\n" if len(current) + len(segment) > 1900: if current: chunks.append(current.rstrip()) current = segment else: current += segment if current: chunks.append(current.rstrip()) success = True for chunk in chunks: payload = json.dumps({"content": chunk}).encode("utf-8") req = urllib.request.Request( f"https://discord.com/api/v10/channels/{OUTREACH_CHANNEL}/messages", data=payload, headers={ "Authorization": f"Bot {token}", "Content-Type": "application/json", }, method="POST", ) try: with urllib.request.urlopen(req, timeout=10) as resp: if resp.status not in (200, 201): success = False time.sleep(0.5) # rate limit courtesy except urllib.error.HTTPError as e: print(f"Discord error {e.code}: {e.read().decode()}", file=sys.stderr) success = False return success # ── Main ────────────────────────────────────────────────────────────────────── def main(): import argparse parser = argparse.ArgumentParser() parser.add_argument("--dry-run", action="store_true", help="Print message, don't post to Discord") parser.add_argument("--count", type=int, default=BATCH_SIZE, help="Number of contacts to pick") args = parser.parse_args() today = datetime.now(timezone.utc).strftime("%a %b %d, %Y") # Weekend skip if datetime.now(timezone.utc).weekday() >= 5: print("Weekend — skipping outreach") sys.exit(0) sent_data = load_sent() sent_urls = set(sent_data.get("sent_urls", [])) # Load contacts try: sheets = get_sheets_client() lukas_contacts = load_lukas_contacts(sheets) except Exception as e: print(f"WARNING: Could not load Lukas CRM contacts: {e}", file=sys.stderr) lukas_contacts = [] assaf_contacts = load_assaf_contacts() all_assaf = {c["url"]: c for c in assaf_contacts if c["url"]} # Find overlaps overlap_urls = find_overlaps(lukas_contacts, list(all_assaf.values())) # Mark overlap contacts as tier 3 for c in lukas_contacts: if c["url"] in overlap_urls: c["tier"] = 3 c["warm_intro"] = True # Assaf-unique = not in Lukas list at all lukas_urls = {c["url"] for c in lukas_contacts if c["url"]} assaf_unique = [c for c in assaf_contacts if c["url"] not in lukas_urls] # Build company density for Assaf scoring company_density = build_company_density(assaf_unique) for c in assaf_unique: c["icp"] = icp_score(c, company_density) # Combine: Lukas tiers 1-3, then Assaf tier 4 # Apply ICP gate to ALL contacts — CRM priority doesn't override def is_valid_icp(contact): title = contact.get("title", "").lower() company = contact.get("company", "").lower() # Hard skip: HR / recruiting / junior titles if any(skip in title for skip in SKIP_TITLES): return False # Hard skip: company is in HR / recruiting / L&D space — not CE's buyer if any(sig in company for sig in SKIP_COMPANY_SIGNALS): return False # Tier A: direct marketing title → always qualify if any(t in title for t in MARKETING_TITLES): return True # Tier B: founder/CEO → only qualify if company is clearly brand-relevant if any(t in title for t in FOUNDER_TITLES): # Hard disqualify: B2B internal tooling if any(sig in company for sig in BRAND_IRRELEVANT_COMPANY_SIGNALS): return False # Qualify only if company has a positive brand-relevant signal if any(sig in company for sig in BRAND_RELEVANT_COMPANY_SIGNALS): return True # No signal either way → exclude. Marketing titles get priority. return False # Everything else: skip return False # Known agency/studio blocklist — pre-filter before fit gate to avoid wasting Gemini calls KNOWN_AGENCIES = { "akqa", "r/ga", "rga", "wieden+kennedy", "wieden + kennedy", "wk", "bbdo", "ddb", "tbwa", "leo burnett", "saatchi", "grey", "jwt", "ogilvy", "publicis", "havas", "mccann", "mccann paris", "mccann worldgroup", "edelman", "razorfish", "monks", "porto rocha", "porto rocha studio", "betc", "betc paris", "droga5", "goodby silverstein", "gsandp", "bbh", "rapp", "arc", "buzzman", "koto", "pentagram", "wolff olins", "sg360", "iris", "mother", "party", "anomaly", "72andsunny", "deutsch", "huge", "digitas", "sapient", "isobar", "vml", "cheil", "dentsu", "lowe", "foote cone belding", "fcb", "mullen", "arnold", "hill holliday", "cramer-krasselt", "barkley", "vitro", "moxie", "proof", "david&goliath", "freelance", "self-employed", "self employed", "independent", "indépendant", "freiberuflich", "selbstständig", } def is_known_agency(contact): company = contact.get("company", "").lower().strip() if not company: return True # no company = freelance, skip for a in KNOWN_AGENCIES: if a in company: return True return False # ── Build per-owner candidate lists ────────────────────────────────────── # Lukas: apply HR/employer-brand skip filter — Marketing is the target ICP. # SKIP_TITLES blocks HR, Employer Branding, CHRO, Recruiting, etc. lukas_candidates = [] for c in lukas_contacts: if c["url"] not in sent_urls and c.get("status", "Not Contacted") != "Outreached": title = c.get("title", "").lower() if any(skip in title for skip in SKIP_TITLES): print(f" ✗ SKIP (Lukas HR filter): {c.get('name','')} — {c.get('title','')}", file=sys.stderr) continue lukas_candidates.append(c) lukas_candidates.sort(key=lambda c: (c.get("tier", 4), -(c.get("icp", 0)))) # Assaf: apply ICP filter + known-agency pre-screen (raw CSV export — needs scoring) assaf_candidates = [] for c in sorted(assaf_unique, key=lambda x: x["icp"], reverse=True): if c["url"] not in sent_urls and c["icp"] > 0 and is_valid_icp(c) and not is_known_agency(c): assaf_candidates.append(c) def run_fit_and_signals(pool_candidates, owner_label): """Run fit gate + signal scan on a candidate list, return BATCH_SIZE picks.""" disqualified_cache = load_disqualified() fit_candidates = [] max_to_check = SIGNAL_POOL * 8 # hard cap: check up to 8x pool to find fit ones print(f"\nRunning company fit gate for {owner_label} (checking up to {max_to_check})...", file=sys.stderr) for c in pool_candidates[:max_to_check]: if len(fit_candidates) >= SIGNAL_POOL: break is_fit, fit_reason = research_company_fit(c, disqualified_cache) if is_fit: c["fit_reason"] = fit_reason fit_candidates.append(c) else: print(f" ✗ SKIP: {c.get('company','')} — {fit_reason}", file=sys.stderr) time.sleep(SEARCH_DELAY) if not fit_candidates: print(f"⚠️ No fit candidates for {owner_label} — pool may be exhausted.", file=sys.stderr) return [] print(f"Scanning {min(len(fit_candidates), SIGNAL_POOL)} qualified {owner_label} candidates for signals...", file=sys.stderr) scored = [] for c in fit_candidates[:SIGNAL_POOL]: signals = search_signals(c) sig_score = signal_score(signals) base_score = c.get("icp", 10) if c["owner"] == "Assaf" else ( 30 if c.get("priority") == "High" else 15 ) tier_weight = (5 - c.get("tier", 4)) * 100 total = tier_weight + sig_score + base_score scored.append({ "contact": c, "signals": signals, "sig_score": sig_score, "total_score": total, "warm_intro": c.get("warm_intro", False), "fit_reason": c.get("fit_reason", "✅ Qualified"), }) time.sleep(SEARCH_DELAY) scored.sort(key=lambda x: x["total_score"], reverse=True) return scored[:BATCH_SIZE] lukas_picks = run_fit_and_signals(lukas_candidates, "Lukas") assaf_picks = run_fit_and_signals(assaf_candidates, "Assaf") picks = lukas_picks + assaf_picks if not picks: print("No contacts remaining for either owner.") sys.exit(0) # Build message — split into two sections message = build_message(lukas_picks, assaf_picks, today) # Mark as sent (skip in dry-run) if not args.dry_run: for item in picks: url = item["contact"]["url"] if url: sent_urls.add(url) sent_data["sent_urls"] = list(sent_urls) sent_data["total_sent"] = len(sent_urls) sent_data["last_run"] = datetime.now(timezone.utc).isoformat() save_sent(sent_data) try: mark_outreached_in_crm(sheets, [item["contact"] for item in picks]) except Exception as e: print(f"WARNING: CRM update failed: {e}", file=sys.stderr) # Save message to file for reliable retrieval msg_file = WORKSPACE / "work/internal-ce/operations/linkedin/last-outreach-message.txt" msg_file.write_text(message, encoding="utf-8") # Output message — posting is handled by the caller (Kitt via message tool) if args.dry_run: print("── DRY RUN ──────────────────────────────────") print("OUTREACH_MESSAGE_START") print(message) print("OUTREACH_MESSAGE_END") signal_count = sum(1 for p in picks if p["signals"]) print(f"CONTACTS_SENT: {len(picks)} (Lukas: {len(lukas_picks)}, Assaf: {len(assaf_picks)})") print(f"SIGNAL_HITS: {signal_count}/{len(picks)}") print(f"TOTAL_SENT: {sent_data['total_sent']}") print(f"LUKAS_REMAINING: {sum(1 for c in lukas_contacts if c['url'] not in sent_urls)}") print(f"ASSAF_REMAINING: {sum(1 for c in assaf_unique if c['url'] not in sent_urls and c['icp'] > 0)}") if __name__ == "__main__": main()