[RPA]Python으로 mini Bot 만들기

• RPA 는 솔루션으로만 구현할 수 있나?

Robotic Process Automation의 약자. 현재는 데이터 입력등의 단순 반복 업무 프로세스의 자동화에 주로 적용되고 있지만,
향후 AI, 머신러닝등의 기술이 발전, RPA와 결합한다면 자료 분석 및 Solution 제시등의 영역까지도 가능(?)할 수 있다.
기존의 Robot이 공장 생산 라인의 실체적 기계였다면 RPA는 Software로 구현된 사무직을 위한 Robot으로 이해하면 쉽다.

from : https://namu.wiki/w/RPA

RPA - 나무위키

2016년3월 딥마인드의 알파고와 이세돌의 바둑이야기로 IT 현장에서는 AI가 마케팅의 하나의 수단으로 사용되었다.

모든 신규 발주프로젝트와 조직은 AI가 붙어야 했고 참으로 준비되지 않은 그땐, 미지의 세계인 실적없는 AI에 모두들 덤벼들어야 했다.

 

그에 비슷한 시기에 RPA 역시 IT의 화두로 떠올랐다. 적어도 내가 속한 조직에서는 단순반복 업무를 조사하고, 그 현황을 파악해서 보고하고 그것을 줄이면 적어도 월 MM가 얼마나 줄건인가에 보고서가 집중되었다.

 

RPA의 대표적인 솔루션은 UIPATH라는 제품이있다.

많은 회사에서 해당 제품을 도입하였고, 여러 뉴스에서도 금융권에 RPA 솔루션 도입과 AI 도입으로, 앞으로 노동자들이 다수 발생할 것이라고 했다.

 

나 역시 최근 python을 학습하게 되면서 java의 Hello world 만 출력하려도 소요되는 엄청된 소스와 그 행위를 이해하기 위한 학습의 절대량이 필요하지만, python은 너무나도 깔끔했다는 것을 느꼈다.

 

이 python 은 나의 업무를 편하게 해줄 RPA 를 구현하기 위한 언어로 안성맞춤이였다.

 

RPA는 솔루션을 사용하지 않아도 나의 단순반복한 업무를 위임시키면 알아서 구현해 줄수 있는 것이라면 모두 해당되며, 내가 생각하는 그 중에는 python은 수많은 확장가능한 라이브러리가 최적의 조합이 최고의 솔루션이 되었다.

 

 

• 샘플1. 자동으로 아침마다 코로나 신규 감염자를 알려준다면?

요즘 우리를 힘들게 하는 코로나, 나의 일과는 아침에 일어나 신규 확진자 소식을 듣는 것으로 시작한다.

그러나 신규 확진자는 당일 0시 기준으로 오전10시에 업데이트가 이뤄진다고 한다.

즉, 회사가서 일하다보면 까먹고,  코로나에 둔감해졌다가 점심먹다가 또는 오후에 뉴스보다가 심각성을 알게 된다.

그래서 아침 10시마다 내게 신규 확진자 숫자를 알려주는 bot이 필요했다.

 

결론은 아래와 같이 구현되었다.

매일 오전 확진자 숫자가 slack 으로 온다.

 

집에 있는 서버를 이용해서 오전에 slack으로 내게 정보를 달라고 했다.

그러면 카카오를 가서 관련 정보를 캡쳐해서 내게 보내준다.

왜 카카오? 정부에서 운영하는 것보다는 숫자를 그래픽으로 보여주는 UI가 한눈에 잘 들어왔다.

데이터는 역시 어떻게 보여주는 것인가가 매우 중요하다.

#! /usr/bin/python3
from selenium import webdriver
from PIL import Image
from io import BytesIO
from slacker import Slacker
from apscheduler.schedulers.blocking import BlockingScheduler
import time
from datetime import datetime


class KakaoCOVID19():
    def __init__(self):
        self.token = '슬랙토큰정보를 입력한다.!~~~~~~~~~~~~~~~~~~'
        self.driver_path = '/usr/bin/chromedriver'
        self.FILE_DOWNLOAD_PATH = "/home/seluser/workspace"
        self.index_url = "https://news.daum.net/covid19"
        self.options = webdriver.ChromeOptions()
        self.options.add_argument('--headless')
        self.options.add_argument("--hide-scrollbars")
        self.options.add_argument('--no-sandbox') 
        self.options.add_argument('--disable-dev-shm-usage') 
        self.driver = webdriver.Chrome(self.driver_path, chrome_options=self.options)
    
    def getImage(self):
        try:
            #보이는 대로 다 캡쳐하려고 윈도우 사이즈를 길게 만들었다.
            self.driver.set_window_size(500,2580)
            self.driver.get(self.index_url )
            #상단의 요약부분이 제일 중요해서 그것만 크롭해서 가져온다.
            element = self.driver.find_element_by_xpath("/html/body/div/header/div");
            image_name = "KakaoCOVID19.png"
            crop_image_name = 'crop_img.png'
            self.driver.save_screenshot(image_name)
            time.sleep(0.5)
            #crop image 생성하기(46번째 라인 한줄이 더 깔끔하다. 아래 주석처리함)
#             location = element.location;
#             size = element.size;
#             x = location['x'];
#             y = location['y'];
#             width = location['x']+size['width'];
#             height = location['y']+size['height'];
#             screenshot_img = Image.open(image_name)
#             crop_img = screenshot_img.crop((int(x), int(y), int(width), int(height)))
#             crop_img.save(crop_image_name)
            #해당 엘리먼트만 뽑아서저장한다.
            element.screenshot(crop_image_name)
            time.sleep(1)
            msg_string = "kakao코로나 소식"
            self.send_msg_on_slack_with_file(crop_image_name, msg_string)
            time.sleep(3)
            self.send_msg_on_slack_with_file(image_name, msg_string)
        except Exception as inst:
            print("exception 발생")
            print(type(inst))    # the exception instance
            print(inst.args)     # arguments stored in .args
            print(inst)       
        finally:
            self.driver.close()
            self.driver.quit()
    
    def send_msg_on_slack_with_file(self, img_path, msg_string):
        try:
            slack= Slacker(self.token)
            slack.files.upload(img_path, channels='#covid19',title=msg_string, initial_comment=msg_string)
        except Exception as inst:
            print(type(inst))    # the exception instance
            print(inst.args)     # arguments stored in .args
            print(inst)  

        
if __name__ == '__main__':
    covid19 = KakaoCOVID19()
    #covid19.getImage()
    now = datetime.now()
    print(now)
    sched = BlockingScheduler()
    # 예약방식 interval로 설정, 10초마다 한번 실행
    #sched.add_job(covid19.getImage, 'interval', seconds=60)
    # 갱신이 10시에 되지만, 매9~12시의 5분에 수신받도록 해본다.
    sched.add_job(covid19.getImage, 'cron', hour='9,10,11,12', minute='*/5')
    # 스케줄링 시작
    sched.start()

시간은 조절하겠지만, 이제 코로나 관련 정보를 오전중에 갱신된 최근 정보로 수신받게 되었다.