#!/usr/bin/env python3 """ Aria Linear Liveness-Scan (Paperclip-pattern adoption). Scans the KAR Linear team for open issues that have no clear next action (no assignee + no blocker + last update > 14 days) and pings Kais on Telegram. Triggered daily via systemd timer aria-linear-liveness.timer. """ from __future__ import annotations import json import os import sys from dataclasses import dataclass from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Optional from urllib import request as urlrequest from urllib.error import HTTPError, URLError LINEAR_API = "https://api.linear.app/graphql" TELEGRAM_API = "https://api.telegram.org/bot{token}/sendMessage" KAIS_CHAT_ID = "1164395546" KAR_TEAM_ID = "e0227fa8-5ce8-4d3a-a8c5-fdc90ba8637f" STALE_DAYS = 14 HTTP_TIMEOUT_S = 20 ACTIVE_STATE_TYPES = ["backlog", "unstarted", "started"] def load_env(path: str) -> dict[str, str]: """Read a `KEY=VALUE` env-file (no shell expansion). Missing file → empty dict.""" out: dict[str, str] = {} p = Path(path) if not p.is_file(): return out for raw in p.read_text(encoding="utf-8").splitlines(): line = raw.strip() if not line or line.startswith("#") or "=" not in line: continue key, _, value = line.partition("=") out[key.strip()] = value.strip().strip('"').strip("'") return out @dataclass class Issue: identifier: str title: str url: str priority: int updated_at: datetime def age_days(self, now: datetime) -> int: return (now - self.updated_at).days def fetch_kar_open_unassigned_issues(api_key: str) -> list[Issue]: """Query Linear for open KAR-team issues without an assignee. Active states in Linear are state-type backlog | unstarted | started. Closed types (completed | canceled) are excluded server-side. """ query = """ query OpenUnassignedKarIssues($teamId: String!, $stateTypes: [String!]!) { team(id: $teamId) { issues( filter: { state: { type: { in: $stateTypes } } assignee: { null: true } } first: 100 orderBy: updatedAt ) { nodes { identifier title url priority updatedAt } } } } """ payload = json.dumps({ "query": query, "variables": { "teamId": KAR_TEAM_ID, "stateTypes": ACTIVE_STATE_TYPES, }, }).encode("utf-8") req = urlrequest.Request( LINEAR_API, data=payload, headers={ "Authorization": api_key, "Content-Type": "application/json", }, method="POST", ) with urlrequest.urlopen(req, timeout=HTTP_TIMEOUT_S) as resp: body = json.loads(resp.read().decode("utf-8")) if "errors" in body: raise RuntimeError(f"Linear API errors: {body['errors']}") nodes = body["data"]["team"]["issues"]["nodes"] return [ Issue( identifier=n["identifier"], title=n["title"], url=n["url"], priority=n.get("priority") or 0, updated_at=datetime.fromisoformat(n["updatedAt"].replace("Z", "+00:00")), ) for n in nodes ] def filter_stale(issues: list[Issue], stale_days: int, now: datetime) -> list[Issue]: cutoff = now - timedelta(days=stale_days) return [i for i in issues if i.updated_at < cutoff] def md_escape(s: str) -> str: """Escape Telegram MarkdownV2 specials outside code blocks.""" specials = r"_*[]()~`>#+-=|{}.!" return "".join("\\" + c if c in specials else c for c in s) def format_message(stale: list[Issue], now: datetime, stale_days: int) -> str: if not stale: return "" lines = ["⚠️ *Linear\\-Liveness\\-Scan*", ""] lines.append(f"{len(stale)} unassigned KAR\\-Issues > {stale_days} Tage ohne Update:") lines.append("") for i in sorted(stale, key=lambda x: (-x.priority, x.updated_at)): prio_marker = {1: "🔴", 2: "🟠", 3: "🟡", 4: "🟢"}.get(i.priority, "⚪") title_safe = md_escape(i.title[:80]) age = i.age_days(now) lines.append(f"{prio_marker} [{md_escape(i.identifier)}] {title_safe} \\— {age}d") lines.append("") lines.append("_Vorschlag: assignen, blocken oder schließen\\._") return "\n".join(lines) def send_telegram(token: str, chat_id: str, text: str) -> None: payload = json.dumps({ "chat_id": chat_id, "text": text, "parse_mode": "MarkdownV2", "disable_web_page_preview": True, }).encode("utf-8") req = urlrequest.Request( TELEGRAM_API.format(token=token), data=payload, headers={"Content-Type": "application/json"}, method="POST", ) with urlrequest.urlopen(req, timeout=HTTP_TIMEOUT_S) as resp: body = json.loads(resp.read().decode("utf-8")) if not body.get("ok"): raise RuntimeError(f"Telegram send failed: {body}") def main(argv: list[str]) -> int: dry_run = "--dry-run" in argv stale_days = STALE_DAYS for arg in argv: if arg.startswith("--stale-days="): stale_days = int(arg.split("=", 1)[1]) aria_env = load_env("/root/aria/.env") tg_env = load_env("/root/.claude/channels/telegram/.env") linear_key = aria_env.get("LINEAR_API_KEY") tg_token = tg_env.get("TELEGRAM_BOT_TOKEN") or aria_env.get("TELEGRAM_BOT_TOKEN") if not linear_key: print("ERROR: LINEAR_API_KEY missing in /root/aria/.env", file=sys.stderr) return 2 if not tg_token and not dry_run: print("ERROR: TELEGRAM_BOT_TOKEN missing", file=sys.stderr) return 2 now = datetime.now(timezone.utc) try: issues = fetch_kar_open_unassigned_issues(linear_key) except (HTTPError, URLError, RuntimeError) as e: print(f"ERROR: Linear fetch failed: {e}", file=sys.stderr) return 3 stale = filter_stale(issues, stale_days, now) print(f"Linear KAR open + unassigned issues: {len(issues)}") print(f"Stale (> {stale_days}d, no assignee): {len(stale)}") for i in stale: print(f" [{i.identifier}] prio={i.priority} age={i.age_days(now)}d {i.title[:60]}") if not stale: return 0 msg = format_message(stale, now, stale_days) if dry_run: print("\n--- Telegram message (dry-run) ---") print(msg) return 0 try: send_telegram(tg_token, KAIS_CHAT_ID, msg) except (HTTPError, URLError, RuntimeError) as e: print(f"ERROR: Telegram send failed: {e}", file=sys.stderr) return 4 print("Telegram nudge sent.") return 0 if __name__ == "__main__": sys.exit(main(sys.argv[1:]))