Integration Functionの設定

import json
from handlers.GithubConstants import GithubConstants
import handlers.command_handler
from zcatalyst_cliq.bot_handler import (
    welcome_handler,
    message_handler,
    context_handler,
    mention_handler,
    menu_action_handler,
    webhook_handler,
    participation_handler,
    BotWelcomeHandlerRequest,
    BotMessageHandlerRequest,
    BotContextHandlerRequest,
    BotMentionHandlerRequest,
    BotMenuActionHandlerRequest,
    BotParticipationHandlerRequest,
    BotWebHookHandlerRequest,
    HandlerResponse
)
from zcatalyst_sdk.catalyst_app import CatalystApp
import logging
@welcome_handler
def welcome_handler_fn(req: BotWelcomeHandlerRequest, res: HandlerResponse, *args):
    res.text = f'hello {req.user.first_name}. Thank you for subscribing :smile:'
    return res
@message_handler
def msg_handler(req: BotMessageHandlerRequest, res: HandlerResponse, *args):
        msg = req.message
        text = ''
        if not msg:
            text = 'Please enable \'Message\' in bot settings'
        elif msg == "webhooktoken":
            btnobj = res.new_button()
            btnobj.type = "+"
            action = btnobj.new_action_object()
            action.type = "invoke.function"
            conform = action.new_confirm_object()
            conform.title = "Generate Webhooks for a GitLab Project"
            conform.description = "Connect to GitLab Projects from within Cliq"
            conform.input  = "user_webhook_token"
            Actdata = action.new_action_data_obj()
            Actdata.name = "authenticate"
            action.data = Actdata
            action.confirm = conform
            btnobj.action = action
            btnobj.label = "Create Webhook"
            res.add_button(btnobj)
            text = "Click on the token generation button below!"
        else:
            text = "Oops! Sorry, I'm not programmed yet to do this :sad:"
        res.text = text
        return res
@context_handler
def ctx_handler(req: BotContextHandlerRequest, res: HandlerResponse, *args):
    app: CatalystApp = args[0]
    if req.context_id == 'personal_details':
        answer = req.answers
        name = answer.name.text
        dept = answer.dept.text
        text = f'Nice! I have collected your info: \n*Name*: {name} \n*Department*: {dept}'
        if answer.cache.text == 'YES':
            try:
                default_segment = app.cache().segment()
                default_segment.put('Name', name)
                default_segment.put('Department', dept)
                text += '\nThis data is now available in Catalyst Cache\'s default segment.'
            except Exception as e:
                logging.error(e)
        res.set_text(text)
        return res
@mention_handler
def mention_handler(req: BotMentionHandlerRequest, res: HandlerResponse, *args):
    text = f"Hey *{req.user.first_name}*, thanks for mentioning me here. I'm from Catalyst city"
    res.set_text(text)
    return res
@menu_action_handler
def action(req: BotMenuActionHandlerRequest, res: HandlerResponse, *args):
    text = ''
    res.bot = res.new_bot_details(GithubConstants.BOT_NAME,text)
    if req.action_name == "Repos":
        reposArray = handlers.command_handler.getRepos()
        if len(reposArray) == 0:
            text = "There aren't are repos created yet."
        else:
             Slide = res.new_slide()
             text = "Here's a list of the *repositories*"
             Slide.type = 'table'
             Slide.title = "Repo details"
             headers = []
             headers.append('Name')
             headers.append('Private')
             headers.append('Open Issues')
             headers.append('Link')
             data = {
                 'headers':headers
             }
             rows = []
             for repo in reposArray:
                row = {
                    "Name": repo.get("name"),
                    "Private": "Yes" if repo.get("private") else "No",
                    "Open Issues": repo.get("open_issues_count"),
                    "Link": f"[Click here]({repo.get('html_url')})"
                }
                rows.append(row)
             data['rows'] = rows
             json_data = json.dumps(data)
             Slide.data = json_data
             res.add_slides(Slide)
    else:
        text = 'Menu action triggered :fist:'
    res.set_text(text)
    return res
@participation_handler
def participation(req: BotParticipationHandlerRequest, res: HandlerResponse, *args):
    text = ''
    if req.operation == 'added':
        text = 'Hi. Thanks for adding me to the channel :smile:'
    elif req.operation == 'removed':
        text = 'Bye-Bye :bye-bye:'
    else:
        text = 'I\'m too a participant of this chat :wink:'
    res.set_text(text)
    return res
@webhook_handler
def webhook_fn(req: BotWebHookHandlerRequest, res: HandlerResponse, *args):
    req_body: dict = json.loads(req.body)
    commitJson = req_body["commits"][0]
    res.new_message()
    res.text="A new commit has been pushed!"
    res.bot = res.new_bot_details(GithubConstants.BOT_NAME,img='')
    commitMsg = res.new_slide()
    commitMsg.type = "text"
    commitMsg.title = "Commit message"
    commitMsg.data = commitJson["message"]
    res.add_slides(commitMsg)
    details = res.new_slide()
    details.type = "label"
    details.title = "Details"
    dataArray = []
    commiter = {}
    commiter['Committer'] = commitJson["author"]["username"]
    dataArray.append(commiter)
    repoName = {}
    repoName['Repo Name'] = req_body["repository"]["name"]
    dataArray.append(repoName)
    timestamp = {}
    timestamp['Timestamp'] = commitJson["timestamp"]
    dataArray.append(timestamp)
    compare = {}
    compare["Compare"] = f"[Click here]({req_body['compare']})"
    dataArray.append(compare)
    details.data = json.dumps(dataArray)
    res.add_slides(details)
    return res

from typing import List
import json
from handlers.GithubConstants import GithubConstants
import zcatalyst_cliq.command_handler as command
import requests
from zcatalyst_cliq.command_handler import (
    execution_handler,
    suggestion_handler,
    CommandHandlerRequest,
    CommandSuggestion,
    HandlerResponse
)
@execution_handler
def executor(req: CommandHandlerRequest, res: HandlerResponse, *args):
    text = ''
    res.bot = res.new_bot_details(GithubConstants.BOT_NAME,img=text)
    cmd = req.name
    if cmd == 'commits':
        suggestions = req.selections
        if not suggestions:
            text = 'Please select a repo from the suggestions.'
        else:
            repoName = suggestions[0].title
            commitsArray = getCommits(repoName)
            if len(commitsArray) == 0:
                text = "There aren't any commits made yet."
            else:
                Slide = res.new_slide()
                text = f"Here's a list of the latest {GithubConstants.PER_PAGE} commits made to the repository *{repoName}*."
                Slide.type = "table"
                Slide.title = "Commit details"
                headers = []
                headers.append('Date')
                headers.append('Commit message')
                headers.append('Committed by')
                headers.append('Link')
                data = {
                    'headers':headers
                }
                rows = []
                for obj in commitsArray:
                    commit = obj.get("commit")
                    author = commit.get("author")
                    row = {
                        "Date": author.get("date"),
                        "Commit message": commit.get("message"),
                        "Committed by": author.get("name"),
                        "Link": f"[Click here]({obj.get('html_url')})"
                    }
                    rows.append(row)
                data["rows"] = rows
                json_data = json.dumps(data)
                Slide.data = json_data
                res.add_slides(Slide)
    elif cmd == 'issues':
        suggestions = req.selections
        if not suggestions:
            text = 'Please select a repo from the suggestions.'
        else:
            repoName = suggestions[0].title
            IssuesArray = getIssues(repoName)
            if len(IssuesArray) == 0:
                text = "There aren't any issues made yet."
            else:
                Slide = res.new_slide()
                text = f"Here's a list of the latest {GithubConstants.PER_PAGE} issues raised to the repository *{repoName}*."
                Slide.type = "table"
                Slide.title = "Issue details"
                headers = []
                headers.append('Created At')
                headers.append('Title')
                headers.append('Created By')
                headers.append('Link')
                data = {
                    'headers':headers
                }
                rows = []
                for issueObj in IssuesArray:
                     row = {
                        "Created At": issueObj.get("created_at"),
                        "Title": issueObj.get("title"),
                        "Created By": issueObj.get("user").get("login"),
                        "Link": f"[Click here]({issueObj.get('html_url')})"
                    }
                rows.append(row)
                data["rows"] = rows
                json_data = json.dumps(data)
                Slide.data = json_data
                res.add_slides(Slide)
    else:
        text = "Slash command executed"
    res.text = text
    return res
def getCommits(name):
    req = f"https://api.github.com/repos/{getUsername()}/{name}/commits?per_page={GithubConstants.PER_PAGE}"
    headers = {}
    headers['Authorization'] = f'Token {GithubConstants.PERSONAL_ACCESS_TOKEN}'
    response = requests.get(url=req,headers=headers)
    data = response.json()
    if response.status_code == 200:
        return data    
    else:
        return []
def getUsername():
    req = "https://api.github.com/user"
    headers = {}
    headers['Authorization'] = f'Token {GithubConstants.PERSONAL_ACCESS_TOKEN}'
    response = requests.get(url=req,headers=headers)
    data  = response.json()
    return data["login"]
def getRepos():
    req = "https://api.github.com/user/repos"
    headers = {}
    headers['Authorization'] = f'Token {GithubConstants.PERSONAL_ACCESS_TOKEN}'
    response = requests.get(url=req,headers=headers)
    data = response.json()
    return data
def getIssues(repoName):
    req = f"https://api.github.com/repos/{getUsername()}/{repoName}/issues?per_page={GithubConstants.PER_PAGE}"
    headers = {}
    headers['Authorization'] = f'Token {GithubConstants.PERSONAL_ACCESS_TOKEN}'
    response = requests.get(url=req,headers=headers)
    data = response.json()
    return (data)
@suggestion_handler
def suggester(req: CommandHandlerRequest, res: List[CommandSuggestion], *args):
    suggestionList = []
    reposArray = getRepos()
    repos = reposArray
    repoNames = []
    for repo in repos:
        name = repo["name"]
        repoNames.append(name)
    if req.name == "commits" or req.name == "issues":
        for name in repoNames:
            sugg= CommandSuggestion(title=name)
            suggestionList.append(sugg)
    return suggestionList

from datetime import datetime
from zcatalyst_cliq import function_handler
from zcatalyst_cliq.function_handler import (
    button_function_handler,
    form_submit_handler,
    form_change_handler,
    form_dynamic_field_handler,
    widget_button_handler,
    ButtonFunctionRequest,
    FormFunctionRequest,
    WidgetFunctionRequest,
    HandlerResponse,
    FormChangeResponse,
    FormDynamicFieldResponse,
    WidgetResponse
)
@button_function_handler
def button_fn_handler(req: ButtonFunctionRequest, res: HandlerResponse, *args):
    text = ''
    if req.name == 'authenticate':
        text = req.arguments.input
    else:
        text = "Button function executed"
    res.text = text
    return res
@form_submit_handler
def form_submit(req: FormFunctionRequest, res: HandlerResponse, *args):
    return {}
@form_change_handler
def change_form(req: FormFunctionRequest, res: FormChangeResponse, *args):
    return res
@form_dynamic_field_handler
def handle_dynamic_field(req: FormFunctionRequest, res: FormDynamicFieldResponse, *args):
    return res
@widget_button_handler
def widget_button(req: WidgetFunctionRequest, res: WidgetResponse, *args):
    return {}
def get_button_section():
    widget_response = function_handler.new_widget_response()
    button_section = widget_response.new_widget_section()
    return button_section

import zcatalyst_cliq._handler
class GithubConstants:
    PERSONAL_ACCESS_TOKEN = '{{Enter your Git personal access token here}}'  
    PER_PAGE = 10
    BOT_NAME = "GitHubBot"