|
1 | | -from codegen import Codebase, ProgrammingLanguage |
2 | | -from typing import List, Dict, Any |
3 | | -from codegen.configs.models.codebase import CodebaseConfig |
4 | | -from data import LinearLabels, LinearIssueUpdateEvent |
5 | | -import os |
6 | | -import logging |
7 | | - |
8 | | - |
9 | | -logging.basicConfig(level=logging.INFO) |
10 | | -logger = logging.getLogger(__name__) |
11 | | - |
| 1 | +from codegen import Codebase |
| 2 | +from codegen.extensions.clients.linear import LinearClient |
| 3 | +from codegen.shared.enums.programming_language import ProgrammingLanguage |
| 4 | +from codegen.extensions.linear.types import LinearEvent, LinearIssue, LinearComment, LinearUser, LinearLabel |
12 | 5 |
|
13 | | -def process_update_event(event_data: dict[str, Any]): |
14 | | - print("processing update event") |
15 | | - |
16 | | - action = event_data.get("action") |
17 | | - actor = event_data.get("actor") |
18 | | - created_at = event_data.get("createdAt") |
19 | | - issue_url = event_data.get("url") |
20 | | - data: Dict[str, Any] = event_data.get("data") |
21 | | - issue_id = data.get("id") |
22 | | - title = data.get("title") |
23 | | - description = data.get("description") |
24 | | - identifier = data.get("identifier") |
25 | | - |
26 | | - labels: List[LinearLabels] = data.get("labels") |
27 | | - updated_from: Dict[str, Any] = event_data.get("updatedFrom") |
28 | | - |
29 | | - update_event = LinearIssueUpdateEvent( |
| 6 | +from typing import Any, Dict, List, Optional, cast |
| 7 | +import os |
| 8 | +from pydantic import BaseModel |
| 9 | + |
| 10 | + |
| 11 | +class LinearIssueUpdateEvent(BaseModel): |
| 12 | + issue_id: str |
| 13 | + issue_url: str |
| 14 | + title: str |
| 15 | + description: Optional[str] = None |
| 16 | + identifier: Optional[str] = None |
| 17 | + |
| 18 | + |
| 19 | +def process_update_event(data: Any) -> LinearIssueUpdateEvent: |
| 20 | + """Process a Linear webhook event and extract issue information.""" |
| 21 | + # Extract issue data from the event |
| 22 | + issue_data = cast(Dict[str, Any], data.get("data", {})) |
| 23 | + |
| 24 | + # Extract issue ID and URL |
| 25 | + issue_id = issue_data.get("id", "") |
| 26 | + issue_url = issue_data.get("url", "") |
| 27 | + |
| 28 | + # Extract labels |
| 29 | + labels = cast(List[LinearLabel], issue_data.get("labels", [])) |
| 30 | + team = cast(Dict[str, Any], issue_data.get("team", {})) |
| 31 | + |
| 32 | + # Extract issue title, description, and identifier |
| 33 | + title = issue_data.get("title", "") |
| 34 | + description = issue_data.get("description", "") |
| 35 | + identifier = f"{team.get('key', '')}-{issue_data.get('number', '')}" |
| 36 | + |
| 37 | + # Create and return the event object |
| 38 | + return LinearIssueUpdateEvent( |
30 | 39 | issue_id=issue_id, |
31 | | - action=action, |
32 | | - actor=actor, |
33 | | - created_at=created_at, |
34 | 40 | issue_url=issue_url, |
35 | | - data=data, |
36 | | - labels=labels, |
37 | | - updated_from=updated_from, |
38 | 41 | title=title, |
39 | 42 | description=description, |
40 | 43 | identifier=identifier, |
41 | 44 | ) |
42 | | - return update_event |
43 | 45 |
|
44 | 46 |
|
45 | | -def format_linear_message(title: str, description: str | None = "") -> str: |
46 | | - """Format a Linear update event into a message for the agent""" |
| 47 | +def has_codegen_label(data: dict) -> bool: |
| 48 | + """Check if the issue has the 'Codegen' label.""" |
| 49 | + issue_data = data.get("data", {}) |
| 50 | + labels = issue_data.get("labels", []) |
| 51 | + |
| 52 | + return any(label.get("name") == "Codegen" for label in labels) |
47 | 53 |
|
48 | | - return f""" |
49 | | - Here is a new issue titled '{title}' and with the description '{description}'. Continue to respond to this query. Use your tools to query the codebase for more context. |
50 | | - When applicable include references to files and line numbers, code snippets are also encouraged. Don't forget to create a pull request with your changes, use the appropriate tool to do so. |
51 | | - """ |
52 | 54 |
|
| 55 | +def format_linear_message(title: str, description: Optional[str]) -> str: |
| 56 | + """Format a Linear issue title and description into a message for the agent.""" |
| 57 | + message = f"Create a PR that implements: {title}" |
| 58 | + |
| 59 | + if description: |
| 60 | + message += f"\n\nDetails:\n{description}" |
| 61 | + |
| 62 | + return message |
53 | 63 |
|
54 | | -def has_codegen_label(*args, **kwargs): |
55 | | - body = kwargs.get("data") |
56 | | - type = body.get("type") |
57 | | - action = body.get("action") |
58 | 64 |
|
59 | | - if type == "Issue" and action == "update": |
60 | | - # handle issue update (label updates) |
61 | | - update_event = process_update_event(body) |
62 | | - |
63 | | - has_codegen_label = any(label.name == "Codegen" for label in update_event.labels) |
64 | | - codegen_label_id = next((label.id for label in update_event.labels if label.name == "Codegen"), None) |
65 | | - had_codegen_label = codegen_label_id in update_event.updated_from.get("labels", []) if codegen_label_id else False |
66 | | - previous_labels = update_event.updated_from.get("labelIds", None) |
67 | | - |
68 | | - if previous_labels is None or not has_codegen_label: |
69 | | - logger.info("No labels updated, skipping codegen bot response") |
70 | | - return False |
71 | | - |
72 | | - if has_codegen_label and not had_codegen_label: |
73 | | - logger.info("Codegen label added, codegen bot will respond") |
74 | | - return True |
75 | | - |
76 | | - logger.info("Codegen label removed or already existed, codegen bot will not respond") |
77 | | - return False |
78 | | - |
79 | | - |
80 | | -def create_codebase(repo_name: str, language: ProgrammingLanguage): |
81 | | - config = CodebaseConfig() |
| 65 | +def create_codebase(repo_name: str, language: ProgrammingLanguage) -> Codebase: |
| 66 | + """Create a Codebase instance for the specified repository.""" |
| 67 | + from codegen.config import Config |
| 68 | + |
| 69 | + config = Config() |
82 | 70 | config.secrets.github_token = os.environ["GITHUB_TOKEN"] |
83 | | - |
| 71 | + |
84 | 72 | return Codebase.from_repo(repo_name, language=language, tmp_dir="/root", config=config) |
85 | 73 |
|
0 commit comments