openclaw/tools/github_to_kanban.py

#!/usr/bin/env python3
"""Import GitHub OpenClaw repos into Kanban as idea tasks.

Usage:
  github_to_kanban.py /tmp/github-openclaw.jsonl --min-stars 1
"""

from __future__ import annotations

import argparse
import json
import sys
import time
import urllib.request
from collections import defaultdict


KANBAN_API = "http://localhost:5003/api/tasks"
KANBAN_KEY = "openclaw-tasks-2026"


def http_json(method: str, url: str, payload: dict | None = None) -> dict:
    data = None
    headers = {
        "User-Agent": "openclaw-github-to-kanban/1.0",
    }
    if payload is not None:
        raw = json.dumps(payload).encode("utf-8")
        data = raw
        headers["Content-Type"] = "application/json"
        headers["X-API-Key"] = KANBAN_KEY
    req = urllib.request.Request(url, data=data, method=method, headers=headers)
    with urllib.request.urlopen(req, timeout=25) as resp:
        return json.loads(resp.read().decode("utf-8"))


def get_existing_github_repos() -> set[str]:
    """Get set of GitHub repos already in Kanban."""
    tasks = http_json("GET", KANBAN_API)
    repos: set[str] = set()
    for t in tasks:
        title = (t.get("title") or "").lower()
        desc = (t.get("description") or "")
        
        # Check title for "GitHub: owner/repo" pattern
        if title.startswith("github:"):
            repo_name = title.replace("github:", "").strip()
            repos.add(repo_name)
        
        # Look for GitHub URLs in description
        for line in desc.split("\n"):
            if "github.com/" in line:
                # Extract repo full name
                parts = line.split("github.com/")
                if len(parts) > 1:
                    repo_path = parts[1].split()[0].split("/")[:2]
                    if len(repo_path) == 2:
                        full_name = "/".join(repo_path).rstrip(").,;")
                        repos.add(full_name.lower())
    return repos


def main(argv: list[str]) -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("jsonl", help="path to GitHub scanner JSONL output")
    ap.add_argument("--min-stars", type=int, default=1, help="minimum stars")
    ap.add_argument("--dry-run", action="store_true", help="show what would be imported")
    ap.add_argument("--sleep", type=float, default=0.2, help="sleep between API calls")
    args = ap.parse_args(argv[1:])

    existing = get_existing_github_repos()
    print(f"# Found {len(existing)} GitHub repos already in Kanban", file=sys.stderr)

    created_ids: list[int] = []
    skipped_count = 0

    with open(args.jsonl, "r", encoding="utf-8") as f:
        for line in f:
            d = json.loads(line)
            repo = d.get("repo", "")
            url = d.get("url", "")
            description = d.get("description") or "no description"
            stars = d.get("stars", 0)
            language = d.get("language") or "unknown"

            if not repo or not url:
                continue

            if stars < args.min_stars:
                skipped_count += 1
                if args.dry_run:
                    print(f"[SKIP] {repo} ({stars}⭐ < {args.min_stars})", file=sys.stderr)
                continue

            # Skip if already imported
            if repo.lower() in existing:
                skipped_count += 1
                if args.dry_run:
                    print(f"[SKIP] {repo} (already in Kanban)", file=sys.stderr)
                continue

            task_title = f"GitHub: {repo}"
            if len(task_title) > 160:
                task_title = task_title[:157] + "…"

            desc_text = (
                f"**GitHub OpenClaw Build**\n\n"
                f"Repo: {url}\n"
                f"Stars: {stars} ⭐\n"
                f"Language: {language}\n\n"
                f"Description: {description}\n\n"
                f"Review this build and decide if it's useful for our OpenClaw setup."
            )

            if args.dry_run:
                print(f"[KEEP] {repo} ({stars}⭐)", file=sys.stderr)
                created_ids.append(0)
                continue

            task = http_json(
                "POST",
                KANBAN_API,
                {
                    "title": task_title,
                    "description": desc_text,
                    "status": "queue",
                },
            )
            task_id = int(task.get("id"))

            # Tag as GitHub build
            tags = f"idea,openclaw,github,build"
            if stars >= 10:
                tags += ",popular"

            http_json(
                "PATCH",
                f"{KANBAN_API}/{task_id}",
                {"tags": tags},
            )

            created_ids.append(task_id)
            time.sleep(args.sleep)

    result = {
        "dry_run": args.dry_run,
        "created": len(created_ids),
        "skipped": skipped_count,
    }

    if not args.dry_run:
        result["created_ids"] = created_ids[:20]

    print(json.dumps(result, indent=2))
    return 0


if __name__ == "__main__":
    raise SystemExit(main(sys.argv))