nandi.uk
This repository has no description
3.9 kB
140 lines
1#!/usr/bin/env python3
2"""Analyze each island with letta CLI and post summaries back to the API."""
3
4import json
5import subprocess
6import sys
7import time
8import urllib.parse
9import urllib.request
10
11API_BASE = "https://stigmergic.latha.org"
12LETTA_BIN = "/home/nandi/.local/npm-global/bin/letta"
13
14
15def fetch_islands():
16 req = urllib.request.Request(
17 f"{API_BASE}/api/strata/islands",
18 headers={"User-Agent": "stigmergic-analyzer/1.0"},
19 )
20 with urllib.request.urlopen(req) as resp:
21 return json.loads(resp.read())["islands"]
22
23
24def build_island_context(island):
25 vertices = island["vertices"]
26 edges = island["edges"]
27 meta = island.get("vertexMeta", {})
28
29 titles = [meta.get(v, {}).get("title", v) for v in vertices[:30]]
30
31 domains = {}
32 for v in vertices:
33 if v.startswith("http"):
34 try:
35 from urllib.parse import urlparse
36 h = urlparse(v).hostname.replace("www.", "")
37 domains[h] = domains.get(h, 0) + 1
38 except Exception:
39 pass
40 top_domains = sorted(domains.items(), key=lambda x: -x[1])[:5]
41
42 edge_types = {}
43 for e in edges:
44 t = e.get("connectionType", "unknown")
45 edge_types[t] = edge_types.get(t, 0) + 1
46
47 skip_patterns = (
48 "english",
49 "japanese",
50 "version",
51 "translation",
52 "original",
53 "alternate url",
54 "DID link",
55 )
56 notes = [
57 e["note"]
58 for e in edges
59 if e.get("note") and not e["note"].lower().startswith(skip_patterns)
60 ][:5]
61
62 return {
63 "vertex_count": len(vertices),
64 "edge_count": len(edges),
65 "top_domains": [f"{d[0]} ({d[1]})" for d in top_domains],
66 "edge_types": edge_types,
67 "sample_titles": titles[:15],
68 "notes": notes,
69 }
70
71
72def summarize_with_letta(context):
73 prompt = (
74 "Write a single concise sentence (max 200 chars) summarizing what this "
75 "cluster of links is about. No preamble, just the summary.\n\n"
76 f"Island data:\n{json.dumps(context, indent=2)}"
77 )
78 try:
79 result = subprocess.run(
80 [LETTA_BIN, "-p", prompt, "--output-format", "json", "--system", "letta", "--memfs"],
81 capture_output=True,
82 text=True,
83 timeout=60,
84 )
85 except subprocess.TimeoutExpired:
86 return None
87 if result.returncode != 0:
88 return None
89 try:
90 data = json.loads(result.stdout)
91 if data.get("type") == "result":
92 return data["result"].strip().strip('"').strip()[:300]
93 except Exception:
94 pass
95 return result.stdout.strip()[:300] or None
96
97
98def put_summary(island_id, summary):
99 encoded_id = urllib.parse.quote(island_id, safe="")
100 body = json.dumps({"summary": summary}).encode()
101 req = urllib.request.Request(
102 f"{API_BASE}/api/strata/islands/{encoded_id}/summary",
103 data=body,
104 method="PUT",
105 headers={"Content-Type": "application/json", "User-Agent": "stigmergic-analyzer/1.0"},
106 )
107 with urllib.request.urlopen(req) as resp:
108 return resp.status == 200
109
110
111def main():
112 islands = fetch_islands()
113 total = len(islands)
114 print(f"Analyzing {total} islands...")
115
116 for i, island in enumerate(islands):
117 island_id = island["id"]
118 if island.get("summary"):
119 print(f"[{i+1}/{total}] Skip (has summary): {island_id[:60]}...")
120 continue
121
122 print(f"[{i+1}/{total}] Analyzing: {island_id[:60]}...")
123 context = build_island_context(island)
124 summary = summarize_with_letta(context)
125
126 if not summary:
127 print(f" Skipping (no summary generated)")
128 continue
129
130 print(f" Summary: {summary}")
131 put_summary(island_id, summary)
132 print(f" Saved.")
133
134 time.sleep(1)
135
136 print("Done.")
137
138
139if __name__ == "__main__":
140 main()