cts_web_crawler.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "c0905ec4662730ed",
   "metadata": {
    "collapsed": false
   },
   "source": [
    "# [관세청 수출입무역통계 사이트](https://tradedata.go.kr/cts/index.do) 크롤러 개발\n",
    "- webdriver 버전"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "id": "initial_id",
   "metadata": {
    "ExecuteTime": {
     "end_time": "2023-10-15T07:50:29.645319400Z",
     "start_time": "2023-10-15T07:50:23.204716800Z"
    },
    "collapsed": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "✅Alerted\n",
      "<class 'Exception'>\n"
     ]
    }
   ],
   "source": [
    "# latest selenium version (October 9, 2023)\n",
    "# !pip install selenium==\"4.14.0\"\n",
    "# !python.exe -m pip install --upgrade pip\n",
    "\n",
    "from selenium import webdriver\n",
    "from selenium.webdriver.edge.service import Service\n",
    "from selenium.webdriver.common.by import By\n",
    "from selenium.webdriver.edge.options import Options\n",
    "from selenium.webdriver.common.keys import Keys\n",
    "# from selenium.webdriver.common.alert import Alert\n",
    "from selenium.webdriver.support.ui import WebDriverWait\n",
    "from selenium.webdriver.support import expected_conditions as EC\n",
    "from selenium.common.exceptions import TimeoutException\n",
    "\n",
    "import time\n",
    "\n",
    "# 반응형 웹 오류 대비해 옵션 설정\n",
    "options = Options()\n",
    "options.add_argument(\"--start-maximized\")   # 전체화면으로 실행\n",
    "options.add_argument(\"--disable-infobars\")  # 정보바 숨김\n",
    "# disable the banner \"Chrome is being controlled by automated test software\"\n",
    "# edgedriver는 해당 실행파일과 같은 경로에 위치해야 함\n",
    "edge_driver_path = \"./msedgedriver.exe\"\n",
    "# 기본 edge 브라우저 경로 설정\n",
    "options.binary_location = \"C:\\Program Files (x86)\\Microsoft\\Edge\\Application\\msedge.exe\"\n",
    "driver = webdriver.Edge(options = options)\n",
    "driver.get('https://tradedata.go.kr/cts/index.do')\n",
    "hs_codes = [\n",
    "    \"8479501000\",\n",
    "    \"8479502000\",\n",
    "    \"8479509000\",\n",
    "    \"8515211010\",\n",
    "    \"8515212010\",\n",
    "    \"8515213010\",\n",
    "    \"8515219010\",\n",
    "    \"8515311010\",\n",
    "    \"8515319010\",\n",
    "    \"8486309020\",\n",
    "    \"8424202010\",\n",
    "    \"8479892010\", \n",
    "\t\"8479892090\", \n",
    "\t\"8428701000\", \n",
    "\t\"8427103000\", \n",
    "\t\"8428702000\", \n",
    "\t\"8428709000\", \n",
    "\t\"8508111000\", \n",
    "\t\"8508192000\"\n",
    "]\n",
    "\n",
    "def click_export_import_statistics(driver):\n",
    "    \"\"\"\n",
    "    Clicks on the '수출입통계' -> '수출입통계 실적' dropdown menu in the top navigation bar.\n",
    "\n",
    "    Args:\n",
    "        driver: The Selenium WebDriver instance to use.\n",
    "\n",
    "    Returns:\n",
    "        None.\n",
    "    \"\"\"\n",
    "    # Click on the '수출입통계' dropdown menu\n",
    "    dropdown_menu = driver.find_element(by=\"xpath\", value='//*[@id=\"topMenuArea\"]/li[1]')\n",
    "    dropdown_menu.click()\n",
    "\n",
    "    # Wait for the dropdown menu to expand\n",
    "    time.sleep(1)\n",
    "\n",
    "    # Click on the '수출입통계 실적' link\n",
    "    link = driver.find_element(by=\"xpath\", value='//*[@id=\"topMenuArea\"]/li[1]/ul/li[2]/a')\n",
    "    link.click()\n",
    "\n",
    "    # Wait for the page to load\n",
    "    time.sleep(1)\n",
    "\n",
    "def change_statistics_type(driver):\n",
    "    \"\"\"\n",
    "    Changes the statistics type to '품목별+국가별' in the left navigation bar.\n",
    "\n",
    "    Args:\n",
    "        driver: The Selenium WebDriver instance to use.\n",
    "\n",
    "    Returns:\n",
    "        None.\n",
    "    \"\"\"\n",
    "    # Click on the '품목별+국가별' radio button\n",
    "    radio_button = driver.find_element(by=\"xpath\", value='//*[@id=\"titleType\"]')\n",
    "    radio_button.click()\n",
    "\n",
    "    # Wait for the radio button to be selected\n",
    "    time.sleep(1)\n",
    "\n",
    "    # Click on the '조회' button\n",
    "    button = driver.find_element(by=\"xpath\", value='//*[@id=\"radioListType\"]/li[2]')\n",
    "    button.click()\n",
    "\n",
    "    # Wait for the page to load\n",
    "    time.sleep(1)\n",
    "\n",
    "def enter_hs_codes(driver, hs_codes):\n",
    "    \"\"\"\n",
    "    Enters the given HS codes into the search input field.\n",
    "\n",
    "    Args:\n",
    "        driver: The Selenium WebDriver instance to use.\n",
    "        hs_codes: A list of HS codes to enter.\n",
    "\n",
    "    Returns:\n",
    "        None.\n",
    "    \"\"\"\n",
    "    # Find the search input field and click on it\n",
    "    search_input = driver.find_element(by=\"xpath\", value='//*[@id=\"ETS0100019Q_hsSgn\"]')\n",
    "    search_input.click()\n",
    "\n",
    "    # Wait for the input field to be ready\n",
    "    time.sleep(0.5)\n",
    "\n",
    "    # Enter each HS code and press Enter\n",
    "    for hs_code in hs_codes:\n",
    "        search_input.send_keys(hs_code)\n",
    "        time.sleep(0.1)\n",
    "        search_input.send_keys(Keys.ENTER)\n",
    "        time.sleep(0.1)\n",
    "\n",
    "def monthly_search(driver):\n",
    "    \"\"\"\n",
    "    Clicks on the '월별 조회' radio button and selects the most recent month.\n",
    "\n",
    "    Args:\n",
    "        driver: The Selenium WebDriver instance to use.\n",
    "\n",
    "    Returns:\n",
    "        None.\n",
    "    \"\"\"\n",
    "    # Click on the '월별 조회' radio button\n",
    "    radio_button = driver.find_element(by=\"xpath\", value='//*[@id=\"filterCon\"]/div[11]/div[1]/div[2]/ul/li[2]')\n",
    "    radio_button.click()\n",
    "\n",
    "    # Wait for the radio button to be selected\n",
    "    time.sleep(0.1)\n",
    "\n",
    "    # Select the most recent month\n",
    "    month_option = driver.find_element(by=\"xpath\", value='//*[@id=\"ETS0100019Q_formYearMonthPc\"]/option[1]')\n",
    "    month_option.click()\n",
    "\n",
    "    # Wait for the page to load\n",
    "    time.sleep(1)\n",
    "\n",
    "    # Click on the '조회' button\n",
    "    search_button = driver.find_element(by=\"xpath\", value='//*[@id=\"filterCon\"]/div[14]/button[2]')\n",
    "    search_button.click()\n",
    "\n",
    "    # Wait for the page to load\n",
    "    time.sleep(1)\n",
    "\n",
    "def yearly_search(driver):\n",
    "    \"\"\"\n",
    "    Changes the search type to yearly and clicks the search button.\n",
    "\n",
    "    Args:\n",
    "        driver: The Selenium WebDriver instance to use.\n",
    "\n",
    "    Returns:\n",
    "        None.\n",
    "    \"\"\"\n",
    "    # Click on the '연도별' radio button\n",
    "    radio_button = driver.find_element(by=\"xpath\", value='//*[@id=\"filterCon\"]/div[11]/div[1]/div[2]/ul/li[1]')\n",
    "    radio_button.click()\n",
    "\n",
    "    # Wait for the radio button to be selected\n",
    "    time.sleep(1)\n",
    "\n",
    "    # Click on the '조회' button\n",
    "    search_button = driver.find_element(by=\"xpath\", value='//*[@id=\"filterCon\"]/div[14]/button[2]')\n",
    "    search_button.click()\n",
    "\n",
    "    # Wait for the page to load\n",
    "    time.sleep(1)\n",
    "\n",
    "def download_trade_data(driver):\n",
    "    \"\"\"\n",
    "    Downloads the monthly export/import trade data in Excel format.\n",
    "\n",
    "    Args:\n",
    "        driver: The Selenium WebDriver instance to use.\n",
    "\n",
    "    Returns:\n",
    "        None.\n",
    "    \"\"\"\n",
    "    # Click on the '엑셀 다운로드' button\n",
    "    download_button = driver.find_element(by=\"xpath\", value='//*[@id=\"mainarea\"]/div/div[1]/div[2]/div[2]/div/div/button')\n",
    "    download_button.click()\n",
    "\n",
    "    # Wait for the download button to be ready\n",
    "    time.sleep(1)\n",
    "\n",
    "    # Click on the data download button\n",
    "    download_button = driver.find_element(by=\"xpath\", value='//*[@id=\"ETS0100019Q_btnXls\"]')\n",
    "    download_button.click()\n",
    "\n",
    "    # Wait for the download to complete\n",
    "    time.sleep(10)\n",
    "\n",
    "\n",
    "\n",
    "driver.execute_script(\"let a = alert('⌛5초 뒤 웹을 자동으로 조작합니다.\\\\n 원활한 작업을 위해 마우스/키보드 조작을 금해주세요🚫');document.body.setAttribute('data-id', a)\")\n",
    "try:\n",
    "    # 5초 간 기다리는데, alert이 뜨면 바로 다음으로 넘어감\n",
    "    WebDriverWait(driver, 5).until(EC.alert_is_present(),\n",
    "                                   'Timed out waiting for PA creation ' +\n",
    "                                   'confirmation popup to appear.')\n",
    "    print(\"✅Alerted\")\n",
    "    alert = driver.switch_to.alert\n",
    "    time.sleep(5)  # must \n",
    "\n",
    "    # 취소 버튼 클릭 시 의도적으로 오류 발생해 다음 절차(finally) 진행\n",
    "    raise Exception(alert.accept())\n",
    "    \n",
    "except TimeoutException:\n",
    "    print(\"⚠️오류가 발생해 프로그램을 종료합니다.\")\n",
    "    driver.quit()\n",
    "\n",
    "except Exception as e:\n",
    "    print(type(e))\n",
    "\n",
    "\n",
    "finally:\n",
    "    click_export_import_statistics(driver)\n",
    "    change_statistics_type(driver)\n",
    "    enter_hs_codes(driver, hs_codes)\n",
    "\n",
    "    # Scroll down to the bottom of the page\n",
    "    driver.execute_script(\"window.scrollTo(0, window.scrollY + 1000)\")\n",
    "    time.sleep(0.5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "id": "ed170a9d",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "✅Alerted\n",
      "<class 'selenium.common.exceptions.NoAlertPresentException'>\n",
      "🖱️사용자로부터 입력받은 버튼: false\n",
      "🗓️연도별 조회\n"
     ]
    }
   ],
   "source": [
    "def download_visualization_data(driver):\n",
    "    \"\"\"\n",
    "    Downloads the visualization data for export/import statistics in bar chart format.\n",
    "\n",
    "    Args:\n",
    "        driver: The Selenium WebDriver instance to use.\n",
    "\n",
    "    Returns:\n",
    "        None.\n",
    "    \"\"\"\n",
    "    # Click on the '막대' chart button\n",
    "    bar_chart = driver.find_element(by=\"xpath\", value='//*[@id=\"ETS0100019Q_btnChartColumn\"]')\n",
    "    bar_chart.click()\n",
    "\n",
    "    # # Wait for the chart to load\n",
    "    # time.sleep(1)\n",
    "    # driver.switch_to.window(driver.window_handles[0])\n",
    "    # blank_area.click()\n",
    "\n",
    "    # Click on the context menu button\n",
    "    context_menu = driver.find_element(by=\"xpath\", value='/html/body/div[3]/section/div[1]/div/div[1]/div[2]/div[4]/div[2]/div/div[3]/div[2]/div[3]/button')\n",
    "    print(context_menu)\n",
    "    context_menu.click()\n",
    "\n",
    "    # Wait for the context menu to appear\n",
    "    time.sleep(0.5)\n",
    "\n",
    "    # Click on the '이미지 다운로드' button\n",
    "    download_button = driver.find_element(by=\"xpath\", value='/html/body/div[3]/section/div[1]/div/div[1]/div[2]/div[4]/div[2]/div/div[3]/div[3]/ul/li[3]')\n",
    "    download_button.click()\n",
    "\n",
    "    # Wait for the download to complete\n",
    "    time.sleep(5)\n",
    "\n",
    "driver.execute_script(\"let a = confirm('⌛5초 뒤 월별 데이터 수집을 진행합니다.\\\\n[취소] 버튼 클릭시 연도별 데이터를 수집합니다.');document.body.setAttribute('data-id', a)\")\n",
    "try:\n",
    "    # 5초 간 기다리는데, alert이 뜨면 바로 다음으로 넘어감\n",
    "    WebDriverWait(driver, 5).until(EC.alert_is_present(),\n",
    "                                   'Timed out waiting for PA creation ' +\n",
    "                                   'confirmation popup to appear.')\n",
    "    print(\"✅Alerted\")\n",
    "    alert = driver.switch_to.alert\n",
    "    time.sleep(5)  # must \n",
    "\n",
    "    # 취소 버튼 클릭 시 의도적으로 오류 발생해 다음 절차(finally) 진행\n",
    "    raise Exception(alert.accept())\n",
    "    \n",
    "except TimeoutException:\n",
    "    print(\"⚠️오류가 발생해 프로그램을 종료합니다.\")\n",
    "    driver.quit()\n",
    "\n",
    "except Exception as e:\n",
    "    print(type(e))\n",
    "\n",
    "finally:\n",
    "    promt_text = driver.find_element(By.TAG_NAME, 'body').get_attribute('data-id') # get the text from the alert\n",
    "    # 다운로드 이벤트 후 발생하는 윈도우로 인해 다음 작업 진행 위한 페이지 빈 공간 지정\n",
    "    blank_area = driver.find_element(By.XPATH, value='/html/body/div[3]/section/div[1]/div/div[1]/div[1]/div[1]/div[1]')\n",
    "\n",
    "    print(\"🖱️사용자로부터 입력받은 버튼: \"+promt_text)\n",
    "\n",
    "    if promt_text == \"true\":\n",
    "        print('🌕월별 조회')\n",
    "        monthly_search(driver)\n",
    "        download_trade_data(driver)\n",
    "        # driver.switch_to.window(driver.window_handles[0])\n",
    "        # download_visualization_data(driver)\n",
    "\n",
    "    elif promt_text == \"false\":\n",
    "        print('🗓️연도별 조회')\n",
    "        yearly_search(driver)\n",
    "        download_trade_data(driver)\n",
    "        # driver.switch_to.window(driver.window_handles[0])\n",
    "        # blank_area.click()\n",
    "        # time.sleep(0.5)\n",
    "        # download_visualization_data(driver)\n",
    "\n",
    "    else:\n",
    "        print(\"⚠️오류가 발생해 프로그램을 종료합니다.\")\n",
    "        driver.quit()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "id": "4ef77a4a",
   "metadata": {},
   "outputs": [],
   "source": [
    "# # 이미지 다운로드(하기 로직은 다른 프로그램으로 대체)\n",
    "# context_menu = driver.find_element(by=\"xpath\", value='/html/body/div[3]/section/div[1]/div/div[1]/div[2]/div[2]/div/button[2]')\n",
    "# print(context_menu)\n",
    "# context_menu.click()\n",
    "# # Wait for the context menu to appear\n",
    "# time.sleep(0.5)\n",
    "\n",
    "# # Click on the '이미지 다운로드' button\n",
    "# download_button = driver.find_element(by=\"xpath\", value='/html/body/div[3]/section/div[1]/div/div[1]/div[2]/div[4]/div[2]/div/div[3]/div[3]/ul/li[3]')\n",
    "# download_button.click()\n",
    "#### 웹상 '파이' 그래프는 효용성 떨어져서 제외 ####\n",
    "# # '파이' 그래프 선택 (금액) 후 이미지 다운로드\n",
    "# pie_chart = driver.find_element(by=\"xpath\", value='//*[@id=\"ETS0100019Q_btnChartPie\"]')\n",
    "# pie_chart.click()\n",
    "# WebDriverWait(driver, 20).until(EC.element_to_be_clickable((By.CLASS_NAME, \"highcharts-menu\"))).click()\n",
    "# time.sleep(1)\n",
    "# 파이차트 컨텍스트 메뉴 버튼 조작\n",
    "# chart_context_menu = driver.find_element(by=\"xpath\", value='/html/body/div[3]/section/div[1]/div/div[1]/div[2]/div[4]/div[2]/div/div[3]/div[2]/div[2]/button')\n",
    "# chart_context_menu.click()\n",
    "# time.sleep(0.5)\n",
    "# chart_img_download = driver.find_element(by=\"xpath\", value='/html/body/div[3]/section/div[1]/div/div[1]/div[2]/div[4]/div[2]/div/div[3]/div[3]/ul/li[3]')\n",
    "# chart_img_download.click()\n",
    "\n",
    "time.sleep(2)\n",
    "driver.quit()"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.11.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}