|
1
|
"""Skill: Generate GitHub issues from task breakdown artifacts.""" |
|
2
|
|
|
3
|
import json |
|
4
|
import shutil |
|
5
|
import subprocess |
|
6
|
from typing import List, Optional |
|
7
|
|
|
8
|
from video_processor.agent.skills.base import AgentContext, Artifact, Skill, register_skill |
|
9
|
|
|
10
|
|
|
11
|
def _task_to_issue(task: dict) -> dict: |
|
12
|
"""Convert a task dict to a GitHub issue object.""" |
|
13
|
deps = task.get("dependencies", []) |
|
14
|
body_parts = [ |
|
15
|
f"## Description\n{task.get('description', task.get('title', ''))}", |
|
16
|
f"**Priority:** {task.get('priority', 'medium')}", |
|
17
|
f"**Estimate:** {task.get('estimate', 'unknown')}", |
|
18
|
] |
|
19
|
if deps: |
|
20
|
body_parts.append(f"**Dependencies:** {', '.join(str(d) for d in deps)}") |
|
21
|
labels = [task.get("priority", "medium")] |
|
22
|
if task.get("labels"): |
|
23
|
labels.extend(task["labels"]) |
|
24
|
return { |
|
25
|
"title": task.get("title", "Untitled task"), |
|
26
|
"body": "\n\n".join(body_parts), |
|
27
|
"labels": labels, |
|
28
|
} |
|
29
|
|
|
30
|
|
|
31
|
def push_to_github(issues_json: str, repo: str) -> Optional[List[dict]]: |
|
32
|
"""Shell out to `gh issue create` for each issue. Returns None if gh unavailable.""" |
|
33
|
if not shutil.which("gh"): |
|
34
|
return None |
|
35
|
issues = json.loads(issues_json) |
|
36
|
results = [] |
|
37
|
for issue in issues: |
|
38
|
cmd = [ |
|
39
|
"gh", |
|
40
|
"issue", |
|
41
|
"create", |
|
42
|
"--repo", |
|
43
|
repo, |
|
44
|
"--title", |
|
45
|
issue["title"], |
|
46
|
"--body", |
|
47
|
issue["body"], |
|
48
|
] |
|
49
|
for label in issue.get("labels", []): |
|
50
|
cmd.extend(["--label", label]) |
|
51
|
proc = subprocess.run(cmd, capture_output=True, text=True) |
|
52
|
results.append( |
|
53
|
{ |
|
54
|
"title": issue["title"], |
|
55
|
"returncode": proc.returncode, |
|
56
|
"stdout": proc.stdout.strip(), |
|
57
|
"stderr": proc.stderr.strip(), |
|
58
|
} |
|
59
|
) |
|
60
|
return results |
|
61
|
|
|
62
|
|
|
63
|
class GitHubIssuesSkill(Skill): |
|
64
|
name = "github_issues" |
|
65
|
description = "Generate GitHub issues from task breakdown" |
|
66
|
|
|
67
|
def execute(self, context: AgentContext, **kwargs) -> Artifact: |
|
68
|
task_artifact = next((a for a in context.artifacts if a.artifact_type == "task_list"), None) |
|
69
|
if task_artifact: |
|
70
|
tasks = json.loads(task_artifact.content) |
|
71
|
else: |
|
72
|
# Generate minimal task list inline from planning entities |
|
73
|
tasks = [ |
|
74
|
{ |
|
75
|
"title": str(e), |
|
76
|
"description": str(e), |
|
77
|
"priority": "medium", |
|
78
|
"estimate": "unknown", |
|
79
|
} |
|
80
|
for e in context.planning_entities |
|
81
|
] |
|
82
|
|
|
83
|
issues = [_task_to_issue(t) for t in tasks] |
|
84
|
content = json.dumps(issues, indent=2) |
|
85
|
return Artifact( |
|
86
|
name="GitHub Issues", |
|
87
|
content=content, |
|
88
|
artifact_type="issues", |
|
89
|
format="json", |
|
90
|
) |
|
91
|
|
|
92
|
|
|
93
|
register_skill(GitHubIssuesSkill()) |
|
94
|
|