トップ > techblog > Alibaba CloudではじめるWebスクレイピング

Alibaba CloudではじめるWebスクレイピング

techblog

はじめまして。7月からSBクラウドにジョインしたソリューションアーキテクトの石井です。 入社直後のお勉強期間で作ったものをアピールしていた結果、もすけ先輩よりブログ書けよと仰せつかりまして初めての投稿をさせていただきます。

はじめに

みなさまWebスクレイピングはご存知でしょうか？ 非常に端的に説明しますと、なんらかのWebページの中から必要な情報を抜き出して加工することです。 あまりいい例が思いつかず恐縮ですが、例えばニュースサイトのヘッドラインになっているニュースのタイトルだけをプログラムを利用することで自動的に抽出するといったことがWebスクレイピングの一例です。 単純に一度切りの話であれば直接サイトを見に行ってもいいのですが、何らかの理由で定期的に同じところの情報を取得したい場合などは、いちいち人手で実施するより、プログラムを書いて自動化してしまえば効率的ですよね！ 今回はAlibabaクラウドのFunction Computeを利用して、定期的に情報を取得する方法について書こうと思います。

この記事では私が毎朝行っている、いろいろな情報収集を少しでも簡単に行うことを例にとって説明します。 おそらく皆様も、ニュースサイトをはじめ、興味のあるサイトを日々巡回されていると思います。 どんなツールを使っても巡回するサイトが増えてくると面倒なもの・・・一箇所位でまとめて、最新の情報だけ見たいと思っているのはきっと私だけではないと思います。 今回はそういった問題を解消するための情報ポータル的なものを作成することを目指します。

ゴール

複数のWebサイトから、必要な情報を抽出、マージしたものを作成するところがゴールとなりますが、ここの説明では単純化のため、１つのWebサイトのみにしています。 出力されるフォーマットとしてはマークダウンとし、以下のようなものが出力されるものとします。

環境

今回は標準外のライブラリを用意するため、ローカル環境に以下のセットアップが必要です。

# 項目 値
1 言語 Python 3.x
2 ツール類 aliyuncli(導入しておいてください), ossutil, fcli(本記事の中でインストール方法をご紹介します)
3 OS 上記動けばなんでもよいです (本検証ではECSにUbuntu 16.04を入れて開発環境としています)

構成

いろいろな設計思想などあると思うのですが、概ね以下のような構成とします。 ① 対象のページやRSSをダウンロードし、OSS上に1次保管する（今回は例としてSBCloudEngineer BlogのRSSを取得します） ② ①でダウンロードしたページごとの特性に応じて情報を抜き出し、テキストとしてOSS上に保管する ③ ②で抜き出したデータを統合し、OSS上に保管、公開する。

作業開始の前に

作成する関数単位でディレクトリを区切っていきます。ここでは説明の便宜上、以下の様なディレクトリ構成であるとします。

# パス 役割
1 ~/dev 開発のルートディレクトリ
2 ~/dev/scraping 今回作成する関数群のプロジェクトディレクトリ
3 ~/dev/scraping/downloader ダウンロード関数を格納するディレクトリ
4 ~/dev/scraping/sbcloud-scraping SBCloudのRSSを解析し、必要なデータを抜き出す関数を格納するディレクトリ
5 ~/dev/scraping/aggregator 抜き出した必要なデータを纏めるディレクトリ

fcliの導入

本質とは関係ありませんが、Function Computeをコマンドラインから操作するfcliというツールがあります。 それの日本語インストールマニュアルがなかったので、どういうふうに導入したかも説明します。 以下サイトから最新版のバイナリ(zip)をダウンロードしてください。本記事を書いたときは0.20が最新でした。 https://github.com/aliyun/fcli/releases

ダウンロードしたものを解答します。

$ unzip fcli-v0.20-linux-amd64.zip

解答されたバイナリをPATHの通っているところに移動させます。私の場合は/usr/local/binとしました。

$ sudo mv fcli /usr/local/bin

これでコマンド自体は利用できるようになったはずです。次に初期セットアップに移ります。

$ fcli

これを実行すると初回のインタラクティブな設定画面が起動します。 事前にAccessKey/SecretKeyを取得しておいてください。

以上でコマンドラインツールが利用可能になりました！ Function Computeでは関数を作成する前にServiceという関数の入れ物（プロジェクトのようなもの）を用意する必要があります。 今回はScrapingというService名で作成します。

$ fcli service create --service-name Scraping

正しく作成されたことを確認するには、以下コマンドを利用します。

$ fcli service list

ossutilの導入

同様にOSSを操作するためのCLIツールであるossutilを導入します。 Go言語で書かれているためGoは別途導入しておいてください。 その後以下コマンドでインストール可能です。

$ go get github.com/aliyun/ossutil 初期設定は以下コマンドで実施します。 $ ossutil config

OSS Bucketの準備

ここでは3つのBucketを用意します。 Bucket名は全ユーザで一意である必要がありますので、ここでは以下名称としますが、 これ以降ではご自身で作成されたもので読み替えてください。

# バケット名 役割
1 my-download-bucket000 ダウンロードしたファイルをキャッシュしておく領域
2 my-scraping-bucket000 個々のファイルから抽出した情報を置いておく領域
3 my-public-bucket000 抽出結果をマージし、公開する領域

それぞれ以下のコマンドで作成します。

$ ossutil mb oss://my-download-bucket000 --acl=private
$ ossutil mb oss://my-scraping-bucket000 --acl=private
$ ossutil mb oss://my-public-bucket000 --acl=private

おっと、3の公開バケットまでプライベートにしてしまいました。 そんなときは以下コマンドでACLを変更できます。

$ ossutil set-acl oss://my-public-bucket000 public-read -b

OSSは一度バケットを作ると削除しても同じ名前のバケットは作れないので、 間違えた場合は変更コマンドで乗り切ってください。 リージョンを間違えてしまうとどうしようもないですが・・・

STS Policyの作成

今回Function Compute上に作成したScrapingサービスはFunction自体の実行に加え、OSSバケットを読み取る必要があります。 Function上でAccessKey/SecretKeyをそ用いて認証しても動作はしますが、セキュリティの問題上推奨されません。 事前に以下のようなRoleを定義し、Scrapingサービスにアタッチすることでコード上にAccessKey/SecretKeyを記載することなくOSSへアクセスすることが可能となります。

そのためには作成したScrapingServiceにAssumeRoleを許可する必要があります。 まずはrole.jsonを以下のように作成します。

{
  "Statement": [
    {
      "Action": "sts:AssumeRole",
      "Effect": "Allow",
      "Principal": {
        "Service": [
          "fc.aliyuncs.com"
        ]
      }
    }
  ],
  "Version": "1"
}

roleを作成します。ここでrole名はexecute-fcとしています。

$ aliyuncli ram CreateRole --RoleName "execute-fc" --AssumeRolePolicyDocument "$(cat ./role.json)"

次にこのroleに対して、どんなポリシーを割り当てるかを設定します。 ここでは、前に作成したOSSバケットに対する読み書きのみ許可します。 以下のようにpolicy.jsonを作成してください。

{
    "Version": "1",
    "Statement": [
        {
            "Action": [
                "oss:GetBucketAcl",
                "oss:ListObjects"
            ],
            "Resource": [
                "acs:oss:*:アカウントid:my-download-bucket000",
                "acs:oss:*:アカウントid:my-public-bucket000",
                "acs:oss:*:アカウントid:my-scraping-bucket000"
            ],
            "Effect": "Allow"
        },
        {
            "Action": [
                "oss:PutObject",
                "oss:GetObject",
                "oss:DeleteObject"
            ],
            "Resource": [
                "acs:oss:*:アカウントid:my-download-bucket000/*",
                "acs:oss:*:アカウントid:my-public-bucket000/*",
                "acs:oss:*:アカウントid:my-scraping-bucket000/*"
            ],
            "Effect": "Allow"
        }
    ]
}

policyを作成します。

$ aliyuncli ram CreatePolicy --PolicyName "OperateScrapingOSS" --PolicyDocument "$(cat ./policy.json)"

次にroleにpolicyをアタッチします。

$ aliyuncli ram AttachPolicyToRole --PolicyName OperateScrapingOSS --RoleName "execute-fc" --PolicyType "Custom"

このあとの設定でこのroleのarnが必要になりますので、以下コマンドで確認します。

$ aliyuncli ram ListRoles
...略...
            {
                "RoleName": "execute-fc",
                "CreateDate": "2018-07-15T13:06:41Z",
                "Description": "",
                "Arn": "acs:ram::アカウントid:role/execute-fc",
                "RoleId": "ロールID"
            },
...略...

事前に定義されているもの含め、たくさんのRoleが表示されます。 Role名がexecute-fcになっているものの、"Arn"を控えてください。

それでは最後に先ほど作成したサービスにこのroleをアタッチしましょう。

$ fcli service update --service-name Scraping --role "先程のコマンドで表示したArn"

以上でScrapingサービスに作成するFunctionからSTSが利用できるようになりました！

Functionの作成① 情報を持っているページやRSSをダウンロード

前置きが長くなりましたが、ここからFunctionの作成に入ります。

ここは単純に事前に設定したURLから情報をダウンロードするスクリプトを作成します。 ダウンロードしたものはOSS上に保管します。

ダウンロードと処理するところは一緒のスクリプトにしてしまってもよいのですが、そうするとテストのたびにアクセスが発生することになり、場合によっては迷惑をかけてしまうこともあることや1つの関数は1つのことをうまくやるというUnix的な思想に基づき、ダウンロードだけを行う関数として実装します。 ファイル名はなんでもいいのですが、デフォルトのままのほうが簡単なので、index.pyという名前で書いていきます。 まずは関数を作成するディレクトリに移動してから作業を開始します。 $ cd ~/dev/scraping/downloader

以下のようにindex.pyを作成します。

# -*- coding: utf-8 -*-
import logging
import requests
import oss2

def handler(event, context):
  logger = logging.getLogger()
  name = "SBCloudEngineerBlog"
  url = "https://techblog.sbcloud.co.jp/feed/"
  logger.info("Start Downloading: " + url)
  res = requests.get(url)
  creds = context.credentials
  auth = oss2.StsAuth(creds.accessKeyId, creds.accessKeySecret, creds.securityToken)
  bucket = oss2.Bucket (auth, 'oss-ap-northeast-1.aliyuncs.com', 'my-download-bucket000')
  bucket.put_object(name + ".xml", res.text)
  return 'Done'

作成できたら、以下のコマンドで新しいFunctionを作成します。

$ fcli function create --service-name Scraping --function-name MyDownloader --runtime python3 --handler index.handler --code-dir ./

成功したら以下コマンドで関数を実行できます。試しに実行して、OSS上にファイルが作成されていることを確認してください。

$ fcli function invoke --service-name Scraping --function-name MyDownloader

もし、うまく動かなかったりして、ソースコードを修正した場合は以下コマンドでアップデートできます。エラーが出てしまった場合はコードを修正してから再度試してみてください。

$ fcli function update --code-dir ./ --function-name MyDownloader --service-name Scraping

Functionの作成② ページごとの特性に応じて情報を抜き出す(RSS編)

RSSの解析は非常に簡単です。 ここではfeedparserというライブラリを利用します。

まずは、この関数のためのディレクトリに移動します。

$ cd ~/dev/scraping/sbcloud-scraping

次にfeedparserをこのディレクトリ内にインストールします。 標準外のライブラリを利用する場合はこのようにローカル側での用意が必要となります。 $ pip3 install -t ./ feedparser

インストールが成功すると、以下のようなファイルができているはずです。

$ ls
feedparser-5.2.1-py3.7.egg-info feedparser.py __pycache__

次にソースコードをindex.pyというファイル名で書きます。 ここでは、例として直近1週間分の記事のみ取得するようにします。

# -*- coding: utf-8 -*-
import logging
from datetime import datetime, timedelta, timezone
import oss2
import feedparser
from time import mktime


def handler(event, context):
  logger = logging.getLogger()

  creds = context.credentials
  auth = oss2.StsAuth(creds.accessKeyId, creds.accessKeySecret, creds.securityToken)
  bucket = oss2.Bucket (auth, 'oss-ap-northeast-1.aliyuncs.com', 'my-download-bucket000')

  md = "## SBCloud Engineer Blog\n"
  xml = bucket.get_object('SBCloudEngineerBlog.xml').read()
  feed = feedparser.parse(xml)
  JST = timezone(timedelta(hours=+9), 'JST')
  today = datetime.now(JST)
  for item in feed["entries"]:
    d = datetime.fromtimestamp(mktime(item["published_parsed"]),JST)
    if today - timedelta(days = 7) <= d:
      md = md + "- " + d.strftime('%Y/%m/%d') + " [" + item["title"] + "](" + item["link"] + ")\n"

  bucket = oss2.Bucket (auth, 'oss-ap-northeast-1.aliyuncs.com', 'my-scraping-bucket000')
  bucket.put_object('SBCloudEngineerBlog.md', md)
  return 'Done'

 

先ほどと同じように以下コマンドで作成＆実行が可能です。

$ fcli function create --service-name Scraping --function-name MySBCloudScraping --runtime python3 --handler index.handler --code-dir ./
$ fcli function invoke --service-name Scraping --function-name MySBCloudScraping

この例ではSBCloudEngineer Blogだけですが、必要に応じて上記のようなものを複数作成します。

Functionの作成③ 抜き出した情報をまとめて表示する

この例では1つだけなので不要なのですが、実際には必要になるので記載します。 本来はFunctionの作成②はSBCloud Engineer Blogだけではなく、ソースごとにファイルを作成しますので、それをまとめる処理が必要になります。 このファイルもindex.pyという名前にします。

# -*- coding: utf-8 -*-
import logging
import oss2
from datetime import datetime, timedelta, timezone

def handler(event, context):
  logger = logging.getLogger()
  targets = ["SBCloudEngineerBlog.md"]
  creds = context.credentials
  auth = oss2.StsAuth(creds.accessKeyId, creds.accessKeySecret, creds.securityToken)
  output_bucket = oss2.Bucket (auth, 'oss-ap-northeast-1.aliyuncs.com', 'my-public-bucket000')
  input_bucket = oss2.Bucket (auth, 'oss-ap-northeast-1.aliyuncs.com', 'my-scraping-bucket000')

  md = "# Today's Topics\n"
  JST = timezone(timedelta(hours=+9), 'JST')
  today = datetime.now(JST)
  md = md + "last update: " + today.strftime('%Y/%m/%d %H:%M') + "\n"
  for target in targets:
    md = md + "\n" + input_bucket.get_object(target).read().decode("utf-8")

  output_bucket.put_object('index.md', md)
  output_bucket.put_object(today.strftime('%Y%m%d') + '.md', md)

  return 'Done'

これまでと同様、以下の様なコマンドで実行できます。

$ fcli function create --service-name Scraping --function-name MyAggregator --runtime python3 --handler index.handler --code-dir ./ 
$ fcli function invoke --service-name Scraping --function-name MyAggregator

完成したものはバケットがパブリックなので、ブラウザからダウンロードしても構いませんし、コマンドでダウンロードしても構いません。コマンドでダウンロードする場合は以下のようにしてください。

$ ossutil cp oss://my-public-bucket000/index.md .
$ cat index.md

実行した日にちによってやや見栄えはかわりますが、以下のようになります。

トリガーの設定

さて、ここまででFunctionは完成しましたが、せっかくなら手動ではなく自動で実行するようにしたいですよね！ ここではそれぞれ以下のトリガーとします。 MyDownloader MySBCloudScraping MyAggregator

それではそれぞれ設定していきましょう。 ちなみにですが、Function Computeでトリガーを時刻で指定するときはUTC時間で表現する必要があります。 JSTはUTC+9時間ですので、-9時間します。 つまり、7時に実行したい場合は、7時-9時間で22時を指定します。

それでは、トリガーの設定を作成します。 1つ目に毎朝7時に実行する場合のものを作成します。 downloadTrigger.ymlという名前で、以下のファイルを作成してください。

triggerConfig:
    payload: "download"
    cronExpression: "0 0 22 * * *"
    enable: true

同様に7時5分に実行する場合のものをaggregationTrigger.ymlという名前で作成しておきます。

triggerConfig:
    payload: "aggregation"
    cronExpression: "0 5 22 * * *"
    enable: true

トリガーの設定はそれぞれ以下のコマンドを実行します。

$ fcli trigger create --service-name Scraping --function-name MyDownloader --trigger-name DownloadTrigger --type timer --config ./downloadTrigger.yml
$ fcli trigger create --service-name Scraping --function-name MyAggregator --trigger-name AggregationTrigger --type timer --config ./aggregationTrigger.yml

さて、最後にossにダウンロードされたことを検知したら自動で解析を始める部分のトリガーを作成します。 ここも先程のSTSが必要です。もう少し権限を絞ったroleを作成しても良さそうですが、長くなるのでここでは先ほど作成したextecute-fc roleをそのまま作成します。気になる人や本番環境では別のroleを作成したほうが影響範囲を最小にできるので良いかもしれません。

以下のような内容でscrapingTrigger.ymlを作成します。

triggerConfig:
    events:
        - oss:ObjectCreated:PostObject
        - oss:ObjectCreated:PutObject
    filter:
        key:
            prefix: SBCloudEngineerBlog
            suffix: .md

そして以下コマンドでトリガーを作成します。上の時刻のときと違い、roleが必要な点に注意してください。

$ fcli trigger create --service-name Scraping --function-name MySBCloudScraping --trigger-name DownloadTrigger --type oss --config ./scrapingTrigger.yml --source-arn "acs:oss:ap-northeast-1:アカウントid:my-download-bucket000" --role "acs:ram::アカウントid:role/execute-fc"

おわりに

これで１日に１回、指定の時間に情報を取得することができるようになりました！ Function Computeはこのように時刻をトリガーとした自動処理にも利用でき、OSなど低レイヤーの運用が不要なため、徹底的にコードに集中できます。 皆さんも是非、Function Computeを使って自動化に取り組んでみてください。