Stephan Kerkman
70c2213d98
Idempotent sync via notion-id marker in issue body. Initial 10 issues already created from BloomBase Notion workspace. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
115 lines
3.6 KiB
Python
115 lines
3.6 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Sync BloomBase Notion feature tracker → Forgejo issues.
|
|
|
|
Notion is source of truth. Runs one-way: Notion → Forgejo.
|
|
Idempotent: tracks Notion page IDs in issue body to avoid duplicates.
|
|
|
|
Usage:
|
|
FORGEJO_TOKEN=<token> python3 scripts/sync_notion_to_forgejo.py
|
|
(Notion access via Claude MCP — run from a Claude Code session instead)
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
import urllib.request
|
|
import urllib.parse
|
|
|
|
FORGEJO_BASE = "https://git.kerkman.io/api/v1"
|
|
FORGEJO_REPO = "stephan/bloombase"
|
|
NOTION_MARKER = "notion-id:"
|
|
|
|
LABEL_MAP = {
|
|
"Aesthetics": 1,
|
|
"Functionality": 2,
|
|
"Bug": 3,
|
|
}
|
|
|
|
STATUS_MAP = {
|
|
"Open": "open",
|
|
"In progress": "open", # + "In progress" label (id=4)
|
|
"Done": "closed",
|
|
}
|
|
|
|
IN_PROGRESS_LABEL_ID = 4
|
|
|
|
|
|
def forgejo(method, path, data=None):
|
|
token = os.environ["FORGEJO_TOKEN"]
|
|
url = f"{FORGEJO_BASE}{path}"
|
|
body = json.dumps(data).encode() if data else None
|
|
req = urllib.request.Request(
|
|
url, data=body, method=method,
|
|
headers={
|
|
"Authorization": f"token {token}",
|
|
"Content-Type": "application/json",
|
|
}
|
|
)
|
|
with urllib.request.urlopen(req) as r:
|
|
return json.loads(r.read())
|
|
|
|
|
|
def get_existing_issues():
|
|
"""Return dict of notion_page_id → issue number for already-synced issues."""
|
|
issues = forgejo("GET", f"/repos/{FORGEJO_REPO}/issues?type=issues&state=open&limit=50")
|
|
issues += forgejo("GET", f"/repos/{FORGEJO_REPO}/issues?type=issues&state=closed&limit=50")
|
|
mapping = {}
|
|
for issue in issues:
|
|
body = issue.get("body") or ""
|
|
m = re.search(rf"{NOTION_MARKER}(\S+)", body)
|
|
if m:
|
|
mapping[m.group(1)] = issue["number"]
|
|
return mapping
|
|
|
|
|
|
def sync_item(item, existing):
|
|
"""Create or update a single Forgejo issue from a Notion item dict."""
|
|
notion_id = item["id"]
|
|
title = item["title"].strip()
|
|
category = item.get("category")
|
|
status = item.get("status", "Open")
|
|
notion_url = item["url"]
|
|
has_screenshot = item.get("has_screenshot", False)
|
|
|
|
labels = []
|
|
if category and category in LABEL_MAP:
|
|
labels.append(LABEL_MAP[category])
|
|
if status == "In progress":
|
|
labels.append(IN_PROGRESS_LABEL_ID)
|
|
|
|
screenshot_note = "\n\n📎 Screenshot attached in Notion" if has_screenshot else ""
|
|
body = (
|
|
f"*Synced from [🌸 BloomBase Notion]({notion_url})*"
|
|
f"{screenshot_note}"
|
|
f"\n\n<!-- {NOTION_MARKER}{notion_id} -->"
|
|
)
|
|
|
|
state = STATUS_MAP.get(status, "open")
|
|
|
|
if notion_id in existing:
|
|
issue_num = existing[notion_id]
|
|
forgejo("PATCH", f"/repos/{FORGEJO_REPO}/issues/{issue_num}", {
|
|
"title": title,
|
|
"body": body,
|
|
"state": state,
|
|
"labels": labels,
|
|
})
|
|
print(f" updated #{issue_num}: {title[:60]}")
|
|
else:
|
|
issue = forgejo("POST", f"/repos/{FORGEJO_REPO}/issues", {
|
|
"title": title,
|
|
"body": body,
|
|
"labels": labels,
|
|
})
|
|
if state == "closed":
|
|
forgejo("PATCH", f"/repos/{FORGEJO_REPO}/issues/{issue['number']}", {"state": "closed"})
|
|
print(f" created #{issue['number']}: {title[:60]}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# Items passed in as JSON via stdin or hard-coded for manual runs.
|
|
# In practice: run from a Claude Code session using the Notion MCP,
|
|
# build the items list from MCP data, then call sync_item() for each.
|
|
print("Run this from a Claude Code session — Notion data comes via MCP.")
|
|
print("See: scripts/sync_notion_to_forgejo.py sync_item() for the API contract.")
|