#!/usr/bin/env python3 """Build vs market balance per product. Build hours come from git: commits by you, grouped into work sessions (a gap of more than `session_gap_hours` starts a new one; each session counts its span plus `session_pad_hours`). Commits that only touch marketing paths (blog, landing pages, README, images...) count as marketing, not build. Market hours come from Wingspan: published posts and drafts per brand, at a fixed number of hours each. Commits in repos that belong to no product show up as "other": time spent off the products. Usage: balance.py [--days N] [--json] """ import argparse, json, os, subprocess, sys from datetime import datetime, timedelta, timezone HERE = os.path.dirname(os.path.abspath(__file__)) CODE = os.path.expanduser("~/Code") DEFAULTS = { "target_build_pct": 20, "author_match": "", "session_gap_hours": 2, "session_pad_hours": 0.5, "hours_per_post": {"published": 0.75, "draft": 0.4}, "marketing_paths": ["blog/", "landing/", "marketing/", "content/", "posts/", "docs/", "index.html", "pricing.html", "sitemap.xml", "robots.txt", "README.md", "og/", "img/"], "projects": {}, "code_dir": "~/Code", } def load_cfg(): global CODE cfg = dict(DEFAULTS) try: cfg.update(json.load(open(os.path.join(HERE, "config.json")))) except FileNotFoundError: pass CODE = os.path.expanduser(os.environ.get("BALANCE_CODE_DIR") or cfg["code_dir"]) if not os.path.isdir(CODE): sys.exit(f"Balance looks for your git repos in {CODE}, and that folder doesn't exist.\n" f"Point it at the folder that holds your repos:\n" f" BALANCE_CODE_DIR=~/projects python3 {os.path.abspath(__file__)}\n" f"or set \"code_dir\" in {os.path.join(HERE, 'config.json')}") if not cfg["author_match"]: cfg["author_match"] = subprocess.run(["git", "config", "--global", "user.name"], capture_output=True, text=True).stdout.strip() if not cfg["projects"]: # No product list yet: every repo you committed to is its own row. cfg["projects"] = {r: {"repos": [r]} for r in sorted(os.listdir(CODE)) if os.path.isdir(os.path.join(CODE, r, ".git"))} cfg["auto"] = True return cfg def is_marketing(path, cfg): p = path.lower() return any(p.startswith(m) or ("/" + m) in p or p.endswith(m) for m in cfg["marketing_paths"]) def commits(repo, since, cfg): """[(datetime, kind)] for this repo's commits by the author since `since`.""" path = os.path.join(CODE, repo) if not os.path.isdir(os.path.join(path, ".git")): return [] try: out = subprocess.run( ["git", "-C", path, "log", "--all", f"--since={since.isoformat()}", f"--author={cfg['author_match']}", "--no-merges", "--pretty=format:@@%aI", "--name-only"], capture_output=True, text=True, timeout=60).stdout except Exception: return [] res, cur, files = [], None, [] def flush(): if cur is None: return kind = "market" if files and all(is_marketing(f, cfg) for f in files) else "build" res.append((cur, kind)) for line in out.splitlines(): if line.startswith("@@"): flush() cur, files = datetime.fromisoformat(line[2:]), [] elif line.strip(): files.append(line.strip()) flush() return res def hours(events, cfg): """Split session hours by commit kind. events: [(dt, kind)].""" out = {"build": 0.0, "market": 0.0} if not events: return out events = sorted(events) gap = timedelta(hours=cfg["session_gap_hours"]) sessions, cur = [], [events[0]] for e in events[1:]: if e[0] - cur[-1][0] > gap: sessions.append(cur) cur = [e] else: cur.append(e) sessions.append(cur) for s in sessions: h = (s[-1][0] - s[0][0]).total_seconds() / 3600 + cfg["session_pad_hours"] for _, kind in s: out[kind] += h / len(s) return out def wingspan_posts(brand, since): """(published, drafts) in the window for a Wingspan brand.""" try: raw = subprocess.run(["wingspan", "posts", brand], capture_output=True, text=True, timeout=60, stdin=subprocess.DEVNULL).stdout d = json.loads(raw) except Exception: return None posts = d.get("posts", d) if isinstance(d, dict) else d pub = drafts = 0 for p in posts: when = None for r in (p.get("results") or {}).values(): if r.get("publishedAt"): when = r["publishedAt"] break when = when or p.get("scheduled") if not when and len(p.get("id", "")) >= 10 and p["id"][:4].isdigit(): when = p["id"][:10] if not when: continue try: dt = datetime.fromisoformat(when.replace("Z", "+00:00")) except ValueError: continue if dt.tzinfo is None: dt = dt.replace(tzinfo=timezone.utc) if dt < since: continue if p.get("status") == "published": pub += 1 elif p.get("status") in ("draft", "scheduled"): drafts += 1 return pub, drafts def main(): ap = argparse.ArgumentParser() ap.add_argument("--days", type=int, default=14) ap.add_argument("--json", action="store_true") a = ap.parse_args() cfg = load_cfg() since = datetime.now(timezone.utc) - timedelta(days=a.days) target = cfg["target_build_pct"] rows, owned = [], set() for name, p in cfg["projects"].items(): if cfg.get("auto") and not commits(p["repos"][0], since, cfg): continue ev = [] for r in p.get("repos", []): owned.add(r) ev += commits(r, since, cfg) for r in p.get("marketing_repos", []): owned.add(r) ev += [(dt, "market") for dt, _ in commits(r, since, cfg)] h = hours(ev, cfg) wp = wingspan_posts(p["brand"], since) if p.get("brand") else None pub, drafts = wp if wp else (0, 0) post_h = pub * cfg["hours_per_post"]["published"] + drafts * cfg["hours_per_post"]["draft"] build, market = h["build"], h["market"] + post_h total = build + market rows.append({ "project": name, "build_h": round(build, 1), "market_h": round(market, 1), "build_pct": round(100 * build / total) if total else None, "commits": len(ev), "posts_published": pub, "posts_drafted": drafts, "wingspan_ok": wp is not None, }) # Everything else in ~/Code: time spent off the products. other = [] for r in sorted(os.listdir(CODE)): if r in owned: continue ev = commits(r, since, cfg) if ev: other.append((r, ev)) other_h = sum(sum(hours(ev, cfg).values()) for _, ev in other) top_other = sorted(((r, round(sum(hours(ev, cfg).values()), 1)) for r, ev in other), key=lambda x: -x[1])[:5] tb = sum(r["build_h"] for r in rows) tm = sum(r["market_h"] for r in rows) result = { "days": a.days, "target_build_pct": target, "projects": rows, "product_build_h": round(tb, 1), "product_market_h": round(tm, 1), "product_build_pct": round(100 * tb / (tb + tm)) if tb + tm else None, "other_h": round(other_h, 1), "other_top": top_other, } if a.json: print(json.dumps(result, indent=1)) return print(f"Build vs market, last {a.days} days (target: {target}% build)\n") W = max([11] + [len(r["project"]) + 2 for r in rows]) print(f"{'PROJECT':<{W}}{'Build h':>9}{'Market h':>10}{'Build %':>9}{'Commits':>9}{'Posts':>7}{'Drafts':>8} verdict") for r in sorted(rows, key=lambda x: -(x["build_h"] + x["market_h"])): pct = r["build_pct"] if pct is None: v = "idle" elif pct > target + 15: v = "over-building → market" elif pct < max(target - 15, 0): v = "under-building" else: v = "balanced" if not r["wingspan_ok"] and cfg["projects"].get(r["project"], {}).get("brand"): v += " (wingspan unreachable)" print(f"{r['project']:<{W}}{r['build_h']:>9}{r['market_h']:>10}{(str(pct)+'%') if pct is not None else '—':>9}" f"{r['commits']:>9}{r['posts_published']:>7}{r['posts_drafted']:>8} {v}") pp = result["product_build_pct"] print(f"\n{'PRODUCTS':<{W}}{tb:>9.1f}{tm:>10.1f}{(str(pp)+'%') if pp is not None else '—':>9}") print(f"{'OTHER':<{W}}{other_h:>9.1f}{'':>10}{'':>9} off-product repos: " + ", ".join(f"{r} {h}h" for r, h in top_other)) if __name__ == "__main__": main()