AWS Chalice 및 Upstash Redis를 사용한 서버리스 생일 Slackbot

때로는 연례 행사에 대한 알림을 만들어 특별한 날짜를 잊지 않고 놓치는 일이 없도록 하는 것이 가장 좋습니다.

귀하와 귀하의 팀/친구가 Slack을 사용하는 경우 slackbot을 통해 이러한 알림을 자동화하는 것이 좋습니다.

그렇게 하는 동안 슬랙봇을 유지 관리가 적게 하고 싶다면; 소스와의 동시 상호 작용을 위해 서버리스 기술을 사용하는 것이 가장 좋습니다. 또한 수평 확장도 가능합니다.

우리가 만들고 있는 것

이벤트 알림 Slackbot을 구축 중입니다. 호스팅을 위해 Python, AWS Chalice, AWS Lambda 및 API Gateway를 사용합니다. 사용자는 다음을 수행할 수 있습니다.

• 사용자의 생일을 설정합니다.
• 사용자의 기념일을 설정합니다.
• 사용자 또는 일반 채널에 대한 맞춤 이벤트 설정

이벤트가 설정되면:

• 이벤트의 중심에 있는 사람(이벤트를 설정할 때 언급된 사람)을 제외하고 특정 이벤트가 다가오고 있음을 사람들에게 상기시킵니다.
• 이벤트 기념일이 되면 이벤트의 중심에 있는 사람(또는 채널의 모든 사람)을 언급하여 일반 채널에 게시합니다.

명령

설정

• /event set birthday <YYYY-MM-DD> <user>

  사용자의 생일을 설정합니다.

  

• /event set anniversary <YYYY-MM-DD> <user>

  사용자가 작업을 시작한 기념일을 설정합니다.

  

• /event set custom <YYYY-MM-DD> <user> <any kind of message with whitespaces>

  제공된 메시지를 사용하여 맞춤 알림을 설정합니다.

  

모두 가져오기

• /event get-all :

  설정된 모든 이벤트를 표시합니다.

  

• /event get-all birthday :

  설정된 모든 생일을 표시합니다.

  

• /event get-all anniversary :

  설정된 모든 기념일을 표시합니다.

  

• /event get-all custom :

  설정된 모든 맞춤 이벤트를 표시합니다.

  

얻기

• /event get birthday <user> :

  사용자의 생일 세부정보를 표시합니다.

  

• /event get anniversary <user> :

  사용자가 작업을 시작한 날짜에 대한 기념일 세부정보를 표시합니다.

  

• /event get custom <event_name>(can be found with get-all) :

  제공된 메시지를 사용하여 사용자 지정 이벤트 세부정보를 표시합니다.

  

제거

• /event remove birthday <user> 제거 :

  사용자의 생일을 제거합니다.

  

• /event remove anniversary <user> :

  사용자가 작업을 시작한 기념일을 제거합니다.

  

• /event remove custom <event_name>(can be found with get-all) :

  제공된 메시지를 사용하여 사용자 정의 이벤트를 제거합니다.

  

예약 알림

• 일반 채널 알림

  시간이 되면 slackbot은 지정된 채널로 알림 메시지를 보냅니다.

• 봇의 개인 메시지

  시간이 다가오면 slackbot이 개인 알림 메시지를 보냅니다.

따라서 이 도구를 사용하여 팀원의 특별 날짜를 추적할 수 있습니다. 이렇게 하면 관계와 상호 의사 소통이 건강한 방식으로 유지될 수 있습니다.

시작하기

데이터베이스 준비

Upstash 콘솔에서 Redis 데이터베이스를 생성할 수 있습니다. UPSTASH_REDIS_REST_URL 및 UPSTASH_REDIS_REST_TOKEN은 AWS의 환경 변수가 되므로 주의하십시오.

AWS 자격 증명 구성

(공식 Chalice Repo에서 가져왔습니다. 자세한 내용은 그곳에서 참조할 수 있습니다.)

$ mkdir ~/.aws
$ cat >> ~/.aws/config
[default]
aws_access_key_id=YOUR_ACCESS_KEY_HERE
aws_secret_access_key=YOUR_SECRET_ACCESS_KEY
region=YOUR_REGION (such as us-west-2, us-west-1, etc)

일부 협약

• 모든 .py app.py 외부의 파일 chalicelib 아래에 위치해야 합니다. 그렇지 않으면 import 문으로 인해 문제가 발생할 수 있습니다.
• 모든 환경 변수는 config.json에서 구성해야 합니다. .chalice 내부의 파일 디렉토리.
  • json 형식, 키:"environment_variables"

프로젝트 소스 개발

• 우선 AWS Chalice를 사용하고 있기 때문에 , 성배 설치:

  pip install chalice

성배 프로젝트 시작

chalice new-project <project_name> 그런 다음 프로젝트 폴더에 cd를 넣습니다. 프로젝트에는 이미 템플릿이 포함되어 있습니다.

실행:chalice local 프로젝트가 작동하는지 확인하십시오.

앱.py

전체 프로젝트 구조 및 Slack 요청을 처리하기 위한 기본 파일입니다.

이를 통해 프로젝트 구조와 끝점을 만듭니다. 이벤트 처리 방법, 알림이 작동하도록 예약할 항목을 결정합니다.

from chalice import Chalice, Cron, Rate
import os
import random
from datetime import date
from chalicelib.utils import responseToDict, postToChannel, diffWithTodayFromString, allSlackUsers, sendDm, validateRequest, convertToCorrectMention
from chalicelib.upstash import setHandler, getAllHandler, getEvent, getAllKeys, removeEvent

app = Chalice(app_name='birthday-slackbot')
NOTIFY_TIME_LIMIT = int(os.getenv("NOTIFY_TIME_LIMIT"))


# Sample route for get requests.
@app.route('/', methods=["GET"])
def something():
    return {
        "Hello": "World"
        }

# Configuring POST request endpoint.
# Command is parsed and handled/directed to handler
@app.route('/', methods=["POST"], content_types=["application/x-www-form-urlencoded"])
def index():

    # Parse the body for ease of use
    r = responseToDict(app.current_request.raw_body)
    headers = app.current_request.headers

    # Check validity of the request.
    if not validateRequest(headers, r):
        return {"Status": "Validation failed."}


    commandArray = r['text'].split()
    command = commandArray.pop(0)

    try:
        if command == "set":
            setHandler(commandArray)
            return {
            'response_type': "ephemeral",
            'text': "Set the event."
            }

        elif command == "get":
            eventType = commandArray[0]
            eventName = eventType + "-" + commandArray[1]
            resultDict = getEvent(eventName)
            return {
            'response_type': "ephemeral",
            'text': "`{}` Details:\n\n Date: {}\nRemaining: {} days!".format(eventName, resultDict[0], resultDict[1])
            }

        elif command == "get-all":

            stringResult = getAllHandler(commandArray)
            return {
            'response_type': "ephemeral",
            'text': "{}".format(stringResult)
            }

        elif command == "remove":
            eventName = "{}-{}".format(commandArray[0], commandArray[1])
            removeEvent(eventName)
            return {
            'response_type': "ephemeral",
            'text': "Removed the event."
            }
        else:
            return {
            'response_type': "ephemeral",
            'text': "Wrong usage of the command."
            }
    except:
        print("some stuff")
        return {
            'response_type': "ephemeral",
            'text': "Some problem occured. Please check your command."
        }


# Run at 10:00 am (UTC) every day.
@app.schedule(Cron(0, 10, '*', '*', '?', '*'))
def periodicCheck(event):
    allKeys = getAllKeys()
    for key in allKeys:
        handleEvent(key)


# Generic event is parsed and directed to relevant handlers.
def handleEvent(eventName):
    eventSplitted = eventName.split('-')

    eventType = eventSplitted[0]

    # discard @ or ! as a first character
    personName = eventSplitted[1][1:]
    personMention = convertToCorrectMention(personName)

    eventDict = getEvent(eventName)
    remainingDays = eventDict[1]
    totalTime = eventDict[2]


    if eventType == "birthday":
        birthdayHandler(personMention, personName, remainingDays)

    elif eventType == "anniversary":
        anniversaryHandler(personMention, personName, remainingDays, totalTime)

    elif eventType == "custom":
        eventMessage = "Not specified"
        if len(eventSplitted) == 3:
            eventMessage = eventSplitted[2]
        customHandler(eventMessage, personMention, personName, remainingDays)

# Handles birthday events.
def birthdayHandler(personMention, personName, remainingDays):
    if remainingDays == 0:
        sendRandomBirthdayToChannel('general', personMention)
    if remainingDays <= NOTIFY_TIME_LIMIT:
        dmEveryoneExcept("{} day(s) until {}'s birthday!".format(remainingDays, personMention), personName)

# Handles anniversary events.
def anniversaryHandler(personMention, personName, remainingDays, totalTime):
    if remainingDays == 0:
        sendRandomAnniversaryToChannel('general', personMention, totalTime)
    if remainingDays <= NOTIFY_TIME_LIMIT:
        dmEveryoneExcept("{} day(s) until {}'s anniversary! It will be {} year(s) since they joined!".format(remainingDays, personMention, totalTime), personName)

# Handles custom events.
def customHandler(eventMessage, personMention, personName, remainingDays):
    if remainingDays == 0:
        postToChannel('general', "`{}` is here {}!".format(eventMessage, personMention))
    elif remainingDays <= NOTIFY_TIME_LIMIT:
        dmEveryoneExcept("{} day(s) until {} `{}`!".format(remainingDays, personMention, eventMessage), personName)


# Sends private message to everyone except for the person given.
def dmEveryoneExcept(message, person):
    usersAndIds = allSlackUsers()
    for user in usersAndIds:
        if user[0] != person:
            sendDm(user[1], message)


# Sends randomly chosen birthday message to specified channel.
def sendRandomBirthdayToChannel(channel, personMention):
    messageList = [
        "Happy Birthday {}! Wishing you the best!".format(personMention),
        "Happy Birthday {}! Wishing you a happy age!".format(personMention),
        "Happy Birthday {}! Wishing you a healthy, happy life!".format(personMention),
    ]
    message = random.choice(messageList)
    return postToChannel('general', message)

# Sends randomly chosen anniversary message to specified channel.
def sendRandomAnniversaryToChannel(channel, personMention, totalTime):
    messageList = [
        "Today is the anniversary of {} joining! It has been {} years since they joined!".format(personMention, totalTime - 1),
        "Celebrating the anniversary of {} joining! It has been {} years!".format(personMention, totalTime - 1),
        "Congratulating {} for entering {}(th) year here!".format(personMention, totalTime),
    ]
    message = random.choice(messageList)
    return postToChannel('general', message)


# We want to run our event handlers when the project is deployed/redeployed.
allKeys = getAllKeys()
for key in allKeys:
    handleEvent(key)

chalicelib/utils.py

도우미 함수 및 추상화에 대한 기본 파일입니다.

우리는 주로 이 파일을 추상화에 사용할 것입니다. 따라서 소스 코드가 복잡하지 않고 가독성이 유지됩니다.

from urllib import request
import urllib
from urllib.parse import parse_qsl
import json
import os
import hmac
import hashlib
from datetime import date


SLACK_BOT_TOKEN = os.getenv("SLACK_BOT_TOKEN")
SLACK_SIGNING_SECRET = os.getenv("SLACK_SIGNING_SECRET")

# Returns real name of the slack user.
def getRealName(slackUsers, username):
    for user in slackUsers:
        if user[0] == username:
            return user[2]
    return "Nameless"

# Returns all slack users in the workspace.
def allSlackUsers():
    resultDict = sendPostRequest("https://slack.com/api/users.list", SLACK_BOT_TOKEN)
    members = resultDict['members']

    userMembers = []
    for member in members:
        if not member['deleted'] and not member['is_bot']:
            userMembers.append([member['name'], member['id'], member['real_name']])

    return userMembers

# Returns the id of the given channel.
def channelNameToId(channelName) :
    resultDict = sendPostRequest("https://slack.com/api/conversations.list", SLACK_BOT_TOKEN)
    for channel in resultDict['channels']:
        if (channel['name'] == channelName):
            return channel['id']
    return None

# Posts to given slack channelId with given message.
def postToSlack(channelId, messageText):
    data = {
        "channel": channelId,
        "text": messageText
    }
    data = json.dumps(data)
    data = str(data)
    data = data.encode('utf-8')
    resultDict = sendPostRequest("https://slack.com/api/chat.postMessage", SLACK_BOT_TOKEN, data)
    return resultDict

# Posts to a slack channel.
def postToChannel(channel, messageText):
    channelId = channelNameToId(channel)
    return postToSlack(channelId, messageText)

# Sends a private message to a user with userId.
def sendDm(userId, messageText):
    return postToSlack(userId, messageText)

# Sends generic post request and returns the result.
def sendPostRequest(requestURL, bearerToken, data={}):
    req = request.Request(requestURL, method="POST", data=data)
    req.add_header("Authorization", "Bearer {}".format(bearerToken))
    req.add_header("Content-Type", "application/json; charset=utf-8")

    r = request.urlopen(req)
    resultDict = json.loads(r.read().decode())
    return resultDict

# Parses and converts the res to dict.
def responseToDict(res):
    return dict(parse_qsl(res.decode()))


# Dates are given as: YYYY-MM-DD
# Returns difference between current day and the anniversary.
def diffWithTodayFromString(dateString):
    now = date.today()
    currentYear = now.year

    dateTokens = dateString.split("-")
    month = int(dateTokens[1])
    day = int(dateTokens[2])

    if now > date(currentYear, month, day):
        return (date((currentYear + 1), month, day) - now).days
    return (date(currentYear, month, day) - now).days


# Dates are given as: YYYY-MM-DD
# Calculates the total time that has passed until current date.
def totalTimefromString(dateString):
    now = date.today()

    dateTokens = dateString.split("-")
    year = int(dateTokens[0])
    month = int(dateTokens[1])
    day = int(dateTokens[2])

    then = date(year, month, day)

    years = now.year - then.year
    return years + 1

# Validate requests coming to endpoint.
# Hashes request body with timestamp and signing secret.
# Then, compares that hash with slack signature.
def validateRequest(header, body):

    bodyAsString = urllib.parse.urlencode(body)

    timestamp = header['x-slack-request-timestamp']
    slackSignature = header['x-slack-signature']
    baseString = "v0:{}:{}".format(timestamp, bodyAsString)

    h =  hmac.new(SLACK_SIGNING_SECRET.encode(), baseString.encode(), hashlib.sha256)
    hashResult = h.hexdigest()
    mySignature = "v0=" + hashResult

    return mySignature == slackSignature

# Converts given name to mention string.
def convertToCorrectMention(name):
    if name == "channel" or name == "here" or name == "everyone":
        return "<!{}>".format(name)
    else:
        return "<@{}>".format(name)

chalicelib/upstash.py

데이터베이스와 직접 관련된 기능에 대한 메인 파일입니다.

여기에서 데이터베이스 호출을 처리합니다. 데이터베이스에서 가져오고 키-값 쌍 등을 설정합니다. 이 파일은 또한 app.py에서 낮은 수준의 세부 정보를 추상화하는 데 도움이 됩니다. , 가독성과 모듈성을 향상시킵니다.

Upstash Redis 데이터베이스의 장점은 RESTFUL API 호출을 지원한다는 것입니다. 이렇게 하면 지속적으로 연결을 만들고 닫을 필요 없이 데이터베이스에 액세스할 수 있습니다. 이는 서버리스 애플리케이션을 위한 방법입니다.

from chalicelib.utils import sendPostRequest, getRealName, allSlackUsers, diffWithTodayFromString, totalTimefromString
import os

UPSTASH_REST_URL = os.getenv("UPSTASH_REST_URL")
UPSTASH_TOKEN = os.getenv("UPSTASH_TOKEN")

# Posts to Upstash Rest Url with parameters given.
def postToUpstash(parameters):
    requestURL = UPSTASH_REST_URL
    for parameter in parameters:
        requestURL += ("/" + parameter)

    resultDict = sendPostRequest(requestURL, UPSTASH_TOKEN)
    return resultDict['result']


# Sets key-value pair for the event with given parameters.
def setEvent(parameterArray):

    postQueryParameters = ['SET']

    for parameter in parameterArray:
        parameter = parameter.split()
        for subparameter in parameter:
            postQueryParameters.append(subparameter)

    resultDict = postToUpstash(postQueryParameters)

    return resultDict


# Returns event details from the event given.
def getEvent(eventName):
    postQueryParameters = ['GET', eventName]
    date = postToUpstash(postQueryParameters)

    timeDiff = diffWithTodayFromString(date)
    totalTime = totalTimefromString(date)
    mergedDict = [date, timeDiff, totalTime]
    return mergedDict

# Fetches all keys (events) from the database
def getAllKeys():
    return postToUpstash(['KEYS', '*'])

# Deletes given event from the database.
def removeEvent(eventName):
    postQueryParameters = ['DEL', eventName]
    resultDict = postToUpstash(postQueryParameters)
    return resultDict


# Handles set request by parsing and configuring setEvent function parameters.
def setHandler(commandArray):
    eventType = commandArray.pop(0)
    date = commandArray.pop(0)
    user = commandArray.pop(0)

    if eventType == "birthday":
        listName = "birthday-" + user
        return setEvent( [listName, date] )

    elif eventType == "anniversary":
        listName = "anniversary-" + user
        return setEvent( [listName, date] )

    elif eventType == "custom":
        message = ""
        for string in commandArray:
            message += string + "_"

        listName = "custom-" + user + "-" + message
        user = commandArray[1]
        return setEvent( [listName, date] )
    else:
        return

# Handles get-all requests.
def getAllHandler(commandArray):
    filterParameter = None
    if len(commandArray) == 1:
        filterParameter = commandArray[0]

    allKeys = getAllKeys()
    birthdays = []
    anniversaries = []
    customs = []

    slackUsers = allSlackUsers()

    stringResult = "\n"
    for key in allKeys:
        if key[0] == 'b':
            birthdays.append(key)
        elif key[0] == 'a':
            anniversaries.append(key)
        elif key[0] == 'c':
            customs.append(key)

    if filterParameter is None or filterParameter == "birthday":
        stringResult += "Birthdays:\n"
        for bday in birthdays:
            tag = bday.split('-')[1]
            username = tag[1:]
            realName = getRealName(slackUsers, username)
            details = getEvent(bday)

            stringResult += "`{}` ({}): {} - `{} days` remaining!\n".format(tag, realName, details[0], details[1])

    if filterParameter is None or filterParameter == "anniversary":
        stringResult += "\nAnniversaries:\n"
        for ann in anniversaries:
            tag = ann.split('-')[1]
            username = tag[1:]
            realName = getRealName(slackUsers, username)
            details = getEvent(ann)

            stringResult += "`{}` ({}): {} - `{} days` remaining!\n".format(tag, realName, details[0], details[1])

    if filterParameter is None or filterParameter == "custom":
        stringResult += "\nCustom Reminders:\n"
        for cstm in customs:
            splitted = cstm.split('-')
            username = splitted[2]
            realName = getRealName(slackUsers, username)
            details = getEvent(cstm)

            stringResult += "`{}-{}` ({}): {}\n".format(splitted[1], splitted[2], getRealName(slackUsers, username), details[0])

    return stringResult

.chalice/config.json

AWS에서 프로젝트를 구성하기 위한 파일입니다.

여기에서 환경 변수 및 배포 단계와 같은 프로젝트 세부 정보를 정의합니다. 이를 위해 다음을 추가하여 환경 변수만 구성합니다.

{
  "environment_variables": {
    "UPSTASH_REST_URL": <UPSTASH_REDIS_REST_URL>,
    "UPSTASH_TOKEN": <UPSTASH_REDIS_REST_TOKEN>,
    "SLACK_BOT_TOKEN": <SLACK_BOT_TOKEN>,
    "SLACK_SIGNING_SECRET": <SLACK_SIGNING_SECRET>,
    "NOTIFY_TIME_LIMIT": "<amount of days before getting notifications for events>"
    }
}

모든 작업이 완료된 후

폴더 구조

폴더 구조는 다음과 같아야 합니다.

<project_name>:
    app.py

    chalicelib:
        utils.py
        upstash.py
        <Some other default files generated by chalice>

    .chalice:
        config.json
        <Some other default files generated by chalice>

로컬에서 실행

Chalice는 로컬 배포를 지원하므로 개발 프로세스가 정말 빨라집니다.

실행:chalice local

고정 IP 주소가 없으면 ngrok와 같은 터널링 서비스를 사용해야 합니다. Slack에 엔드포인트를 표시할 수 있도록:

./ngrok http 8000 --> localhost:8000 터널링

Slack 구성

1. Slack API 앱 페이지로 이동:

• 새 앱 만들기
  • 처음부터
  • 앱 이름 지정 및 작업 공간 선택
• 인증 및 권한으로 이동
  • 다음 범위 추가
    • 채널:읽기
    • 채팅:쓰기
    • 채팅:write.public
    • 명령
    • 그룹:읽기
    • 사용자:읽기
  • 작업 공간에 앱 설치
    • 기본 정보 --> 앱 설치 --> 작업 공간에 설치

3. Slack API 앱 페이지로 이동하여 관련 앱을 선택합니다.

배포 후 REST_API_URL을(를) 사용할 수 있습니다. 또는 ngrok_domain <domain>으로 .

1. Slack API 앱 페이지로 이동하여 관련 앱을 선택합니다.

• 슬래시 명령으로 이동:
  • 새 명령 만들기:
    • 명령어:event
    • 요청 URL:<domain>
    • 나머지는 원하는 대로 구성하세요.

• 이러한 변경 후에 Slack에서 앱을 다시 설치해야 할 수 있습니다.

축하합니다!

이제 작동하는 서버리스 Slackbot이 있습니다! 원하는 대로 자유롭게 사용자 정의할 수 있습니다.

로컬 호스팅 및 결과에 만족하면 다음과 같이 하십시오.

• chalice deploy AWS Lambda 및 API Gateway에 대한 최종 배포용.

이제 Slack 구성에서 AWS Chalice에서 제공하는 REST_API_URL을 사용할 수 있습니다.

전체 프로젝트를 보려면 Github Repo를 방문하세요.