rootiest
071544e2d2
Some checks failed
Sync Beads to Gitea Issues / sync-beads (push) Failing after 6s
122 lines
4.2 KiB
Python
122 lines
4.2 KiB
Python
import json
|
||
import os
|
||
import requests
|
||
import sys
|
||
|
||
# Configuration from Environment
|
||
TOKEN = os.getenv("MASTER_TOKEN")
|
||
URL = os.getenv("GITEA_URL")
|
||
REPO = os.getenv("REPO_NAME")
|
||
HEADERS = {"Authorization": f"token {TOKEN}", "Content-Type": "application/json"}
|
||
|
||
|
||
def sync():
|
||
beads_path = ".beads/issues.jsonl"
|
||
|
||
if not os.path.exists(beads_path):
|
||
print(f"❌ ERROR: {beads_path} not found.")
|
||
sys.exit(1)
|
||
|
||
print(f"🔍 Reading Beads from: {beads_path}")
|
||
|
||
# 1. Fetch existing issues from Gitea
|
||
try:
|
||
api_url = f"{URL}/api/v1/repos/{REPO}/issues?state=all"
|
||
resp = requests.get(api_url, headers=HEADERS)
|
||
resp.raise_for_status()
|
||
# Map by [ID] to identify existing ones
|
||
existing = {
|
||
i["title"].split("]")[0][1:]: i for i in resp.json() if "]" in i["title"]
|
||
}
|
||
print(f"📡 Found {len(existing)} existing issues in Gitea.")
|
||
except Exception as e:
|
||
print(f"❌ Gitea API Connection Failed: {e}")
|
||
sys.exit(1)
|
||
|
||
# 2. Parse the Beads JSONL file
|
||
processed_count = 0
|
||
skipped_count = 0
|
||
deleted_count = 0
|
||
|
||
with open(beads_path, "r") as f:
|
||
for line in f:
|
||
line = line.strip()
|
||
if not line:
|
||
continue
|
||
|
||
try:
|
||
data = json.loads(line)
|
||
bid = data.get("id")
|
||
status = data.get("status")
|
||
title = data.get("title")
|
||
itype = data.get("issue_type", "unknown")
|
||
|
||
if not bid:
|
||
print("⚠️ Skipping line: No ID found.")
|
||
continue
|
||
|
||
# --- DELETION / TOMBSTONE LOGIC ---
|
||
if status == "tombstone":
|
||
if bid in existing:
|
||
issue_num = existing[bid]["number"]
|
||
print(
|
||
f"🗑️ Deleting Gitea Issue #{issue_num} (Tombstone found for {bid})"
|
||
)
|
||
del_resp = requests.delete(
|
||
f"{URL}/api/v1/repos/{REPO}/issues/{issue_num}",
|
||
headers=HEADERS,
|
||
)
|
||
if del_resp.status_code == 204:
|
||
print(f"✅ Successfully deleted {bid}")
|
||
deleted_count += 1
|
||
else:
|
||
print(f"⚠️ Could not delete {bid}: {del_resp.status_code}")
|
||
else:
|
||
print(f"👻 Skipping tombstone: {bid} (Already gone from Gitea)")
|
||
skipped_count += 1
|
||
continue
|
||
|
||
# --- STANDARD SYNC LOGIC ---
|
||
processed_count += 1
|
||
gitea_state = (
|
||
"closed" if status in ["closed", "done", "finished"] else "open"
|
||
)
|
||
|
||
payload = {
|
||
"title": f"[{bid}] {title}",
|
||
"body": f"{data.get('description', 'No description provided.')}\n\n---\n**Beads ID:** `{bid}`\n**Type:** `{itype}`",
|
||
"state": gitea_state,
|
||
}
|
||
|
||
if bid in existing:
|
||
print(f"✅ Updating {bid} (Issue #{existing[bid]['number']})")
|
||
requests.patch(
|
||
f"{URL}/api/v1/repos/{REPO}/issues/{existing[bid]['number']}",
|
||
headers=HEADERS,
|
||
json=payload,
|
||
)
|
||
else:
|
||
print(f"➕ Creating New Gitea Issue for {bid}")
|
||
r = requests.post(
|
||
f"{URL}/api/v1/repos/{REPO}/issues",
|
||
headers=HEADERS,
|
||
json=payload,
|
||
)
|
||
r.raise_for_status()
|
||
|
||
except Exception as e:
|
||
print(f"⚠️ Error processing bead line: {e}")
|
||
|
||
print(
|
||
f"🏁 Finished. Active: {processed_count}, Deleted: {deleted_count}, Skipped: {skipped_count}"
|
||
)
|
||
|
||
# Safety check to ensure we didn't process a completely empty file
|
||
if (processed_count + deleted_count + skipped_count) == 0:
|
||
print("❌ ERROR: File was found but it was EMPTY.")
|
||
sys.exit(1)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sync()
|