twenty-partner-triage

1#!/usr/bin/env python32"""Deterministic ranker for the twenty-partner-triage skill.3 4Ranks APPLICATION-stage partners by NET-NEW value vs the current VALIDATED set:5geography and language we don't yet cover (weighted high), plus a "real Twenty6work" proof signal read from the application notes. Skill volume is deliberately7capped so generic dev shops that spray skill lists don't dominate.8 9Reads creds from ~/.twenty/credentials.env. Emits ranked JSON to stdout.10Run `rank.py --selftest` to verify scoring without hitting the API.11 12# ponytail: fixed weights + tier thresholds. Tune the WEIGHTS/THRESHOLDS dicts13# below if the ranking drifts; everything else is mechanical.14"""15import json16import os17import re18import sys19import urllib.request20 21WEIGHTS = {"geo": 3, "lang": 3, "scope": 1, "skill": 1, "skill_cap": 3, "proof": 6}22THRESHOLDS = {"A": 12, "B": 5}  # >=A => A; >=B => B; else C. Any proof => at least A.23 24# "real Twenty work" signals in the free-text notes (the high-confidence axis).25PROOF_WORKSPACE = re.compile(r"https?://[^\s]*(twenty|crm)[^\s]*", re.I)26PROOF_CUSTOMERS = re.compile(r"(customers?\s+onboarded|real implementation|clients?\s+(moving|migrat)|delivered \d|named customer)", re.I)27PROOF_MIGRATION = re.compile(r"(switch\w*|migrat\w*|moved|replac\w*|our crm|own crm|we use twenty|dogfood|managed service)", re.I)28 29CRED_PATH = os.path.expanduser("~/.twenty/credentials.env")30 31 32def load_creds():33    if not os.path.exists(CRED_PATH):34        sys.exit(f"Missing {CRED_PATH}. Copy the partners key from the app's "35                 ".env.prod into it (TWENTY_PARTNERS_API_URL / _API_KEY).")36    env = {}37    with open(CRED_PATH) as fh:38        for line in fh:39            line = line.strip()40            if line and not line.startswith("#") and "=" in line:41                k, v = line.split("=", 1)42                env[k.strip()] = v.strip()43    url = env.get("TWENTY_PARTNERS_API_URL")44    key = env.get("TWENTY_PARTNERS_API_KEY")45    if not url or not key:46        sys.exit("credentials.env is missing TWENTY_PARTNERS_API_URL or TWENTY_PARTNERS_API_KEY.")47    return url.rstrip("/"), key48 49 50def fetch_all_partners(url, key):51    recs, after = [], None52    while True:53        path = f"{url}/rest/partners?limit=200&depth=1" + (f"&starting_after={after}" if after else "")54        req = urllib.request.Request(path, headers={"Authorization": f"Bearer {key}", "User-Agent": "Mozilla/5.0"})55        data = json.load(urllib.request.urlopen(req))56        page = data["data"]["partners"]57        recs += page58        info = data.get("pageInfo", {})59        if info.get("hasNextPage") and page:60            after = info["endCursor"]61            continue62        return recs63 64 65def tok(value):66    """Flatten any nested value into a set of lowercased non-empty string tokens."""67    out = set()68    if value is None:69        return out70    if isinstance(value, str):71        s = value.strip().lower()72        if s:73            out.add(s)74    elif isinstance(value, list):75        for item in value:76            out |= tok(item)77    elif isinstance(value, dict):78        for item in value.values():79            out |= tok(item)80    else:81        out.add(str(value).lower())82    return out83 84 85def notes_str(rec):86    v = rec.get("applicationNotes")87    if isinstance(v, str):88        return v89    return "" if v is None else json.dumps(v)90 91 92def contact(rec):93    persons = rec.get("persons") or []94    if not persons:95        return None, None, rec.get("linkedin")96    p = persons[0]97    name = p.get("name") or {}98    full = " ".join(x for x in [name.get("firstName"), name.get("lastName")] if x) or None99    email = (p.get("emails") or {}).get("primaryEmail")100    linkedin = p.get("linkedinLink") or rec.get("linkedin")101    return full, email, linkedin102 103 104def baseline(validated):105    geo, lang, scope, skill = set(), set(), set(), set()106    for r in validated:107        geo |= tok(r.get("region")) | tok(r.get("country"))108        lang |= tok(r.get("languagesSpoken"))109        scope |= tok(r.get("partnerScope"))110        skill |= tok(r.get("skills"))111    return {"geo": geo, "lang": lang, "scope": scope, "skill": skill}112 113 114def score_one(rec, base):115    new_geo = sorted((tok(rec.get("region")) | tok(rec.get("country"))) - base["geo"])116    new_lang = sorted(tok(rec.get("languagesSpoken")) - base["lang"])117    new_scope = sorted(tok(rec.get("partnerScope")) - base["scope"])118    new_skill = sorted(tok(rec.get("skills")) - base["skill"])119    notes = notes_str(rec)120    proof = {121        "workspace_url": bool(PROOF_WORKSPACE.search(notes)),122        "customers": bool(PROOF_CUSTOMERS.search(notes)),123        "migration": bool(PROOF_MIGRATION.search(notes)),124    }125    has_proof = any(proof.values())126    score = (WEIGHTS["geo"] * len(new_geo)127             + WEIGHTS["lang"] * len(new_lang)128             + WEIGHTS["scope"] * len(new_scope)129             + WEIGHTS["skill"] * min(len(new_skill), WEIGHTS["skill_cap"])130             + (WEIGHTS["proof"] if has_proof else 0))131    if has_proof or score >= THRESHOLDS["A"]:132        tier = "A"133    elif score >= THRESHOLDS["B"]:134        tier = "B"135    else:136        tier = "C"137    name, email, linkedin = contact(rec)138    return {139        "name": rec.get("name"),140        "score": score,141        "tier": tier,142        "new_geo": new_geo,143        "new_lang": new_lang,144        "new_scope": new_scope,145        "new_skills": new_skill,146        "proof": {k: v for k, v in proof.items() if v},147        "team": rec.get("typeOfTeam"),148        "contact_name": name,149        "email": email,150        "linkedin": linkedin if isinstance(linkedin, str) else (linkedin or {}).get("primaryLinkUrl") if isinstance(linkedin, dict) else None,151        "website": (rec.get("website") or {}).get("primaryLinkUrl") if isinstance(rec.get("website"), dict) else None,152        "notes": notes.strip()[:400],153    }154 155 156def rank(recs):157    validated = [r for r in recs if r.get("validationStage") == "VALIDATED"]158    apps = [r for r in recs if r.get("validationStage") == "APPLICATION"]159    base = baseline(validated)160    ranked = sorted((score_one(r, base) for r in apps), key=lambda d: -d["score"])161    return {162        "validated_count": len(validated),163        "application_count": len(apps),164        "coverage": {k: sorted(v) for k, v in base.items()},165        "ranked": ranked,166    }167 168 169def selftest():170    base_recs = [{"validationStage": "VALIDATED", "country": ["france"],171                  "languagesSpoken": ["french", "english"], "partnerScope": ["development"],172                  "skills": ["react", "postgres"]}]173    apps = [174        {"validationStage": "APPLICATION", "name": "gap+proof", "country": ["germany"],175         "languagesSpoken": ["german"], "applicationNotes":176         "Live workspace https://crm.acme.de — customers onboarded: Foo GmbH"},177        {"validationStage": "APPLICATION", "name": "skill-sprayer",178         "skills": ["php", "vue", "kotlin", "swift", "laravel", "mongodb"]},179        {"validationStage": "APPLICATION", "name": "empty"},180    ]181    out = rank(base_recs + apps)["ranked"]182    order = [r["name"] for r in out]183    assert order == ["gap+proof", "skill-sprayer", "empty"], order184    assert out[0]["tier"] == "A", out[0]185    assert out[1]["score"] == WEIGHTS["skill_cap"], out[1]  # 6 new skills capped at 3186    assert out[2]["tier"] == "C", out[2]187    print("selftest ok:", order)188 189 190if __name__ == "__main__":191    if "--selftest" in sys.argv:192        selftest()193    else:194        url, key = load_creds()195        print(json.dumps(rank(fetch_all_partners(url, key)), indent=2, ensure_ascii=False))196