동기

저번에 구축한 REST API에서 추가로 내가 원하는 동작을 할 수 있도록 lambda_function.py를 수정하려고 한다.

과정

AWS Lambda는 python 기본 라이브러리를 제공해주지만 외부 라이브러리는 자동으로 제공해 주지 않는다.

Untitled

Untitled

코드가 계속 오류가 떠서 엄청 고생했는데 해결방법은 '일반 구성'에서 메모리와 제한 시간을 늘려주면 된다.

코드가 계속 오류가 떠서 엄청 고생했는데 해결방법은 '일반 구성'에서 메모리와 제한 시간을 늘려주면 된다.

내가 구현하고 싶은 메인 동작은 사용자가 메시지의 버튼을 누르면 책 제작 신청 url과 함께 데이터가 API로 전송되고, Lambda에서 그 정보를 받아서 사이트에 로그인해서 해당 책을 신청하는 것이다.

그렇다면 해야 할 일은 다음과 같다.

  1. 버튼 정보에 책 url 포함시키기

    메시지 block의 버튼의 value에 최근에 업로드된 책 제작 신청 url를 넘겨주면 이후 API가 전송받은 json 정보에서 json.dumps(event)["actions"][0]["value"]를 추출하면 된다.

    import requests
    import json
    from bs4 import BeautifulSoup
    import re
    import time
    from datetime import datetime
    
    BOT_TOKEN = 'xoxb-*************-*************-************************'
    
    def post_message(token, channel_id, info):
        url = "<https://slack.com/api/chat.postMessage>"
        headers = {
            "Content-type": "application/json; charset=utf-8",
            "Authorization": "Bearer " + token
        }
        data = {
            "channel": channel_id,
            "blocks": [
                {
                    "type": "section",
                    "text": {
                        "type": "mrkdwn",
                        "text": info["text"]
                    }
                },
                {
                    "type": "image",
                    "image_url": info["img_url"],
                    "alt_text": info["title"]
                },
                {
                    "type": "actions",
                    "elements": [
                        {
                            "type": "button",
                            "text": {
                                "type": "plain_text",
                                "text": "책 제작 신청하기",
                                "emoji": True
                            },
                            #"style": "primary",
                            "value": info["book_url"],
                            "action_id": "actionId-0"
                        }
                    ]
                }
            ]
        }
    
        response = requests.post(url, headers=headers, data=json.dumps(data))
        print(response)
        if response.json().get('ok'):
            print('메시지를 성공적으로 보냈습니다.\\n')
        else:
            print('메시지를 성공적으로 보내지 못했습니다. 오류메시지 : ' + str(response.json()))
    
    def alarm_books():
        site_url = "<https://www.itlo.org>"
        url = "<https://www.itlo.org/booking>"
        r = requests.get(url)  # url에 접근
        # 참고로 r.text는 binary, r.content는 str형태의 데이터이다.
        bs = BeautifulSoup(r.content, "lxml")  # "lxml"이라는 parser가 앞서 가져온 데이터에서 html요소로 접근
        divs = bs.select("div.col-xs-12.col-md-6.col-lg-3.cropbkiddata")  # select의 결과 = 리스트
    
        # 시간 출력 => <https://codechacha.com/ko/python-get-current-time/> 참고함
        current_time = datetime.now()
        print('-' * 10)
        print("{hour}시 {minute}분".format(hour=current_time.hour, minute=current_time.minute))
        print('-' * 10)
    
        for d in divs:
            images = d.select("img")[0]
            image = images.get("src")
            img_url = site_url + image
            ahref = d.select("a")
            link = ahref[0].get("href")
            link_url = site_url + link
            title = ahref[1].select("font")[0].text  # 태그 사이의 글은 text로 접근 가능
    
            b_url = site_url + link
            b_r = requests.get(b_url)
            bbs = BeautifulSoup(b_r.content, "lxml")
            b_pages = bbs.select("span")[-1].text
            numbers = re.findall(r'\\d+', b_pages)
    
            if numbers[0] != '0':
                text_form = "*[NEW!!] 제작가능한 도서 등록됨* :loudspeaker:\\n제목:{title}\\n{book_pages}"
                text = text_form.format(title=title, book_pages=b_pages)
                info = {"text": text, "title": title, "book_url": link_url, "img_url": img_url}
                # print(info)
                return info
    
        print("새로 등록된 도서가 없습니다.\\n")
    
    if __name__ == "__main__":
        try:
            print("To Exit, press 'CTRL+C'.")
            while True:
                book_info = alarm_books()
                if book_info is not None:
                    post_message(BOT_TOKEN, "#봉사활동-도서-알림", book_info)
                time.sleep(60 * 5)  # 5분
        except KeyboardInterrupt:
            print("You pressed 'CTRL+C'. Stop running program.")
    

    테스트 결과 의도한 형식대로 알림도 정상적으로 오고, 버튼을 눌렀을 때 API로 json 정보도 제대로 전달된다.

    알림 사진(왼쪽)과 API로 전달된 json 정보(오른쪽). "actions"의 "value"에 책 제작 신청 url이 잘 전달된 것을 확인할 수 있다.

    알림 사진(왼쪽)과 API로 전달된 json 정보(오른쪽). "actions"의 "value"에 책 제작 신청 url이 잘 전달된 것을 확인할 수 있다.

  2. lambda 함수 코딩

    이제 버튼을 눌렀을 때 API쪽(lambda 함수)에서 동작할 코드를 작성해야 한다. 전에 selenium에서 했던 것처럼 자동 로그인 후 책 제작 신청을 하고 그 결과에 따라 기존 메시지를 응답 메시지로 업데이트하는 것까지 코딩했다.

lambda_function.py

selenium_function.py

message_function.py

version 1.0

version 1.0

version 3.0 (이미 페이지를 작업 중인 경우에 대한 처리 추가)

version 3.0 (이미 페이지를 작업 중인 경우에 대한 처리 추가)

version 2.0

version 2.0

결과

현재 제작 가능한 도서는 없어서 일단 가장 최근에 등록된 도서로 테스트를 해봤다. selenium 코드의 동작 시간이 조금 길기 때문에 메시지가 update되기까지 시간이 조금 걸린다.