`scripts/build_profile.py`

194 lines
#!/usr/bin/env python3
"""build_profile.py — distill a reusable LinkedIn VOICE PROFILE from your own
past posts. This is the "analyzes your past posts to learn your themes + voice,
style, and tone" half of what Okara's LinkedIn Agent v2 sells — run locally at
$0 instead of paying for it.

INPUTS (pick one):
  --export  <Shares.csv>   LinkedIn "Download your data" export (the honest,
                           TOS-clean, $0 way to read YOUR OWN past posts).
                           Get it at: linkedin.com/mypreferences/d/download-my-data
                           → "Posts" → wait for email → unzip → Shares.csv
  --posts   <file.txt>     Plain text: posts separated by a blank line or a
                           line of "---". Paste-in fallback.

OUTPUT:
  data/profiles/<name>.json   the voice profile the generator consumes.

The style STATS are computed deterministically in Python (free, reproducible);
only the natural-language theme/voice synthesis calls local Ollama.

Usage:
  python3 scripts/build_profile.py --export ~/Downloads/Shares.csv --name steve
  python3 scripts/build_profile.py --posts  ~/Desktop/my_posts.txt  --name steve
"""
import argparse
import csv
import datetime
import json
import os
import re
import sys

HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, HERE)
import _ollama  # noqa: E402

ROOT = os.path.dirname(HERE)
PROFILE_DIR = os.path.join(ROOT, "data", "profiles")

EMOJI_RE = re.compile(
    "[\U0001F300-\U0001FAFF\U00002600-\U000027BF\U0001F1E6-\U0001F1FF←-⇿⬀-⯿]"
)


def load_from_csv(path):
    """LinkedIn Shares.csv → list of post texts. Column name has varied over
    the years (ShareCommentary / Commentary / Text), so auto-detect."""
    posts = []
    with open(path, newline="", encoding="utf-8", errors="replace") as f:
        reader = csv.DictReader(f)
        cols = reader.fieldnames or []
        # rank candidate text columns
        cand = None
        for want in ("ShareCommentary", "Commentary", "ShareText", "Text", "content"):
            for c in cols:
                if c and c.strip().lower() == want.lower():
                    cand = c
                    break
            if cand:
                break
        if not cand:
            sys.exit(f"Could not find a post-text column in {os.path.basename(path)}. "
                     f"Columns present: {cols}")
        for row in reader:
            t = (row.get(cand) or "").strip()
            if t:
                posts.append(t)
    return posts


def load_from_txt(path):
    raw = open(path, encoding="utf-8", errors="replace").read()
    # split on --- lines or blank-line gaps of 1+
    chunks = re.split(r"\n\s*(?:-{3,}|\*{3,})\s*\n|\n{2,}", raw)
    return [c.strip() for c in chunks if c.strip()]


def compute_stats(posts):
    n = len(posts)
    def avg(f):
        return round(sum(f(p) for p in posts) / n, 2) if n else 0
    hashtags = [len(re.findall(r"#\w+", p)) for p in posts]
    return {
        "n_posts": n,
        "avg_chars": avg(len),
        "avg_lines": avg(lambda p: p.count("\n") + 1),
        "avg_words": avg(lambda p: len(p.split())),
        "pct_with_question": round(100 * sum(1 for p in posts if "?" in p) / n) if n else 0,
        "pct_with_emoji": round(100 * sum(1 for p in posts if EMOJI_RE.search(p)) / n) if n else 0,
        "pct_with_hashtag": round(100 * sum(1 for h in hashtags if h) / n) if n else 0,
        "avg_hashtags": round(sum(hashtags) / n, 2) if n else 0,
        "pct_with_list": round(100 * sum(1 for p in posts if re.search(r"(?m)^\s*(?:[-•*\d]|\d[.)])", p)) / n) if n else 0,
    }


def top_hooks(posts, k=12):
    """First line of the longest / most-engaging-looking posts = hook exemplars."""
    ranked = sorted(posts, key=len, reverse=True)
    hooks, seen = [], set()
    for p in ranked:
        first = p.strip().splitlines()[0].strip()
        if 8 <= len(first) <= 140 and first.lower() not in seen:
            hooks.append(first)
            seen.add(first.lower())
        if len(hooks) >= k:
            break
    return hooks


def synthesize(posts, stats, name):
    sample = "\n\n---\n\n".join(p[:600] for p in posts[:25])
    prompt = f"""You are a brand-voice analyst. Below are real LinkedIn posts written by {name}.
Study them and return ONLY a JSON object (no prose) with these keys:

  "core_themes":     array of 4-8 short topic labels this person posts about
  "voice_summary":   2-4 sentences describing their voice, tone, and register
                     (e.g. warm/direct, plain-spoken, wry, expert, first-person)
                     — specific enough to imitate, grounded in the samples.
  "signature_moves": array of 3-6 concrete recurring stylistic habits
                     (e.g. "opens with a one-line contrarian hook",
                      "uses short 1-2 line paragraphs with whitespace",
                      "ends with a soft question", "no hashtags").
  "avoid":           array of 2-5 things that would break the voice
                     (things they never do — corporate buzzwords, hashtag spam, etc.)

Measured style stats (respect these when describing habits): {json.dumps(stats)}

POSTS:
{sample}

Return the JSON object only."""
    try:
        raw, model = _ollama.ask(prompt, temperature=0.4, num_predict=900)
        data = _ollama.extract_json(raw)
        if isinstance(data, dict) and data.get("voice_summary"):
            data["_voice_source"] = f"ollama:{model} (local, $0)"
            return data
    except Exception as e:
        print(f"  (ollama synthesis failed: {e} — falling back to stats-only)", file=sys.stderr)
    # graceful fallback: build a stats-derived profile so the tool still works
    return {
        "core_themes": [],
        "voice_summary": (f"{name} writes ~{int(stats['avg_words'])}-word LinkedIn posts, "
                          f"{'often asking a question' if stats['pct_with_question'] > 40 else 'mostly declarative'}, "
                          f"{'emoji-friendly' if stats['pct_with_emoji'] > 40 else 'light on emoji'}, "
                          f"{'hashtag-light' if stats['avg_hashtags'] < 2 else 'hashtag-forward'}."),
        "signature_moves": [],
        "avoid": [],
        "_voice_source": "stats-only (ollama unavailable)",
    }


def main():
    ap = argparse.ArgumentParser()
    src = ap.add_mutually_exclusive_group(required=True)
    src.add_argument("--export", help="LinkedIn Shares.csv from Download-your-data")
    src.add_argument("--posts", help="plain .txt, posts split by blank line or ---")
    ap.add_argument("--name", required=True, help="profile name, e.g. steve")
    ap.add_argument("--max", type=int, default=200, help="cap posts analysed")
    args = ap.parse_args()

    posts = load_from_csv(args.export) if args.export else load_from_txt(args.posts)
    if not posts:
        sys.exit("No posts found in the input.")
    posts = posts[: args.max]
    print(f"Loaded {len(posts)} posts for '{args.name}'.")

    stats = compute_stats(posts)
    print("Computing voice profile via local Ollama ($0)...")
    synth = synthesize(posts, stats, args.name)

    profile = {
        "name": args.name,
        "source": os.path.basename(args.export or args.posts),
        "built_at": datetime.datetime.now().isoformat(timespec="seconds"),
        "stats": stats,
        "sample_hooks": top_hooks(posts),
        **synth,
    }
    os.makedirs(PROFILE_DIR, exist_ok=True)
    out = os.path.join(PROFILE_DIR, f"{args.name}.json")
    with open(out, "w") as f:
        json.dump(profile, f, indent=2, ensure_ascii=False)

    print(f"\n✅ Voice profile → {out}")
    print(f"   themes: {', '.join(profile.get('core_themes') or ['(none synthesised)'])}")
    print(f"   voice : {profile['voice_summary']}")
    print(f"   source: {profile['_voice_source']}")
    print(f"\nNext:  python3 scripts/generate.py --profile {args.name} --count 3")


if __name__ == "__main__":
    main()