netsift.py

#!/usr/bin/env python3
"""netsift — turn your LinkedIn connections export into ranked ICP candidate lists.

Runs 100% locally: no network calls, no LinkedIn API, no account risk. You give it
your own data export and a YAML/JSON config describing who you're looking for, and
it scores, segments and ranks your connections into candidate lists.

Usage:
    python netsift.py --connections Connections.csv --config config.yaml --out out/
    python netsift.py -c export.zip -k audience.yaml -o out/ --top 50

See README.md for the full guide.
"""
from __future__ import annotations

import argparse
import csv
import io
import json
import re
import sys
import zipfile
from collections import Counter
from pathlib import Path

# ---------------------------------------------------------------------------
# Config loading
# ---------------------------------------------------------------------------

def load_config(path: Path) -> dict:
    text = path.read_text(encoding="utf-8")
    if path.suffix.lower() in (".yaml", ".yml"):
        try:
            import yaml  # type: ignore
        except ImportError:
            sys.exit(
                "PyYAML is required for YAML configs. Install it with "
                "`pip install pyyaml`, or use a JSON config instead."
            )
        return yaml.safe_load(text)
    if path.suffix.lower() == ".json":
        return json.loads(text)
    sys.exit(f"Unsupported config format: {path.suffix} (use .yaml, .yml or .json)")


# ---------------------------------------------------------------------------
# Connections loading (handles LinkedIn's quirky export)
# ---------------------------------------------------------------------------

def _read_connections_text(path: Path) -> str:
    """Return the raw text of Connections.csv from a .csv, a .zip or a directory."""
    if path.is_dir():
        hit = next(path.rglob("Connections.csv"), None)
        if not hit:
            sys.exit(f"No Connections.csv found anywhere under {path}")
        return hit.read_text(encoding="utf-8-sig")
    if path.suffix.lower() == ".zip":
        with zipfile.ZipFile(path) as zf:
            names = [n for n in zf.namelist() if n.endswith("Connections.csv")]
            if not names:
                sys.exit(f"No Connections.csv inside {path}")
            with zf.open(names[0]) as fh:
                return io.TextIOWrapper(fh, encoding="utf-8-sig").read()
    return path.read_text(encoding="utf-8-sig")


def load_connections(path: Path) -> list[dict]:
    """Parse Connections.csv, skipping LinkedIn's 'Notes:' preamble, dedup by URL."""
    lines = _read_connections_text(path).splitlines()
    try:
        start = next(i for i, l in enumerate(lines) if l.startswith("First Name,"))
    except StopIteration:
        sys.exit("Could not find the 'First Name,...' header row — is this a "
                 "LinkedIn Connections.csv?")
    rows, seen = [], set()
    for r in csv.DictReader(lines[start:]):
        url = (r.get("URL") or "").strip()
        if url and url in seen:
            continue
        if url:
            seen.add(url)
        first = (r.get("First Name") or "").strip()
        last = (r.get("Last Name") or "").strip()
        name = f"{first} {last}".strip()
        if not name:
            continue
        rows.append({
            "name": name,
            "position": (r.get("Position") or "").strip(),
            "company": (r.get("Company") or "").strip(),
            "url": url,
            "email": (r.get("Email Address") or "").strip(),
            "connected": (r.get("Connected On") or "").strip(),
        })
    return rows


# ---------------------------------------------------------------------------
# Matching
# ---------------------------------------------------------------------------

def make_matcher(term: str, substring: bool):
    """Compile a matcher.

    Default behaviour (substring=False) is "smart":
      * short single-token acronyms (<=3 alphanumeric chars like 'vd', 'cio',
        'dpo', 'grc') are matched as WHOLE WORDS, so 'vd' won't hit 'avdelning'
        and 'cio' won't hit 'association'.
      * everything else is matched as a SUBSTRING, so stems work ('internrevis'
        -> 'internrevisionschef') and company fragments work ('bank' ->
        'Swedbank', 'tech' -> 'Reachtech').
    Set substring_match: true in config to force naive substring matching for all
    terms (disables acronym protection)."""
    term = term.lower()
    esc = re.escape(term)
    if not term:
        return re.compile(esc, re.IGNORECASE | re.UNICODE)
    if not substring and term.isalnum() and len(term) <= 3:
        pat = r"(?<!\w)" + esc + r"(?!\w)"
    else:
        pat = esc
    return re.compile(pat, re.IGNORECASE | re.UNICODE)


class Compiled:
    """Pre-compiled config for fast scoring."""

    def __init__(self, cfg: dict):
        s = cfg.get("settings", {}) or {}
        self.substring = bool(s.get("substring_match", False))
        self.industry_bonus = float(s.get("industry_bonus", 0))
        self.search_company = bool(s.get("search_company_for_keywords", True))

        self.segments = []  # list of (key, name, priority, [(kw, weight, rx)])
        for seg in cfg.get("segments", []):
            kws = [(kw, float(w), make_matcher(kw, self.substring))
                   for kw, w in (seg.get("keywords", {}) or {}).items()]
            self.segments.append({
                "key": str(seg["key"]),
                "name": seg.get("name", str(seg["key"])),
                "priority": int(seg.get("priority", 99)),
                "keywords": kws,
            })
        self.seg_priority = {s["key"]: s["priority"] for s in self.segments}
        self.seg_name = {s["key"]: s["name"] for s in self.segments}

        self.company_signals = []
        for cs in cfg.get("company_signals", []):
            self.company_signals.append({
                "segment": str(cs["segment"]),
                "weight": float(cs.get("weight", 1)),
                "label": cs.get("label", "company-signal"),
                "requires_existing_match": bool(cs.get("requires_existing_match", False)),
                "rx": [make_matcher(m, self.substring) for m in cs.get("match", [])],
            })

        self.industries = {
            label: [make_matcher(m, self.substring) for m in terms]
            for label, terms in (cfg.get("industries", {}) or {}).items()
        }

        noise = cfg.get("noise", {}) or {}
        self.noise_pos = [make_matcher(m, self.substring) for m in noise.get("positions", [])]
        self.noise_co = [make_matcher(m, self.substring) for m in noise.get("companies", [])]


def industry_of(compiled: Compiled, company_l: str):
    for label, rxs in compiled.industries.items():
        if any(rx.search(company_l) for rx in rxs):
            return label
    return None


def score_one(compiled: Compiled, conn: dict) -> dict:
    pos_l = conn["position"].lower()
    co_l = conn["company"].lower()
    haystack = pos_l + " || " + co_l if compiled.search_company else pos_l

    # noise filter first
    if any(rx.search(pos_l) for rx in compiled.noise_pos) or \
       any(rx.search(co_l) for rx in compiled.noise_co):
        return {**conn, "segment": "NOISE", "score": 0.0, "industry": "", "signals": ""}

    scores, signals = {}, {}
    for seg in compiled.segments:
        s, hits = 0.0, []
        for kw, w, rx in seg["keywords"]:
            if rx.search(haystack):
                s += w
                hits.append(kw)
        scores[seg["key"]] = s
        signals[seg["key"]] = hits

    for cs in compiled.company_signals:
        if cs["requires_existing_match"] and scores.get(cs["segment"], 0) <= 0:
            continue
        if any(rx.search(co_l) for rx in cs["rx"]):
            scores[cs["segment"]] = scores.get(cs["segment"], 0) + cs["weight"]
            signals.setdefault(cs["segment"], []).append(cs["label"])

    matched = {k: v for k, v in scores.items() if v > 0}
    if not matched:
        return {**conn, "segment": "UNSURE", "score": 0.0, "industry": "", "signals": ""}

    # best = highest score, tie-break on lower priority value
    best = max(matched, key=lambda k: (matched[k], -compiled.seg_priority.get(k, 99)))
    industry = industry_of(compiled, co_l)
    total = matched[best] + (compiled.industry_bonus if industry else 0)
    return {
        **conn,
        "segment": best,
        "score": round(total, 1),
        "industry": industry or "",
        "signals": ", ".join(dict.fromkeys(signals.get(best, []))),
    }


# ---------------------------------------------------------------------------
# Output
# ---------------------------------------------------------------------------

FIELDS = ["name", "segment", "position", "company", "industry", "score",
          "signals", "email", "url", "connected"]
HEADERS = ["Name", "Segment", "Title", "Company", "Industry", "Score",
           "Signals", "Email", "LinkedIn", "Connected On"]


def sort_key(o):
    return (-o["score"], 0 if o["industry"] else 1, o["name"].lower())


def write_csv(path: Path, rows: list[dict]):
    with open(path, "w", newline="", encoding="utf-8") as f:
        w = csv.writer(f)
        w.writerow(HEADERS)
        for o in rows:
            w.writerow([o.get(k, "") for k in FIELDS])


def md_table(rows, seg_name):
    out = ["| # | Name | Segment | Title | Company | Industry | Score | LinkedIn |",
           "|---|---|---|---|---|---|---|---|"]
    for i, o in enumerate(rows, 1):
        em = " 📧" if o["email"] else ""
        link = f"[profile]({o['url']})" if o["url"] else ""
        out.append(
            f"| {i} | {o['name']}{em} | {seg_name.get(o['segment'], o['segment'])} | "
            f"{o['position'] or '—'} | {o['company'] or '—'} | {o['industry']} | "
            f"{o['score']} | {link} |"
        )
    return "\n".join(out)


def main():
    ap = argparse.ArgumentParser(
        description="Rank your LinkedIn connections into ICP candidate lists (runs locally).")
    ap.add_argument("-c", "--connections", required=True,
                    help="Connections.csv, the LinkedIn export .zip, or a folder containing it")
    ap.add_argument("-k", "--config", required=True, help="audience config (.yaml/.json)")
    ap.add_argument("-o", "--out", default="out", help="output directory (default: out/)")
    ap.add_argument("--top", type=int, default=50, help="size of the merged top-N list (default: 50)")
    ap.add_argument("--per-segment", type=int, default=40,
                    help="rows per segment in report.md (0 = all; default: 40)")
    args = ap.parse_args()

    cfg = load_config(Path(args.config))
    compiled = Compiled(cfg)
    if not compiled.segments:
        sys.exit("Config has no segments — define at least one under 'segments:'.")

    conns = load_connections(Path(args.connections))
    scored = [score_one(compiled, c) for c in conns]

    out = Path(args.out)
    out.mkdir(parents=True, exist_ok=True)

    seg_name = compiled.seg_name
    by_seg = {s["key"]: sorted([o for o in scored if o["segment"] == s["key"]], key=sort_key)
              for s in compiled.segments}

    # per-segment CSVs
    for key, rows in by_seg.items():
        write_csv(out / f"candidates_{key}.csv", rows)
    write_csv(out / "all_scored.csv", sorted(scored, key=sort_key))
    write_csv(out / "noise.csv", [o for o in scored if o["segment"] == "NOISE"])

    # merged top-N: sort matched by (priority, -score, ...)
    matched = [o for o in scored if o["segment"] in seg_name]
    matched.sort(key=lambda o: (compiled.seg_priority.get(o["segment"], 99),) + sort_key(o))
    top = matched[:args.top]
    top_md = [f"# Top {len(top)} candidates (merged)", "",
              "Sorted by segment priority, then ICP score. 📧 = email present in export.",
              "", md_table(top, seg_name), ""]
    (out / f"top{args.top}.md").write_text("\n".join(top_md))

    # full report
    rep = ["# Candidate report", "",
           f"Source: LinkedIn connections export — {len(scored)} contacts after dedup.  ",
           "Ranked per segment, strongest ICP match first. 📧 = email present.", "",
           "| Segment | Count | With email | Regulated industry |",
           "|---|---|---|---|"]
    for s in sorted(compiled.segments, key=lambda x: x["priority"]):
        rows = by_seg[s["key"]]
        rep.append(f"| {s['name']} | {len(rows)} | "
                   f"{sum(1 for o in rows if o['email'])} | "
                   f"{sum(1 for o in rows if o['industry'])} |")
    n_unsure = sum(1 for o in scored if o["segment"] == "UNSURE")
    n_noise = sum(1 for o in scored if o["segment"] == "NOISE")
    rep += [f"| (no match) | {n_unsure} | | |",
            f"| (filtered noise) | {n_noise} | | |", "",
            "Full lists are in the per-segment CSVs. Below: "
            f"{'all rows' if args.per_segment == 0 else f'top {args.per_segment}'} per segment.", ""]
    lim = None if args.per_segment == 0 else args.per_segment
    for s in sorted(compiled.segments, key=lambda x: x["priority"]):
        rows = by_seg[s["key"]]
        rep += [f"## {s['name']} ({len(rows)})", "", md_table(rows[:lim], seg_name)]
        if lim and len(rows) > lim:
            rep.append(f"\n*…+{len(rows) - lim} more in candidates_{s['key']}.csv.*")
        rep.append("")
    (out / "report.md").write_text("\n".join(rep))

    # console summary
    c = Counter(o["segment"] for o in scored)
    print(f"Scored {len(scored)} connections → {out}/")
    for s in sorted(compiled.segments, key=lambda x: x["priority"]):
        print(f"  {s['key']} {s['name']}: {c.get(s['key'], 0)}")
    print(f"  UNSURE (no match): {n_unsure}")
    print(f"  NOISE (filtered): {n_noise}")
    print(f"Wrote: report.md, top{args.top}.md, candidates_*.csv, all_scored.csv, noise.csv")


if __name__ == "__main__":
    main()