Splunkのダウンロードリンクをスクレイピングで取得

6492 ワード
Splunk Python3 APIGateway AWSLambda Python3 テキストリンク

1.インストーラのダウンロードリンクを取得したい

業務でSplunkの検証環境をサクっと作りたいときに、AWSのEC2を利用しています。
建てたり壊したりをCloudFormationで自動化した際に、Splunkのインストーラについてダウンロードリンクを毎回サイトまで取りに行かないといけません。
そのリンク取得作業がなかなかに面倒なので、Pythonでスクレイピングを実装することにしました。

2.使用する機能や技術

  • Python 3.8
  • AWS Lambda
  • AWS API Gateway

API GatewayやLambdaへのコード実装は、先人の方がQiitaにたくさんナレッジを落としてくれているので、本記事では割愛します。
以下が参考にさせて頂いた記事になります。

【Python】AWS Lambdaで外部モジュールを使用する
AWS Lambdaで簡単なREST APIを作ってみた

3.ライブラリ

  • Beautiful Soup 4
  • requests
  • re

requestsでSplunkの公式サイトからHTMLを引っ張ってきます。
取得したHTMLをBeautiful Soupで読み込み、必要な情報をreを使って正規表現で抽出します。
スクレイピングは他にもやり方がたくさんあると思うのですが、画面遷移が必要とならないケースではrequestsを使うのがわかりやすいと思っています。
Seleniumはブラウザのドライバー管理も面倒だし、何より利用した経験が少ないので、今回は利用していません。

4.コード

早速ですが、以下が実装したコードになります。

lambda_function.py
import requests
import re
from bs4 import BeautifulSoup

def lambda_handler(event, context):
    # OS情報の取得
    os_type = event.get("os")

    # Linuxの場合は拡張子も取得する
    filename_extension = ""
    if os_type == "Linux":
        filename_extension = event.get("filename_extension")

    # InstallerTypeの取得
    installer_type = event.get("installer")

    # 対象バージョンの取得
    target_version = event.get("version")

    # 対象バージョンのHTMLタグを取得
    html_tag = get_old_installer_link(os_type, installer_type, filename_extension, target_version)
    # 対象バージョンがOlderReleasesに存在しなかった場合、最新バージョンを取得する
    if len(html_tag) == 0:
        html_tag = get_new_installer_link(os_type, installer_type, filename_extension)

    # 取得したタグからダウンロードリンクを抽出
    dl_link = dl_link_extraction(html_tag)

    # 実行結果の返却
    return {
        'statusCode': 200,
        'body': dl_link
    }


def get_old_installer_link(os, installer, extension, version):
    # installer毎に実行内容を分岐
    if installer == "EP":
        # EnterPrise
        # 旧バージョン取得
        old_r = requests.get('https://www.splunk.com/page/previous_releases')
        old_soup = BeautifulSoup(old_r.content, "html.parser")

        # os毎に実行内容を分岐
        if os == "Windows":
            html_list = old_soup.find_all("a", attrs={"data-version": version, "data-arch": "x86_64", "data-platform": "Windows"})
        elif os == "Linux":
            html_list = old_soup.find_all("a", attrs={"data-version": version, "data-arch": "x86_64", "data-platform": "Linux", "data-link": re.compile(r'\.' + extension)})

    elif installer == "UF":
        # UniversalForwarder
        # 旧バージョン取得
        old_r = requests.get('https://www.splunk.com/page/previous_releases/universalforwarder')
        old_soup = BeautifulSoup(old_r.content, "html.parser")

        # os 毎に実行内容を分岐
        if os == "Windows":
            html_list = old_soup.find_all("a", attrs={"data-version": version, "data-arch": "x86_64", "data-platform": "Windows"})
        elif os == "Linux":
            html_list = old_soup.find_all("a", attrs={"data-version": version, "data-arch": "x86_64", "data-platform": "Linux", "data-link": re.compile(r'\.' + extension)})

    return html_list


def get_new_installer_link(os, installer, extension):
    # installer毎に実行内容を分岐
    if installer == "EP":
        # EnterPrise
        # 新バージョン取得
        new_r = requests.get('https://www.splunk.com/ja_jp/download/splunk-enterprise.html')
        new_soup = BeautifulSoup(new_r.content, "html.parser")

        # os 毎に実行内容を分岐
        if os == "Windows":
            html_list = new_soup.find_all("a", attrs={"data-arch": "x86_64", "data-platform": "Windows"})
        elif os == "Linux":
            html_list = new_soup.find_all("a", attrs={"data-arch": "x86_64", "data-platform": "Linux", "data-link": re.compile(r'\.' + extension)})

    elif installer == "UF":
        # UniversalForwarder
        new_r = requests.get('https://www.splunk.com/ja_jp/download/universal-forwarder.html')
        new_soup = BeautifulSoup(new_r.content, "html.parser")

        # os 毎に実行内容を分岐
        if os == "Windows":
            html_list = new_soup.find_all("a", attrs={"data-arch": "x86_64", "data-platform": "Windows"})
        elif os == "Linux":
            html_list = new_soup.find_all("a", attrs={"data-arch": "x86_64", "data-platform": "Linux", "data-link": re.compile(r'\.' + extension)})

    return html_list


def dl_link_extraction(tag):
    # 正規表現でダウンロードリンクを抽出
    link = re.search(r'data-link=\"([^\"]+)\"', str(tag[0])).group(1)
    return link

基本的にエラーコントロールはしていません。
どんなことになってもステータスは200で返します。
本当はその辺もしっかりしたほうが良いのですが、今回はこれで勘弁してください。

5.実行例

API Gatewayへパラメータを付与し、curlでエンドポイントを叩きます。
付与するパラメータはインストール先のOS、対象となるインストーラ、インストーラの拡張子、対象となるバージョンの4種です。
各パラメータに対応している設定値は以下になります。
(それ以外を入れるとダウンロードリンクが返ってきません...)

  • os
    • Windows
    • Linux
  • installer
    • EP
    • UF
  • filename_extension
    • tgz
    • rpm
  • version

Windowsの場合は拡張子を選択する必要はないので、filename_extensionが空でも問題ありません。
以下はLinuxで7.2.3のUniversal Forwarderインストーラをtgzで取得するケースです。

curl -X POST "https://xxxxxxxxxx.execute-api.ap-xxxxxxxxx-x.amazonaws.com/xxx/xxxxx" -d "{\"os\": \"Linux\",\"installer\": \"UF\",\"version\": \"7.2.3\",\"filename_extension\": \"tgz\"}"

以下が実行結果になります。
ステータスコードとダウンロードリンクが返却されます。
(ステータスコードはかならず200で帰ってきます...)

6.まとめ

ザっと書いたとはいえ、結構荒いコードになっているなぁ、実感しました。
機会があったら修正しようと思います。

なにはともあれ、これでダウンロードリンクをスクレイピングで取得できるようになりました。
CloudFormationのuserdataにcurl文を追加し、ダウンロードリンクをEC2インスタンスの作成中に取得できるようになりました。
いちいちダウンロードリンクを取ってきて、張り付ける作業から解放されたのはうれしい...
次はCloudFormationを書かなきゃ...