와탭랩스는 직원들의 업무능력 향상을 위해 도서 구매를 지원하고 있습니다. 임직원들이 가장 좋아하는 복지 중 하나라고 말할 수 있을 것 같습니다.

다만 도서 구매 절차가 생각보다 많은 인력이 소모되고 있었습니다

개편 이전 도서 구매 절차는 다음과 같았습니다.

와탭랩스 직원들이 가장 좋아하는 복지인 만큼 도서 구매 신청 양도 상당하기 때문에 이를 전부 확인하고 시트에 작성하는 것은 상당한 시간이 소요될 수밖에 없었습니다.

도서 구매를 자주 하는 입장에서 뒤에서 노력해 주고 계신 경영지원팀에 감사하는 마음으로 해당 구간을 자동화해야겠다고 생각했습니다

구도

Flow

제가 개선한 도서 구매 자동화는 아래와 같은 Flow로 동작을 합니다

동작 원리

BookSupportBot

• telegram bot을 활용하여 실시간으로 도서 구매 요청 정보 습득 (URL, 신청자)

• 와탭랩스 도서 구매 복지는 알라딘만 허용하기 때문에 알라딘 URL이 맞는지 확인
  • 단축 URL의 경우 Redirect 되는 URL 획득 후 확인

• 알라딘 URL에서 ItemID를 얻고 알라딘 API를 통해서 도서 판매 정보 획득

  [https://blog.aladin.co.kr/openapi/popup/6695306]

  • 알라딘에서 각 도서는 고유의 ItemID를 가지고 있음

• 구글 서비스 어카운트를 이용해 도서 판매 정보를 시트에 작성

• telegram bot을 활용해 도서 구매 신청 완료 메세지를 전송

코드

• Telegram message polling

  from telegram.ext import Updater
  from telegram.ext import MessageHandler, Filters
  
  #telegram bot token
  token =""
  bot = telegram.Bot(token)
  
  #message polling
  updater = Updater(token=token, use_context=True)
  dispatcher = updater.dispatcher
  updater.start_polling()
  
  #update event handler
  def handler(update, context):
  		#telegram chat id 
  		_id=update.message.chat.id
  
  		#telegram message 내용
      user_text = update.message.text
  		if user_text.startswith("/") == False:
          return
  
  		#telegram message 전송자
  		if update.message.from_user.last_name is None:
          user_name=update.message.from_user.first_name
      else:
          user_name = update.message.from_user.last_name+update.message.from_user.first_name 
  
  echo_handler = MessageHandler(Filters.text, handler)
  dispatcher.add_handler(echo_handler)

  • telegram bot이 polling
    • 메시지 발생 시 handler라는 함수 실행
    • 매개변수로 message 내용 chat_id 등 다양한 정보 포함
  • 일반적인 대화와 혼동을 막기 위해 / 으로 시작해야 함

• URL parsing

  from urllib import response
  from urllib.parse import urlparse, parse_qs
  import requests
  import json
  
  def parsing_url(url):
      try:
          parsed_url = urlparse(url)
  				
  				#단축 url확인 (aladin단축 url은 aladin.kr을 host로가짐)
          if parsed_url.hostname=="aladin.kr":
              print(url)
  						#redirect url확인
              final_url=get_final_url(url)
              parsed_url=urlparse(final_url)
              if parsing_url==None:
                  return None
  				
          query_params = parse_qs(parsed_url.query)
          
  				#ItemId 획득
          item_id = query_params.get('ItemId', [None])[0]
          if item_id:
              return item_id
          else:
              item_id = query_params.get('itemid', [None])[0]
              if item_id:
                  return item_id
              else:
                  print("Error: ItemId not found in the URL.")
                  return None
  
      except Exception as e:
          return None
  

  • 단축 URL 의 경우 redirect URL을 가져옴
    • 판단기준 : host
  • URL에서 parameter(ItemId)를 가져옴

• Redirect URL 가져오기

  from urllib import response
  from urllib.parse import urlparse, parse_qs
  import requests
  import json
  
  def get_final_url(url):
      try:
          response = requests.head(url, allow_redirects=True)
          final_url = response.url
          return final_url
      except requests.RequestException as e:
          print(f"Error: {e}")
          return None

• 도서 정보 가져오기

  from urllib import response
  from urllib.parse import urlparse, parse_qs
  import requests
  import json
  
  def searchAladin(item_id):
      try:
          url = f"http://www.aladin.co.kr/ttb/api/ItemLookUp.aspx?ttbkey={key}&itemIdType=ItemId&ItemId={item_id}&output=js&Version=20131101"
          response = requests.get(url)
          response_json = json.loads(response.text)
          title = response_json['item'][0]['title']
          mallType = response_json['item'][0]['mallType']
          priceSales= response_json['item'][0]['priceSales']
          return title+"@"+mallType+"@"+str(priceSales)
      except Exception as e:
          return None
  

  • 알라딘 검색 API활용
  • mallType에는 BOOK EBOOk 등이 있음

• 시트 작성 및 Telegram 전송

  import telegram
  from telegram.ext import Updater
  from telegram.ext import MessageHandler, Filters
  import gspread
  
  #telegram bot token
  token =""
  #알라딘 api key
  key= ""
  #스프레드 시트 key
  spreadsheet_key = ""
  #구글 서비스 어카운트
  gc = gspread.service_account(filename="")
  bot = telegram.Bot(token)
  
  #sheet open
  gcopen = gc.open_by_key(spreadsheet_key)
  
  #책정보를 가져옴
  for book_info in book_infos:
          book_info_split=book_info.split('@')
  				#sheet에 넣음
          sh.insert_row([date_split[0],"["+book_info_split[1]+"]"+book_info_split[0],"",user_name ,book_info_split[2],book_info_split[3]], int(line),value_input_option='USER_ENTERED')
  #신청확정메시지 전송
  bot.send_message(chat_id=_id, text=user_name+"님 도서 구매 요청 "+str(success_count)+"건 성공 했습니다.")

  • 시트에 edit 권한을 가진 구글 서비스 어카운트를 이용해 시트 수정

• 전체 코드

from urllib import response
from urllib.parse import urlparse, parse_qs
import telegram
from telegram.ext import Updater
from telegram.ext import MessageHandler, Filters
import gspread
from datetime import datetime
from pytz import timezone
import requests
import json

#telegram bot token
token =""
#알라딘 api key
key= ""
#스프레드 시트 key
spreadsheet_key = ""
#구글 서비스 어카운트
gc = gspread.service_account(filename="")
bot = telegram.Bot(token)


gcopen = gc.open_by_key(spreadsheet_key)

updater = Updater(token=token, use_context=True)
dispatcher = updater.dispatcher
updater.start_polling()
 
def handler(update, context):
    _id=update.message.chat.id
    user_text = update.message.text
    print(user_text)
    if user_text.startswith("/") == False:
        return

    if update.message.from_user.last_name is None:
        user_name=update.message.from_user.first_name
    else:
        user_name = update.message.from_user.last_name+update.message.from_user.first_name 
    
    date=datetime.now(timezone('Asia/Seoul'))
    sheet=str(date.month)+"월"
    date=date.strftime("'%Y-%m-%d '%H:%M:%S")
    sh=gcopen.worksheet(sheet)
    
    line= 2
    if "구매" in user_text :
        try:  
            user_text_split=user_text.split()
            count=0
            book_infos=[]
            success=True
            success_count=0
            for book in user_text_split:
                if count!=0:   
                    item_id=parsing_url(book)
                    print(item_id)
                    if item_id!=None:
                        result=searchAladin(item_id)
                        if result!=None:
                            book_infos.append(result+"@"+book)
                            success_count+=1
                        else:
                            success=False
                            break
                    else :
                        success=False
                        break
                else:
                    count+=1
            
            if success:
                date_split= str(date).split(' ')
                for book_info in book_infos:
                    book_info_split=book_info.split('@')
                    sh.insert_row([date_split[0],"["+book_info_split[1]+"]"+book_info_split[0],"",user_name ,book_info_split[2],book_info_split[3]], int(line),value_input_option='USER_ENTERED')
                bot.send_message(chat_id=_id, text=user_name+"님 도서 구매 요청 "+str(success_count)+"건 성공 했습니다.")
            else:
                bot.send_message(chat_id=_id, text=user_name+"님 도서 구매 요청에 실패하였습니다.\n 정상적인 알라딘 URL인지 확인해 주세요.")
        except Exception as e:
            print(f"Error during test: {e}")
            bot.send_message(chat_id=_id, text=user_name+"님 도서 구매 요청에 실패하였습니다.\n"+str(e))
    elif '/?' == user_text or '/help' == user_text:
        message="===도서 구매=== \n도서 구매 시 /구매 입력 후 알라딘 주소를 입력해주면 됩니다. \n여러권의 책을 구매하실 때는 띄어쓰기 혹은 줄바꿈으로 구분해주셔야합니다.\n\n도서 구매 현황은 https://docs.google.com/spreadsheets/d/1HJfOMeKSS3A2AcWRg7BPAI4T6y_tERwMV27_zfcXs1g/edit#gid=654922794 를 확인하시면 됩니다.\n추가적인 문의사항은 데브옵스팀 최정민에게 문의주시면 됩니다."
        bot.send_message(chat_id=_id, text=message)
 
echo_handler = MessageHandler(Filters.text, handler)
dispatcher.add_handler(echo_handler)



def parsing_url(url):
    try:
        parsed_url = urlparse(url)
        if parsed_url.hostname=="aladin.kr":
            print(url)
            final_url=get_final_url(url)
            parsed_url=urlparse(final_url)
            if parsing_url==None:
                return None
        query_params = parse_qs(parsed_url.query)
        
        item_id = query_params.get('ItemId', [None])[0]
        if item_id:
            return item_id
        else:
            item_id = query_params.get('itemid', [None])[0]
            if item_id:
                return item_id
            else:
                print("Error: ItemId not found in the URL.")
                return None

    except Exception as e:
        return None


def searchAladin(item_id):
    try:
        url = f"http://www.aladin.co.kr/ttb/api/ItemLookUp.aspx?ttbkey={key}&itemIdType=ItemId&ItemId={item_id}&output=js&Version=20131101"
        response = requests.get(url)
        response_json = json.loads(response.text)
        title = response_json['item'][0]['title']
        mallType = response_json['item'][0]['mallType']
        priceSales= response_json['item'][0]['priceSales']
        return title+"@"+mallType+"@"+str(priceSales)
    except Exception as e:
        return None


def get_final_url(url):
    try:
        response = requests.head(url, allow_redirects=True)
        final_url = response.url
        return final_url
    except requests.RequestException as e:
        print(f"Error: {e}")
        return None

환경

• 사내 관리용 EKS 클러스터에서 운영 중

결과

• 텔레그램

• Google Sheet

마무리하며

와탭랩스 데브옵스 엔지니어로 다양한 일을 하고 있습니다. 그 중 가장 뿌듯한 일 중 하나는 직원 분들이 어려움을 겪고 있을 때 도움을 주는 일인 것 같습니다.

해당 작업 이후, 도서 구매량이 많아진 건 안 비밀입니다!