openclaw/tools/kanban_reddit_cleanup.py

#!/usr/bin/env python3
"""Kanban cleanup for imported Reddit ideas.

Actions:
1) Cluster reddit idea tasks into epics (creates new epic tasks)
2) Tag each original with cluster tags
3) Mark near-duplicates within a cluster (tag duplicate + move to waiting)

Heuristic clustering based on title/description keywords.

Usage:
  kanban_reddit_cleanup.py --tag reddit --create-epics --dedupe --prepare

Notes:
- Uses the Kanban HTTP API directly for speed.
- Avoids deletes.
"""

from __future__ import annotations

import argparse
import json
import re
import sys
import time
import urllib.request
from dataclasses import dataclass
from typing import Any


API = "http://localhost:5003/api/tasks"
KEY = "openclaw-tasks-2026"


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


@dataclass
class Task:
    id: int
    title: str
    description: str
    status: str
    tags: str


def parse_task(d: dict[str, Any]) -> Task:
    return Task(
        id=int(d["id"]),
        title=(d.get("title") or ""),
        description=(d.get("description") or ""),
        status=(d.get("status") or ""),
        tags=(d.get("tags") or "") or "",
    )


THEMES: list[tuple[str, str, list[str]]] = [
    (
        "cost-models",
        "Cost, rate limits, model selection",
        [
            r"cheap",
            r"cost",
            r"pricing",
            r"tokens?",
            r"quota",
            r"rate\s*limit",
            r"model",
            r"glm",
            r"deepseek",
            r"minimax",
            r"kimi",
            r"local\s*llm",
        ],
    ),
    (
        "onboarding",
        "Onboarding, setup, beginner docs",
        [r"beginner", r"101", r"setup", r"start", r"new\s*to", r"tutorial", r"install"],
    ),
    (
        "updates",
        "Updating/reinstall/upgrade flow",
        [r"update", r"upgrade", r"reinstall", r"deinstall", r"install", r"version"],
    ),
    (
        "security",
        "Security, malware, sandboxing, runtime defense",
        [r"secur", r"malware", r"comprom", r"zero\s*trust", r"microvm", r"antivirus", r"banned", r"tos"],
    ),
    (
        "context-memory",
        "Context window, memory, transparency",
        [r"context", r"forget", r"memory", r"window", r"included\s*in\s*the\s*context"],
    ),
    (
        "observability",
        "Tooling, debugging, observability",
        [r"tool\s*call", r"viewer", r"debug", r"trace", r"logs?", r"metrics"],
    ),
    (
        "channels",
        "Messaging channels reliability/config",
        [r"telegram", r"whatsapp", r"signal", r"discord", r"slack", r"imessage"],
    ),
    (
        "voice",
        "Voice, latency, calls",
        [r"voice", r"latency", r"call", r"tts"],
    ),
    (
        "showcase-hardware",
        "Showcases: wearables/robots/hardware",
        [r"ray-?ban", r"glasses", r"robot", r"legs", r"physical", r"body", r"3d\s*printer"],
    ),
    (
        "other",
        "Other / uncategorized",
        [],
    ),
]


def classify(t: Task) -> str:
    text = f"{t.title}\n{t.description}".lower()
    for key, _, pats in THEMES:
        if not pats:
            continue
        if any(re.search(p, text) for p in pats):
            return key
    return "other"


def norm_title(s: str) -> str:
    s = s.lower()
    s = re.sub(r"\(reddit/[^)]+\):\s*", "", s)
    s = re.sub(r"[^a-z0-9]+", " ", s)
    s = re.sub(r"\s+", " ", s).strip()
    return s


def main(argv: list[str]) -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--tag", default="reddit", help="filter tasks whose tags contain this")
    ap.add_argument("--create-epics", action="store_true")
    ap.add_argument("--dedupe", action="store_true")
    ap.add_argument("--prepare", action="store_true", help="write start-ready epic descriptions")
    ap.add_argument("--sleep", type=float, default=0.05)
    args = ap.parse_args(argv[1:])

    tasks_raw = http_json("GET", API)
    tasks = [parse_task(d) for d in tasks_raw]
    reddit = [t for t in tasks if args.tag in (t.tags or "")]

    buckets: dict[str, list[Task]] = {}
    for t in reddit:
        buckets.setdefault(classify(t), []).append(t)

    # Dedupe within bucket by normalized title
    dup_map: dict[int, int] = {}  # dup_id -> keep_id
    if args.dedupe:
        for k, items in buckets.items():
            seen: dict[str, int] = {}
            for t in sorted(items, key=lambda x: x.id):
                nt = norm_title(t.title)
                if nt in seen:
                    dup_map[t.id] = seen[nt]
                else:
                    seen[nt] = t.id

        # Tag dupes and move to waiting
        for dup_id, keep_id in dup_map.items():
            http_json(
                "PATCH",
                f"{API}/{dup_id}",
                {"tags": f"duplicate,keep:{keep_id}"},
            )
            http_json("PATCH", f"{API}/{dup_id}", {"status": "waiting"})
            time.sleep(args.sleep)

    created_epics: dict[str, int] = {}
    if args.create_epics:
        for key, label, _ in THEMES:
            items = buckets.get(key, [])
            if not items:
                continue

            # Don't count duplicates
            items2 = [t for t in items if t.id not in dup_map]

            top = sorted(items2, key=lambda x: x.id)[:40]
            lines = [
                f"Cluster: {label}",
                "",
                "Source: Reddit deep dive (top/month)",
                "",
                "Included tasks:",
            ]
            for t in top:
                lines.append(f"- [{t.id}] {t.title}")
            if len(items2) > len(top):
                lines.append(f"- … +{len(items2) - len(top)} more")

            if args.prepare:
                lines += [
                    "",
                    "Start-ready plan:",
                    "1) Read 5–10 representative threads in this cluster.",
                    "2) Write a 1-page problem statement + target user flows.",
                    "3) Propose 2–3 solution options (MVP vs full), with risks.",
                    "4) Convert to 3–8 implementation tasks + docs changes.",
                    "",
                    "Acceptance criteria:",
                    "- Concrete proposal (docs/PR plan) exists",
                    "- At least 1 shippable MVP path identified",
                ]

            epic = http_json(
                "POST",
                API,
                {
                    "title": f"Epic: Reddit ideas — {label}",
                    "description": "\n".join(lines),
                    "status": "queue",
                },
            )
            epic_id = int(epic.get("id"))
            created_epics[key] = epic_id

            # Tag epic
            http_json(
                "PATCH",
                f"{API}/{epic_id}",
                {"tags": f"epic,idea,openclaw,reddit,cluster:{key}"},
            )
            time.sleep(args.sleep)

    # Tag originals with cluster tag and raw marker
    for k, items in buckets.items():
        for t in items:
            # skip duplicates already retagged
            if t.id in dup_map:
                continue
            tags = t.tags or ""
            # normalize to avoid repeated tags
            tagset = {x.strip() for x in tags.split(",") if x.strip()}
            tagset.add("raw")
            tagset.add("idea")
            tagset.add("reddit")
            tagset.add(f"cluster:{k}")
            # Keep existing subreddit tag if present
            new_tags = ",".join(sorted(tagset))
            http_json("PATCH", f"{API}/{t.id}", {"tags": new_tags})
            time.sleep(args.sleep)

    print(
        json.dumps(
            {
                "reddit_tasks": len(reddit),
                "clusters": {k: len(v) for k, v in buckets.items() if v},
                "duplicates_marked": len(dup_map),
                "created_epics": created_epics,
            },
            indent=2,
        )
    )

    return 0


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