1  #!/usr/bin/env python3
  2  """
  3  Gravity Topology Tracker - Map the gravitational structure of the graph
  4  
  5  Implements the gravity well concept from Sovereign_OS:
  6  - Pages that accumulate more links gain "mass"
  7  - High-mass pages are gravity wells - concepts with strong pull
  8  - The topology emerges from use, not from design (Hayek)
  9  
 10  Usage:
 11      python scripts/gravity_tracker.py [--output FILE] [--top N] [--format FORMAT]
 12  
 13  Outputs:
 14      - Ranked list of pages by gravitational mass
 15      - Cluster analysis (what's in each gravity well's orbit)
 16      - Topology changes over time (if historical data exists)
 17  """
 18  
 19  import os
 20  import re
 21  import sys
 22  import json
 23  from collections import defaultdict
 24  from pathlib import Path
 25  from datetime import datetime
 26  from typing import Dict, List, Set, Tuple
 27  import argparse
 28  
 29  SCAN_DIRS = ['docs', 'patterns', 'sessions', 'dashboards']
 30  SKIP_DIRS = {'.git', 'node_modules', '__pycache__', 'templates'}
 31  SKIP_FILES = {'README.md', 'CHANGELOG.md'}
 32  
 33  
 34  def normalize_link(link: str) -> str:
 35      """Normalize wiki-link to comparable format."""
 36      name = link.split('/')[-1].lower()
 37      if name.endswith('.md'):
 38          name = name[:-3]
 39      return name.replace(' ', '-')
 40  
 41  
 42  def file_to_link_name(filepath: Path) -> str:
 43      """Convert filepath to wiki-link name."""
 44      return normalize_link(filepath.stem)
 45  
 46  
 47  def extract_wiki_links(content: str) -> Set[str]:
 48      """Extract all [[wiki-links]] from content."""
 49      pattern = r'\[\[([^\]|]+)(?:\|[^\]]+)?\]\]'
 50      return {normalize_link(m) for m in re.findall(pattern, content)}
 51  
 52  
 53  def extract_title(content: str) -> str:
 54      """Extract page title."""
 55      match = re.search(r'^# (.+)$', content, re.MULTILINE)
 56      return match.group(1) if match else "Untitled"
 57  
 58  
 59  def scan_graph(base_path: Path) -> Dict[str, dict]:
 60      """Scan the graph and extract all pages and links."""
 61      pages = {}
 62  
 63      for scan_dir in SCAN_DIRS:
 64          dir_path = base_path / scan_dir
 65          if not dir_path.exists():
 66              continue
 67  
 68          for filepath in dir_path.rglob('*.md'):
 69              if any(skip in filepath.parts for skip in SKIP_DIRS):
 70                  continue
 71              if filepath.name in SKIP_FILES:
 72                  continue
 73  
 74              try:
 75                  content = filepath.read_text(encoding='utf-8')
 76              except Exception:
 77                  continue
 78  
 79              link_name = file_to_link_name(filepath)
 80              pages[link_name] = {
 81                  'path': filepath,
 82                  'title': extract_title(content),
 83                  'outbound': extract_wiki_links(content),
 84              }
 85  
 86      return pages
 87  
 88  
 89  def calculate_gravity(pages: Dict[str, dict]) -> Dict[str, dict]:
 90      """
 91      Calculate gravitational mass for each page.
 92  
 93      Mass = f(inbound links, outbound links, link diversity)
 94  
 95      A page with many inbound links has high gravitational pull.
 96      This is emergent - mass comes from use, not assignment.
 97      """
 98      # Build inbound link map
 99      inbound: Dict[str, Set[str]] = defaultdict(set)
100      for source, data in pages.items():
101          for target in data['outbound']:
102              if target in pages:  # Only count links to existing pages
103                  inbound[target].add(source)
104  
105      # Calculate gravity for each page
106      gravity = {}
107      for name, data in pages.items():
108          inbound_count = len(inbound.get(name, set()))
109          outbound_count = len(data['outbound'])
110  
111          # Mass formula:
112          # - Inbound links are primary (others are pulled toward you)
113          # - Outbound links matter less (you reaching out)
114          # - Bidirectional links (mutual) count extra
115          bidirectional = sum(
116              1 for t in data['outbound']
117              if t in pages and name in pages.get(t, {}).get('outbound', set())
118          )
119  
120          mass = (inbound_count * 2.0) + (outbound_count * 0.5) + (bidirectional * 1.0)
121  
122          gravity[name] = {
123              'name': name,
124              'title': data['title'],
125              'path': str(data['path']),
126              'mass': mass,
127              'inbound': inbound_count,
128              'outbound': outbound_count,
129              'bidirectional': bidirectional,
130              'orbit': list(inbound.get(name, set()))[:10],  # Pages in orbit
131          }
132  
133      return gravity
134  
135  
136  def classify_wells(gravity: Dict[str, dict]) -> Dict[str, List[str]]:
137      """
138      Classify pages into gravity well tiers.
139  
140      - Black holes: Massive wells that everything links to
141      - Stars: Major concepts with strong pull
142      - Planets: Medium-mass pages
143      - Asteroids: Low-mass, potentially orphaned
144      """
145      masses = [g['mass'] for g in gravity.values()]
146      if not masses:
147          return {}
148  
149      max_mass = max(masses)
150      if max_mass == 0:
151          return {}
152  
153      tiers = {
154          'black_holes': [],   # > 0.8 of max
155          'stars': [],         # 0.4 - 0.8
156          'planets': [],       # 0.1 - 0.4
157          'asteroids': [],     # < 0.1
158      }
159  
160      for name, data in gravity.items():
161          ratio = data['mass'] / max_mass
162          if ratio > 0.8:
163              tiers['black_holes'].append(name)
164          elif ratio > 0.4:
165              tiers['stars'].append(name)
166          elif ratio > 0.1:
167              tiers['planets'].append(name)
168          else:
169              tiers['asteroids'].append(name)
170  
171      return tiers
172  
173  
174  def print_report(gravity: Dict[str, dict], tiers: Dict[str, List[str]], top_n: int = 20):
175      """Print the gravity topology report."""
176      print("=" * 70)
177      print("GRAVITY TOPOLOGY - Sovereign_OS")
178      print("=" * 70)
179      print()
180      print(f"Total pages analyzed: {len(gravity)}")
181      print(f"Timestamp: {datetime.now().isoformat()}")
182      print()
183  
184      # Summary by tier
185      print("=" * 70)
186      print("GRAVITATIONAL CLASSIFICATION")
187      print("=" * 70)
188      print()
189      print(f"🕳️  Black Holes (maximum gravity): {len(tiers.get('black_holes', []))}")
190      print(f"⭐ Stars (high gravity): {len(tiers.get('stars', []))}")
191      print(f"🪐 Planets (medium gravity): {len(tiers.get('planets', []))}")
192      print(f"🪨 Asteroids (low gravity): {len(tiers.get('asteroids', []))}")
193      print()
194  
195      # Top gravity wells
196      sorted_pages = sorted(gravity.values(), key=lambda x: x['mass'], reverse=True)
197  
198      print("=" * 70)
199      print(f"TOP {top_n} GRAVITY WELLS")
200      print("=" * 70)
201      print()
202      print(f"{'Rank':<5} {'Mass':>7} {'In':>4} {'Out':>4} {'Bi':>3} Title")
203      print("-" * 70)
204  
205      for i, page in enumerate(sorted_pages[:top_n], 1):
206          tier_emoji = (
207              "🕳️ " if page['name'] in tiers.get('black_holes', [])
208              else "⭐" if page['name'] in tiers.get('stars', [])
209              else "🪐" if page['name'] in tiers.get('planets', [])
210              else "🪨"
211          )
212          title = page['title'][:40] + "..." if len(page['title']) > 40 else page['title']
213          print(f"{i:<5} {page['mass']:>7.1f} {page['inbound']:>4} {page['outbound']:>4} {page['bidirectional']:>3} {tier_emoji} {title}")
214  
215      print()
216  
217      # Black holes detail
218      if tiers.get('black_holes'):
219          print("=" * 70)
220          print("🕳️  BLACK HOLES (maximum gravitational pull)")
221          print("=" * 70)
222          print()
223          for name in tiers['black_holes']:
224              data = gravity[name]
225              print(f"  {data['title']}")
226              print(f"    Path: {data['path']}")
227              print(f"    Mass: {data['mass']:.1f} | Inbound: {data['inbound']} | Bidirectional: {data['bidirectional']}")
228              if data['orbit']:
229                  print(f"    Orbit: {', '.join(data['orbit'][:5])}{'...' if len(data['orbit']) > 5 else ''}")
230              print()
231  
232      # Asteroids warning
233      asteroid_count = len(tiers.get('asteroids', []))
234      if asteroid_count > 0:
235          print("=" * 70)
236          print(f"⚠️  ASTEROIDS ({asteroid_count} pages with low gravity)")
237          print("   These pages may drift away without more connections.")
238          print("   Run: python3 scripts/resonance_engine.py --fix-orphans")
239          print("=" * 70)
240  
241  
242  def save_topology(
243      gravity: Dict[str, dict],
244      tiers: Dict[str, List[str]],
245      output_file: Path
246  ):
247      """Save topology to a file for tracking over time."""
248      topology = {
249          'timestamp': datetime.now().isoformat(),
250          'stats': {
251              'total_pages': len(gravity),
252              'black_holes': len(tiers.get('black_holes', [])),
253              'stars': len(tiers.get('stars', [])),
254              'planets': len(tiers.get('planets', [])),
255              'asteroids': len(tiers.get('asteroids', [])),
256          },
257          'top_10': [
258              {'name': g['name'], 'title': g['title'], 'mass': g['mass']}
259              for g in sorted(gravity.values(), key=lambda x: x['mass'], reverse=True)[:10]
260          ],
261          'tiers': tiers,
262      }
263  
264      output_file.write_text(json.dumps(topology, indent=2), encoding='utf-8')
265      print(f"\nTopology saved to: {output_file}")
266  
267  
268  def save_markdown(
269      gravity: Dict[str, dict],
270      tiers: Dict[str, List[str]],
271      output_file: Path
272  ):
273      """Save topology as a markdown file for the graph."""
274      sorted_pages = sorted(gravity.values(), key=lambda x: x['mass'], reverse=True)
275  
276      lines = [
277          "# Gravity Topology",
278          "",
279          f"*Generated: {datetime.now().strftime('%Y-%m-%d %H:%M')}*",
280          "",
281          "---",
282          "",
283          "## Summary",
284          "",
285          f"- **Total pages**: {len(gravity)}",
286          f"- **Black holes**: {len(tiers.get('black_holes', []))}",
287          f"- **Stars**: {len(tiers.get('stars', []))}",
288          f"- **Planets**: {len(tiers.get('planets', []))}",
289          f"- **Asteroids**: {len(tiers.get('asteroids', []))}",
290          "",
291          "---",
292          "",
293          "## Top Gravity Wells",
294          "",
295          "| Rank | Mass | Title |",
296          "|------|------|-------|",
297      ]
298  
299      for i, page in enumerate(sorted_pages[:20], 1):
300          tier = (
301              "🕳️" if page['name'] in tiers.get('black_holes', [])
302              else "⭐" if page['name'] in tiers.get('stars', [])
303              else "🪐" if page['name'] in tiers.get('planets', [])
304              else "🪨"
305          )
306          lines.append(f"| {i} | {page['mass']:.0f} | {tier} [[{page['name']}\\|{page['title']}]] |")
307  
308      lines.extend([
309          "",
310          "---",
311          "",
312          "## Black Holes",
313          "",
314          "Pages with maximum gravitational pull:",
315          "",
316      ])
317  
318      for name in tiers.get('black_holes', []):
319          data = gravity[name]
320          lines.append(f"### [[{name}|{data['title']}]]")
321          lines.append(f"- Mass: {data['mass']:.0f} | Inbound: {data['inbound']} | Bidirectional: {data['bidirectional']}")
322          if data['orbit']:
323              orbit_links = ', '.join(f"[[{o}]]" for o in data['orbit'][:5])
324              lines.append(f"- Orbit: {orbit_links}")
325          lines.append("")
326  
327      lines.extend([
328          "---",
329          "",
330          "*Gravity emerges from use, not assignment (Hayek)*",
331      ])
332  
333      output_file.write_text('\n'.join(lines), encoding='utf-8')
334      print(f"\nMarkdown saved to: {output_file}")
335  
336  
337  def main():
338      parser = argparse.ArgumentParser(
339          description='Gravity Topology Tracker - Map the gravitational structure of the graph'
340      )
341      parser.add_argument(
342          '--output', '-o',
343          type=Path,
344          help='Output file for topology data (JSON)'
345      )
346      parser.add_argument(
347          '--markdown', '-m',
348          type=Path,
349          help='Output file for markdown topology'
350      )
351      parser.add_argument(
352          '--top', '-t',
353          type=int,
354          default=20,
355          help='Number of top pages to show (default: 20)'
356      )
357      parser.add_argument(
358          '--path', '-p',
359          type=Path,
360          default=Path(__file__).parent.parent,
361          help='Path to Sovereign_OS repo'
362      )
363  
364      args = parser.parse_args()
365  
366      print(f"Scanning graph from {args.path}...\n")
367      pages = scan_graph(args.path)
368  
369      if not pages:
370          print("No pages found!", file=sys.stderr)
371          return 1
372  
373      gravity = calculate_gravity(pages)
374      tiers = classify_wells(gravity)
375  
376      print_report(gravity, tiers, args.top)
377  
378      if args.output:
379          save_topology(gravity, tiers, args.output)
380  
381      if args.markdown:
382          save_markdown(gravity, tiers, args.markdown)
383  
384      return 0
385  
386  
387  if __name__ == '__main__':
388      sys.exit(main())