SlackとLambdaを使った情報収集

ゴールデンウィークから作っていたものが一段落ついたので、そのときに調べたことのまとめです。

# 概要

Twitter, RSS, Pinterestから流れてくる情報をSlackに集約し、Lambdaで加工してS3に保存します。その後、ログをSlackに流します。LambdaではPythonを使いました。

![全体図](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/223255/e8dc7fd8-438f-0f59-3000-925b2d9ec74c.png)

# 詳細

## Zapierで情報をSlackに集約する
![zapier](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/223255/94ac84a6-2203-0a44-8fcb-b81467c26347.png)

[Zapier](https://zapier.com)というサービスを使いました。Zapというものを設定してWebサービスを連携させることができます。[連携できるサービス](https://zapier.com/apps)は、たとえば、TwitterやPinterest, Instagram, Tumblrなどのソーシャルメディア、Trello, Asana, Github, Googleの各種サービスなど、たくさんあって捗ります。

Freeプランでは5つのZapを設定でき、1ヶ月に100タスク実行することができます。今回の利用目的だと1ヶ月100タスクはちょっと少ないのでStarterにしました。

### Zapの作り方

Make a Zap!から、たとえばTwitterを選択して、
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/223255/c17989d0-3328-d65b-88ec-c7457e01a7a6.png)


自分のツイートをトリガーに設定。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/223255/c8519797-9c68-fe12-9cb9-5459cda44e30.png)

Twitterアカウントを登録して、
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/223255/2d346cb7-b9cf-0850-b883-ba3a118b4431.png)

テストに使うツイートを設定しておきます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/223255/5a47ebfa-3684-7509-96aa-3a57d8922587.png)

2つめのステップにActionを選択して、Slackを登録。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/223255/508ddfe2-ec96-7305-f669-0656e6949295.png)

チャンネルにメッセージを送信するように設定
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/223255/45a13727-9b54-ba84-0d18-006bd1ae43f3.png)

Slackアカウントを登録して、
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/223255/4148face-9621-cad6-4ffd-e2d5f2cf09ca.png)

Slackにどんなメッセージを送るのかを決めます。たとえば#generalにツイートのurlを送るように設定。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/223255/dea1ae69-bffa-8f65-05a9-c4034f1bcb09.png)

最後にテストして登録したSlackワークスペースの#generalにテストツイートのurlが送信されれば完了です。


## SlackからLambdaを起動する
![](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/223255/36807953-5eee-57d9-bfd0-33ee41aa92df.png)

調べてみるととOutgoing WebHooksを使っている記事が多いようなんですが、API Gatewayでマッピングテンプレートを設定する必要があり、serverlessからのデプロイがうまくいかなかったのでやめました。代わりにSlack Appを使いました。

- 参考：[Outgoing WebHooksの代わりをSlack APPで実装する - Qiita](https://qiita.com/NishimuraRisuto/items/342256f6ed6cb504059a)

API Gatewayから直接Step Functionsを起動しすることもできますが、challengeを返すのと、verification tokenの確認だけ前段のLambdaで行っています。

```python:URL_verification部分抜粋
def check_challenge(body):
    if body.get('type') == 'url_verification':
        return {
            'statusCode': 200,
            'headers': {},
            'body': json.dumps({'challenge': body.get('challenge')}),
            'isBase64Encoded': False,
        }
    return None
```

```python:verificatoin_token確認部分抜粋
def check_token(body):
    if body.get('token') != SLACK_APP_VERIFICATION_TOKEN:
        raise UnauthorizedError
```

Lambdaプロキシ統合を使っている場合にLambdaのレスポンスの形式は以下。

- 参考：[API Gateway で統合レスポンスを設定する - Amazon API Gateway](https://docs.aws.amazon.com/ja_jp/apigateway/latest/developerguide/api-gateway-integration-settings-integration-response.html)

## LambdaからStep Functionsの起動する
![](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/223255/3acc8c25-cf1d-cc6b-1835-0a666957e700.png)

- 参考：[AWS Step Functionsを定期的に実行するLambda関数をServerless Frameworkでデプロイする #reinvent ｜ DevelopersIO](https://dev.classmethod.jp/cloud/aws/invoking-aws-step-functions-with-lambda/)

```python:stepfunctionsの実行
stepfunctions = boto3.client('stepfunctions')

def execute_stepfunction(params):
    stepfunctions.start_execution(
        **{
            'input': json.dumps(params),
            'stateMachineArn': STEPFUNCTION_ARN,
        }
    )
```

## AWSで使ったサービス

### Lambda
![](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/223255/0268a3ab-46a1-263e-78e3-b632c0456c77.png)

共通のコードはLayerに作成します。

- 参考：[AWS LambdaにLayerなるものが追加されたのでPythonで簡単に使ってみる #reinvent ｜ DevelopersIO](https://dev.classmethod.jp/server-side/python/lambda-layer-simply-use-by-python/)

`requirements.py`というのをLayerに作っておいて、各ディレクトリのsite-packagesにパスが通るようにしました。

```python:requirements.py
import inspect
import os
import sys

sys.path.append(os.path.join(os.path.dirname(__file__), 'site-packages'))

stack = inspect.stack()[1:]
ms = [inspect.getmodule(_[0]) for _ in stack]
m = next((m for m in ms if m), None)
if m:
    path = os.path.dirname(m.__file__)
    if path not in sys.path:
        sys.path.append(path)
        sys.path.append(os.path.join(path, 'site-packages'))
```

- 参考：[インポート元モジュールのパスを取得する - Qiita](https://qiita.com/yoichi22/items/3d6606912659d9fd7d52)


### Step Functions
![](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/223255/952b449a-36b9-9a58-3095-8ca90f320b54.png)

Parallelで下のようなStateを作ると、出力が２つ出てくる。

```json:ParallelState
"ParallelState": {
  "Type": "Parallel",
  "Next": "NextState",
  "Branches": [
    {
      "StartAt": "Branch1",
      "States": {
        "Branch1": {
          "Type": "Task",
          "Resource": "ARN1",
          "InputPath": "$",
          "ResultPath": "$.branch1Result",
          "OutputPath": "$",
          "End": true
        }
      }
    },
    {
      "StartAt": "Branch2",
      "States": {
        "Branch2": {
          "Type": "Task",
          "Resource": "ARN2",
          "InputPath": "$",
          "ResultPath": "$.branch2Result",
          "OutputPath": "$",
          "End": true
        }
      }
    }
  ]
}
```

```json:ParallelStateの出力
[
  {
    "input": "hoge",
    "branch1Result": {
      "result": "branch1"
    }
  },
  {
    "input": "hoge",
    "branch2Result": {
      "result": "branch2"
    }
  }
]
```

この結果を１つにマージしたかったので、Parallelの後に入力をマージするだけのLambdaを１つ追加しました。もっとよい方法があるのかもしれない。

```python:merge_results.py
def merge(params):
    results = {}
    for p in params:
        results.update(p)
    return result
```

- 参考：[Step Functions の入出力処理 - AWS Step Functions](https://docs.aws.amazon.com/ja_jp/step-functions/latest/dg/concepts-input-output-filtering.html)

### Key Management Service (KMS)
![](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/223255/6d2c82f1-a046-b525-9681-e26a612d108f.png)

Slackのverification tokenやincoming WebHookのURLなどはKMSで暗号化してLambdaの環境変数に設定しました。

```python:get_env
def get_env(key, kms_client=None):
    env = os.environ[key]
    if kms_client:
        env = kms_client.decrypt(
            CiphertextBlob=b64decode(os.environ[key])
        )['Plaintext'].decode()
    return env

kms = boto3.client('kms')
VERIFICATION_TOKEN = get_env('VERIFICATION_TOKEN', kms)
```

### DynamoDB
![](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/223255/1deba1a6-8114-29b2-1b0d-ccecace55b1c.png)

処理したサイト等はキャッシュしておき、２回目は処理しないようにしたかったので、キャッシュをDynamoDBに保存します。

今回は`put_item`と`get_item`しか使いませんでした。

```python:dynamodb_get_and_put_item
dynamodb = boto3.resource('dynamodb')
DYNAMODB_TABLE_NAME = get_env('DYNAMODB_TABLE_NAME')
dynamodb_table = dynamodb.Table(DYNAMODB_TABLE_NAME)

def put_item(url, result):
    dynamodb_table.put_item(
        Item={
            "id": url,
            "result": result,
        }
    )

def get_item(url):
    r = dynamodb_table.get_item(
        Key={"id": url}
    )
    return r.get('Item')
```


### S3
![](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/223255/f147e465-2681-e6b3-6e76-f3ffed8ec35f.png)

`request`でダウンロードした画像をs3に保存

```python
import requests

def download_image_and_save_to_s3(img_url, key):
    r = request.get(img_url)
    bucket.put_object(
        Key=key,
        Body=r.content
    )
```

s3から画像を取得

```python
def get_image_from_s3(bucket, key):
    name = os.path.basename(key)
    obj = bucket.Object(key).get()
    buffer = io.BytesIO(obj.get('Body').read())
    return name, buffer
```


pillowの画像をpngでをs3に保存

```python
import io

def save_image_to_s3_as_png(image, bucket, key):
    image_buffer = io.BytesIO()
    image.save(image_buffer, 'PNG')
    image_buffer.seek(0)
    return bucket.put_object(
        Key=key,
        Body=image_buffer.read()
    )
```


### Rekognition
![](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/223255/49595c41-b999-3dc5-58fd-9a0d64e25152.png)

顔の検出

```python
rekognition_client = boto3.client('rekognition')

def detect_faces(bucket_name, key):
    return rekognition_client.detect_faces(
        Image={
            'S3Object': {
                'Bucket': bucket_name,
                'Name': key
            }
        },
        Attributes=['ALL']
    )
```

- 参考：[イメージ内の顔の検出 - Amazon Rekognition](https://docs.aws.amazon.com/ja_jp/rekognition/latest/dg/faces-detect-images.html)

顔の比較

```python
rekognition_client = boto3.client('rekognition')

def compare_faces(bucket_name, key):
    return rekognition_client.compare_faces(
        SourceImage={
            'S3Object': {
                'Bucket': S3_BUCKET_NAME,
                'Name': SOURCE_IMAGE_NAME,
            }
        },
        TargetImage={
            'S3Object': {
                'Bucket': bucket_name,
                'Name': key,
            }
        },
        SimilarityThreshold=80
    )
```

- 参考：[イメージ間の顔の比較 - Amazon Rekognition](https://docs.aws.amazon.com/ja_jp/rekognition/latest/dg/faces-comparefaces.html)


## Slackのチャンネルにメッセージを送信
![](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/223255/4aa8bd51-2813-5946-03d6-17b2516e5d8e.png)

- 参考：[SlackのIncoming Webhooksを使い倒す - Qiita](https://qiita.com/ik-fib/items/b4a502d173a22b3947a0)

## Serverlessでデプロイ
![](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/223255/7a433092-35d4-a3dd-971b-af1e6c7c485d.png)

### やりたいこと

ディレクトリ構成

```
|--functions
|  |--function1
|  |  |--function1.py
|  |--function2
|  |  |--function2.py
|  |  |--requirements.txt
|  |  |--site-packages
|--layers
|  |--requirements
|  |  |--python
|  |  |  |--requirements.py
|  |--lib
|  |  |--python
|  |  |  |--lib.py
|  |  |  |--requirements.txt
|--serverless.yml
```

`functionｓ`の各ディレクトリの下に`requirements.txt`があれば`pip`し、site-packagesを作ってLambdaにアップロードしたい。ライブラリがOSに依存しているとLambda環境では動かないので[lambci/lambda - Docker Hub](https://hub.docker.com/r/lambci/lambda/)上でpipしてsite-packagesを作成する。layerも同様。

[UnitedIncome/serverless-python-requirements](https://github.com/UnitedIncome/serverless-python-requirements)とか[cfchou/serverless-python-individually](https://github.com/cfchou/serverless-python-individually)で出来そうだなと思ってやってみたんですが、うまくいかなかったので自分でプラグインを書きました。

### serverless-python-package

- [yskoht/serverless-python-package](https://github.com/yskoht/serverless-python-package)

`./serverless_plugins/serverless-python-package`に配置して↓のような感じでymlを書くと`sls deploy`時に`functions`と`layers`の各ディレクトリで`requirements.txt`があれば`pip`して`site-packages`を作ります。`dockerize: true`が設定されていればdockerを使います。

```yml:serverless.yml
plugins:
  - serverless-python-package

custom:
  pythonPackage:
    dockerize: false

provider:
  name: aws
  runtime: python3.7
  stage: dev

package:
  individually: true

layers:
  requirements:
    path: layers/requirements
    compatibleRuntimes:
      - python3.7
    retain: false
  lib:
    path: layers/lib
    compatibleRuntimes:
      - python3.7
    retain: false

functions:
  function1:
    name: ${self:provider.stage}-function1
    handler: functions/function1/function1.lambda_handler
    layers:
      - {Ref: RequirementsLambdaLayer}
      - {Ref: LibLambdaLayer}
    package:
      exclude:
        - '**/*'
      include:
        - functions/function1/**

  function2:
    name: ${self:provider.stage}-function2
    handler: functions/function2/function2.lambda_handler
    layers:
      - {Ref: RequirementsLambdaLayer}
      - {Ref: LibLambdaLayer}
    package:
      exclude:
        - '**/*'
      include:
        - functions/detect_faces/**
    pythonPackage:
      dockerize: true
```

### serverless-step-functions

StepFunctionsをserverlessでデプロイするために、`serverless-step-functions`を使いました。

- [horike37/serverless-step-functions: AWS Step Functions plugin for Serverless Framework ⚡️](https://github.com/horike37/serverless-step-functions)