mirror of
https://github.com/wahyd4/kb.git
synced 2026-08-08 21:06:27 +10:00
hermes
1f0201a80f
- 3-layer architecture: raw (monthly buckets) → wiki (domain subdirs) → SCHEMA - Privacy: repo set to private (public-safe content only) - Sync: daily Links App sync via raw/links/YYYY-MM/ - Design: converged via Grok 4.5 2-round review
311 lines
9.5 KiB
Python
Executable File
311 lines
9.5 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
Sync new posts from Links App to the wiki vault.
|
|
Run daily via cron to keep wiki in sync with Links App.
|
|
|
|
Usage:
|
|
python3 sync-links-to-wiki.py [--dry-run]
|
|
"""
|
|
|
|
import urllib.request
|
|
import json
|
|
import os
|
|
import sys
|
|
import re
|
|
import subprocess
|
|
import shutil
|
|
from datetime import datetime
|
|
|
|
# Config
|
|
KB_DIR = os.path.expanduser("~/Code/kb")
|
|
LINKS_API = "http://go/api/posts"
|
|
SYNC_STATE_FILE = os.path.join(KB_DIR, ".sync-state")
|
|
RAW_LINKS_BASE = os.path.join(KB_DIR, "raw", "links")
|
|
|
|
def run_cmd(cmd, cwd=None, capture=True):
|
|
"""Run a shell command and return result."""
|
|
result = subprocess.run(
|
|
cmd, shell=True, cwd=cwd or KB_DIR,
|
|
capture_output=capture, text=True
|
|
)
|
|
return result
|
|
|
|
def get_last_sync_time():
|
|
"""Get the last sync timestamp from state file."""
|
|
if os.path.exists(SYNC_STATE_FILE):
|
|
with open(SYNC_STATE_FILE, 'r') as f:
|
|
return f.read().strip()
|
|
return None
|
|
|
|
def save_last_sync_time(timestamp):
|
|
"""Save the sync timestamp to state file."""
|
|
with open(SYNC_STATE_FILE, 'w') as f:
|
|
f.write(timestamp)
|
|
|
|
def slugify(title, max_len=50):
|
|
"""Convert title to a valid filename slug."""
|
|
slug = re.sub(r'[^\w\s\u4e00-\u9fff-]', '', title)
|
|
slug = re.sub(r'[-\s]+', '-', slug)
|
|
slug = slug.lower().strip('-')[:max_len]
|
|
return slug or 'untitled'
|
|
|
|
def fetch_posts_since(since_timestamp=None, limit=100):
|
|
"""Fetch posts from Links App API, stop when we hit posts older than since_timestamp."""
|
|
posts = []
|
|
page = 1
|
|
stop = False
|
|
|
|
while True and not stop:
|
|
url = f"{LINKS_API}?limit={limit}&page={page}"
|
|
print(f"Fetching page {page}...", end=" ", flush=True)
|
|
|
|
try:
|
|
with urllib.request.urlopen(url, timeout=30) as resp:
|
|
data = json.loads(resp.read().decode('utf-8'))
|
|
except Exception as e:
|
|
print(f"Error: {e}")
|
|
break
|
|
|
|
results = data.get('results', [])
|
|
if not results:
|
|
print("no results, stopping")
|
|
break
|
|
|
|
if since_timestamp:
|
|
for post in results:
|
|
if post['created_at'] > since_timestamp:
|
|
posts.append(post)
|
|
else:
|
|
stop = True
|
|
print(f"reached cutoff at post {post['id']}")
|
|
break
|
|
else:
|
|
posts.extend(results)
|
|
|
|
if not data.get('next') or stop:
|
|
print("no more pages" if not stop else "")
|
|
break
|
|
|
|
page += 1
|
|
if page > 500:
|
|
print("safety limit reached")
|
|
break
|
|
|
|
return posts
|
|
|
|
def post_to_markdown(post):
|
|
"""Convert a post object to KB raw markdown."""
|
|
post_id = post['id']
|
|
title = post.get('title', 'Untitled')
|
|
content = post.get('content', post.get('summary', ''))
|
|
created = post.get('created_at', '')[:10]
|
|
updated = post.get('updated_at', '')[:10]
|
|
tags = [t['name'] for t in post.get('tag_details', [])]
|
|
external_url = f"http://go/ui/posts/{post_id}/"
|
|
|
|
safe_title = title.replace('"', '\\"')
|
|
frontmatter = f"""---
|
|
source_url: "{external_url}"
|
|
ingested: {datetime.now().strftime('%Y-%m-%d')}
|
|
status: inbox
|
|
tags: [{', '.join(tags)}]
|
|
title: "{safe_title}"
|
|
created: {created}
|
|
updated: {updated}
|
|
---
|
|
|
|
{content}"""
|
|
|
|
return frontmatter
|
|
|
|
def git_pull_with_rebase():
|
|
"""Pull with rebase to keep history clean, handle conflicts."""
|
|
print("🔄 Git pull (rebase)...")
|
|
|
|
# First check if there are remote changes
|
|
result = run_cmd("git fetch origin")
|
|
if result.returncode != 0:
|
|
print(f"⚠️ Fetch failed: {result.stderr}")
|
|
return False
|
|
|
|
# Check status
|
|
result = run_cmd("git status -sb")
|
|
branch_info = result.stdout.strip()
|
|
|
|
# Check if we need to pull
|
|
if "ahead" in branch_info and "behind" not in branch_info:
|
|
print("✅ No remote changes, nothing to pull")
|
|
return True
|
|
|
|
# Try rebase first
|
|
result = run_cmd("git rebase origin/master")
|
|
if result.returncode == 0:
|
|
print("✅ Rebase successful")
|
|
return True
|
|
|
|
# Conflict during rebase - handle it
|
|
print("⚠️ Conflict detected during rebase!")
|
|
print(f"Conflict output: {result.stdout}")
|
|
print(f"Conflict stderr: {result.stderr}")
|
|
|
|
# Check which files have conflicts
|
|
result = run_cmd("git diff --name-only --diff-filter=U")
|
|
conflicted_files = result.stdout.strip().split('\n') if result.stdout.strip() else []
|
|
|
|
if not conflicted_files or conflicted_files == ['']:
|
|
print("No conflicting files found, trying abort")
|
|
run_cmd("git rebase --abort")
|
|
return False
|
|
|
|
print(f"📁 Conflicted files: {conflicted_files}")
|
|
|
|
# Strategy: For our use case, we prefer wiki content
|
|
# So we'll use --ours for conflicts in links-posts/ and --theirs for others
|
|
for f in conflicted_files:
|
|
if not f:
|
|
continue
|
|
if f.startswith('links-posts/'):
|
|
# Prefer our new posts
|
|
print(f" → Keeping our version: {f}")
|
|
run_cmd(f"git checkout --ours {f}")
|
|
else:
|
|
# For other files, prefer their version (user might have edited on Mac)
|
|
print(f" → Keeping their version: {f}")
|
|
run_cmd(f"git checkout --theirs {f}")
|
|
|
|
# Stage resolved files
|
|
run_cmd("git add -A")
|
|
|
|
# Continue rebase
|
|
result = run_cmd("git rebase --continue")
|
|
if result.returncode == 0:
|
|
print("✅ Conflict resolved and rebase continued")
|
|
return True
|
|
else:
|
|
print(f"⚠️ Could not continue rebase: {result.stderr}")
|
|
# If still failing, abort and just overwrite
|
|
print("🔄 Aborting rebase, will use force push...")
|
|
run_cmd("git rebase --abort")
|
|
return False
|
|
|
|
def git_push():
|
|
"""Push to remote."""
|
|
print("📤 Pushing to GitHub...")
|
|
|
|
# Try normal push first
|
|
result = run_cmd("git push")
|
|
if result.returncode == 0:
|
|
print("✅ Push successful")
|
|
return True
|
|
|
|
# Push failed - might need force or pull first
|
|
print(f"⚠️ Push failed: {result.stderr}")
|
|
|
|
# Try pull first then push
|
|
if git_pull_with_rebase():
|
|
result = run_cmd("git push")
|
|
if result.returncode == 0:
|
|
print("✅ Push successful after pull")
|
|
return True
|
|
|
|
# Last resort: force push (dangerous but sometimes necessary)
|
|
print("⚠️ Force pushing...")
|
|
result = run_cmd("git push --force-with-lease")
|
|
if result.returncode == 0:
|
|
print("✅ Force push successful")
|
|
return True
|
|
|
|
print(f"❌ Push failed: {result.stderr}")
|
|
return False
|
|
|
|
def main():
|
|
dry_run = '--dry-run' in sys.argv
|
|
|
|
print(f"\n[{datetime.now().isoformat()}] 🚀 Starting Links App sync...")
|
|
os.chdir(KB_DIR)
|
|
|
|
# Git setup
|
|
run_cmd("git config user.name 'hermes'")
|
|
run_cmd("git config user.email 'hermes@junv'")
|
|
|
|
# Pull first to get any changes from Mac
|
|
if not dry_run:
|
|
print("\n--- Step 0: Sync with remote ---")
|
|
git_pull_with_rebase()
|
|
|
|
# Get last sync time
|
|
last_sync = get_last_sync_time()
|
|
print(f"\n--- Step 1: Fetch new posts ---")
|
|
print(f"Last sync: {last_sync or 'First sync'}")
|
|
|
|
# Fetch posts
|
|
posts = fetch_posts_since(last_sync)
|
|
print(f"Found {len(posts)} new posts")
|
|
|
|
if not posts:
|
|
print("No new posts to sync")
|
|
# Still do a push to sync any other changes
|
|
if not dry_run:
|
|
print("\n--- Step 3: Sync other changes ---")
|
|
git_push()
|
|
return
|
|
|
|
if dry_run:
|
|
print(f"\n[DRY RUN] Would sync {len(posts)} posts:")
|
|
for p in posts[:5]:
|
|
print(f" - {p['title']}")
|
|
if len(posts) > 5:
|
|
print(f" ... and {len(posts) - 5} more")
|
|
return
|
|
|
|
# Create posts directory with year-month subdirectory
|
|
month_dir = datetime.now().strftime("%Y-%m")
|
|
posts_dir = os.path.join(RAW_LINKS_BASE, month_dir)
|
|
os.makedirs(posts_dir, exist_ok=True)
|
|
|
|
# Write posts
|
|
newest_timestamp = None
|
|
print(f"\n--- Step 2: Write {len(posts)} posts to raw/links/{month_dir}/ ---")
|
|
for post in posts:
|
|
post_id = post['id']
|
|
title = post.get('title', 'Untitled')
|
|
slug = slugify(title)
|
|
filename = f"{post_id}-{slug}.md"
|
|
filepath = os.path.join(posts_dir, filename)
|
|
|
|
with open(filepath, 'w', encoding='utf-8') as f:
|
|
f.write(post_to_markdown(post))
|
|
|
|
created = post.get('created_at', '')
|
|
if not newest_timestamp or created > newest_timestamp:
|
|
newest_timestamp = created
|
|
|
|
print(f" ✅ {filename}")
|
|
|
|
# Update sync state
|
|
if newest_timestamp:
|
|
save_last_sync_time(newest_timestamp)
|
|
print(f"\nUpdated sync state to {newest_timestamp}")
|
|
|
|
# Git operations
|
|
print("\n--- Step 3: Commit and push ---")
|
|
run_cmd("git add raw/links/ .sync-state")
|
|
|
|
result = run_cmd("git status --porcelain")
|
|
if result.stdout.strip():
|
|
commit_msg = f"Sync: {len(posts)} new posts from Links App ({datetime.now().strftime('%Y-%m-%d %H:%M')})"
|
|
run_cmd(f"git commit -m '{commit_msg}'")
|
|
print(f"📝 Committed: {commit_msg}")
|
|
|
|
if git_push():
|
|
print(f"\n✅ SUCCESS: Synced {len(posts)} posts!")
|
|
else:
|
|
print(f"\n❌ Push failed - posts are saved locally")
|
|
else:
|
|
print("No changes to commit")
|
|
|
|
print(f"\n[{datetime.now().isoformat()}] 🎉 Sync complete!")
|
|
|
|
if __name__ == "__main__":
|
|
main()
|