mirror of
https://github.com/langgenius/dify-plugin-sdks.git
synced 2026-07-24 21:15:37 -04:00
Harry
46b1ef471a
- Removed the deprecated IssuesTrigger class and replaced it with a more focused IssueOpenedTrigger class to handle specific events when a new issue is opened. - Updated the GitHub trigger YAML configuration to reflect changes in the subscription schema and parameters. - Enhanced the validation logic for issue parameters, including title patterns, labels, assignees, authors, milestones, and body content. - Improved the overall structure and clarity of the trigger implementation, ensuring better maintainability and functionality. These changes contribute to a more efficient and organized handling of GitHub issue events, enhancing the plugin's responsiveness to user interactions.
129 lines
4.6 KiB
Python
129 lines
4.6 KiB
Python
from collections.abc import Mapping
|
|
import re
|
|
from typing import Any
|
|
|
|
from dify_plugin.errors.trigger import TriggerIgnoreEventError
|
|
from werkzeug import Request
|
|
|
|
from dify_plugin.entities.trigger import Event
|
|
from dify_plugin.interfaces.trigger import TriggerEvent
|
|
|
|
|
|
class IssueOpenedTrigger(TriggerEvent):
|
|
"""
|
|
GitHub Issue Opened Event Trigger
|
|
|
|
This trigger handles GitHub issue opened events and extracts relevant
|
|
information from the webhook payload to provide as variables to the workflow.
|
|
"""
|
|
|
|
def _check_title_pattern(self, issue: dict, pattern: str) -> None:
|
|
"""Check if issue title matches the pattern"""
|
|
if not pattern:
|
|
return
|
|
|
|
title = issue.get("title", "")
|
|
if not re.match(pattern, title):
|
|
raise TriggerIgnoreEventError()
|
|
|
|
def _check_labels(self, issue: dict, labels_param: str) -> None:
|
|
"""Check if issue has required labels"""
|
|
if not labels_param:
|
|
return
|
|
|
|
required_labels = [label.strip() for label in labels_param.split(",") if label.strip()]
|
|
if not required_labels:
|
|
return
|
|
|
|
issue_labels = [label.get("name") for label in issue.get("labels", [])]
|
|
if not any(label in issue_labels for label in required_labels):
|
|
raise TriggerIgnoreEventError()
|
|
|
|
def _check_assignee(self, issue: dict, assignee_param: str) -> None:
|
|
"""Check if issue is assigned to allowed users"""
|
|
if not assignee_param:
|
|
return
|
|
|
|
allowed_assignees = [assignee.strip() for assignee in assignee_param.split(",") if assignee.strip()]
|
|
if not allowed_assignees:
|
|
return
|
|
|
|
issue_assignees = []
|
|
|
|
# Collect assignee usernames
|
|
single_assignee = issue.get("assignee")
|
|
if single_assignee and single_assignee.get("login"):
|
|
issue_assignees.append(single_assignee["login"])
|
|
|
|
for assignee in issue.get("assignees", []):
|
|
login = assignee.get("login")
|
|
if login:
|
|
issue_assignees.append(login)
|
|
|
|
if not any(assignee in issue_assignees for assignee in allowed_assignees):
|
|
raise TriggerIgnoreEventError()
|
|
|
|
def _check_authors(self, issue: dict, authors_param: str) -> None:
|
|
"""Check if issue author is in allowed list"""
|
|
if not authors_param:
|
|
return
|
|
|
|
allowed_authors = [author.strip() for author in authors_param.split(",") if author.strip()]
|
|
if not allowed_authors:
|
|
return
|
|
|
|
issue_author = issue.get("user", {}).get("login")
|
|
if issue_author not in allowed_authors:
|
|
raise TriggerIgnoreEventError()
|
|
|
|
def _check_milestone(self, issue: dict, milestone_param: str) -> None:
|
|
"""Check if issue milestone matches allowed milestones"""
|
|
if not milestone_param:
|
|
return
|
|
|
|
allowed_milestones = [milestone.strip() for milestone in milestone_param.split(",") if milestone.strip()]
|
|
if not allowed_milestones:
|
|
return
|
|
|
|
milestone = issue.get("milestone")
|
|
if not milestone:
|
|
raise TriggerIgnoreEventError()
|
|
|
|
milestone_title = milestone.get("title")
|
|
if not milestone_title or milestone_title not in allowed_milestones:
|
|
raise TriggerIgnoreEventError()
|
|
|
|
def _check_body_contains(self, issue: dict, body_contains_param: str) -> None:
|
|
"""Check if issue body contains required keywords"""
|
|
if not body_contains_param:
|
|
return
|
|
|
|
keywords = [keyword.strip().lower() for keyword in body_contains_param.split(",") if keyword.strip()]
|
|
if not keywords:
|
|
return
|
|
|
|
issue_body = (issue.get("body") or "").lower()
|
|
if not any(keyword in issue_body for keyword in keywords):
|
|
raise TriggerIgnoreEventError()
|
|
|
|
def _trigger(self, request: Request, parameters: Mapping[str, Any]) -> Event:
|
|
"""
|
|
Handle GitHub issue opened event trigger
|
|
"""
|
|
payload = request.get_json()
|
|
if not payload:
|
|
raise ValueError("No payload received")
|
|
|
|
issue = payload.get("issue")
|
|
if not issue:
|
|
raise ValueError("No issue data in payload")
|
|
|
|
# Apply all filters
|
|
self._check_title_pattern(issue, parameters.get("title_pattern"))
|
|
self._check_labels(issue, parameters.get("labels"))
|
|
self._check_assignee(issue, parameters.get("assignee"))
|
|
self._check_authors(issue, parameters.get("authors"))
|
|
self._check_milestone(issue, parameters.get("milestone"))
|
|
self._check_body_contains(issue, parameters.get("body_contains"))
|
|
|
|
return Event(variables={**payload}) |