やりたいこと

タイトルのことをしたい。（と思い、できたので、備忘録としてまとめました）
背景として、AWS Lambda関数（WebAPI）からイベントをトリガーして、1日1回ファイルを更新する処理を作りたいと思いました。

try and error の経緯は、以下の感じです。

WebAPIの内部ファイルに書き込もうとする
/tmp/フォルダ内のファイルに書き込もうとする
S3バケットにファイルを保存してそれを参照する　←今ここ

1.では、そもそもWebAPIの内部ファイルはread onlyとなっており、書き込みできませんでした。
2.では、/tmp/フォルダがAWSにキャッシュとして存在しており、時間で（インスタンス消去時に？）消えてしまうため、後からファイルにアクセスできませんでした。

※ WebAPIのイベントの設定方法もまとめておりますので、ご参考ください。
zappa経由でデプロイしたAWSLambda関数（WebAPI）にCloudWatch Eventsを設定する

フロー

スクレイピングしたデータを、csvデータとして/tmp/フォルダに保存する
/tmp/フォルダに保存したcsvデータを、S3バケットにアップロードする
S3バケットのcsvデータを参照する

スクレイピングしたデータを、csvデータとして`/tmp/`フォルダに保存する

以下のような形式のリストデータを保存します。
ここでは、コミックスの新刊情報を取得しています。

datas = [[data1], [data2], [data3], [data4],...]

書き込み処理

import csv

# キャッシュファイルパス
filepath = '/tmp/data.csv'

# ヘッダー
header = ['date', 'title', 'author', 'publisher', 'label']

# 書き込み
with open(filepath, 'w', newline='', encoding='utf-16', errors='ignore') as f:
    writer = csv.writer(f, dialect='excel-tab', quoting=csv.QUOTE_ALL)
    writer.writerow(header)
    writer.writerows(datas) # リスト形式のデータを一行ずつ書き込む

mode：'w'(書き込み・上書き)，'a'(書き込み・追記)，'r'(読み込み)
newline：改行設定
encoding：'utf-16'にすることで文字化けを回避できるようです。(https://akiyoko.hatenablog.jp/entry/2017/12/09/010411)
errors：エラーとなる行の回避設定
dialect：要素同士の区切り方
quoting：文字列化するデータ型の指定

作成したdata.csvは、以下のようになります。

`/temp/`フォルダに保存したcsvデータを、S3バケットにアップロードする

AWSコンソールで、S3バケットを作成します。
バケット名は、release-comicsとしました。

アップロード処理

import boto3

filepath = '/tmp/data.csv'
baket_name = 'release-comics'
savepath = 'data.csv'

s3 = boto3.resource('s3')
s3.meta.client.upload_file(filepath, baket_name, savepath)

バケット直下にcsvデータをアップロードするので、savepathはアップロードファイル名のみ指定しています。

S3バケットのcsvデータを参照する

import csv
import boto3

baket_name = 'release-comics'
savepath = 'data.csv'

s3 = boto3.client('s3')
response = s3.get_object(Bucket=baket_name, Key=savepath)
lines = response['Body'].read().decode('utf-16').splitlines() # 改行で区切ってリスト化
reader = csv.DictReader(lines, dialect='excel-tab', quoting=csv.QUOTE_ALL)
datas = [row for row in reader]

全体像

sever.py

import s3
from flask import Flask, jsonify, request, abort

app = Flask(__name__)
app.config["JSON_AS_ASCII"] = False

# exp：http://127.0.0.1:5000/search
@app.route("/search")
def search():
    results = s3.read()
    return jsonify(results)

if __name__ == "__main__":
    app.run(debug=True, use_reloader=False)

lambda_function.py

import os
import time, datetime
import s3, scraping

# Lambdaトリガーでスケジュール更新する
def update(event, context):
    print('------------------------------')
    today = datetime.datetime.fromtimestamp(time.time())
    t = today.strftime('%Y/%m/%d %H:%M:%S')
    date = t + '：' + 'called lambda_function.update'
    print(date)

    # スクレイピングをしてcsvデータを更新する（AWSのキャッシュフォルダに一時保存）
    filepath = scraping.analysis()
    print('update with scraping：' + filepath)

    # キャッシュフォルダからs3バケットに保存
    uploadpath = s3.upload(filepath)
    print('upload to s3：' + uploadpath)

    # キャッシュの削除
    os.remove(filepath)
    print('remove cache：' + filepath)
    print('------------------------------')

printは、AWSでログを確認する用です。

scraping.py

import os
import re
import csv
import requests
from bs4 import BeautifulSoup

URI = 'https://hogehoge'
DATA_PATH = '/tmp/data.csv'

def analysis():
    # htmlデータの取得
    resp = requests.get(URI)

    # 全角半角変換
    zen = "".join(chr(0xff01 + i) for i in range(94))
    han = "".join(chr(0x21 + i) for i in range(94))
    zen += '　'
    han += ' '
    zen = zen.replace('～', '')
    han = han.replace('~', '')
    zen2han = str.maketrans(zen, han)
    # han2zen = str.maketrans(han, zen)

    pattern = re.compile(r'(\d+)月(\d+)日')

    # 解析
    soup = BeautifulSoup(resp.text, "html.parser")
    soup2 = soup.find_all("table", class_="stTableData01")

    datas = []
    for s2 in soup2:
        soup3 = s2.find("tbody")
        soup4 = soup3.find_all("tr")

        for s4 in soup4:
            line = [content.text.translate(zen2han) for content in s4.contents if content != '\n']

            if pattern.match(line[0]):
                datas.append(line)

    # データの書き込み
    header = ['date', 'title', 'author', 'publisher', 'label']

    with open(DATA_PATH, 'w', newline='', encoding='utf-16', errors='ignore') as f:
        writer = csv.writer(f, dialect='excel-tab', quoting=csv.QUOTE_ALL)
        writer.writerow(header)
        writer.writerows(datas)

説明に載せていないコードがありますが、やっていることは、datasリストを作成して、/tmp/data.csvに保存しています。

s3.py

import os
import csv
import boto3

BAKET_NAME = 'search-comics'
SAVE_FILE = 'data.csv'

def upload(filepath):
    s3 = boto3.resource('s3')
    s3.meta.client.upload_file(filepath, BAKET_NAME, SAVE_FILE)
    return os.path.join(BAKET_NAME, SAVE_FILE)

def read():
    s3 = boto3.client('s3')
    response = s3.get_object(Bucket=BAKET_NAME, Key=SAVE_FILE)
    lines = response['Body'].read().decode('utf-16').splitlines() # 改行で区切ってリスト化
    reader = csv.DictReader(lines, dialect='excel-tab', quoting=csv.QUOTE_ALL)
    return [row for row in reader]

確認

AWSコンソールで、動作の確認をします。

S3バケットで、data.csvファイルの更新日時の確認
CloudWatchで、ログの確認

参考

csv文字化け対策
 S3バケットから取得したCSVファイルを読み込むには？
【Lambda】python で s3 を使ってテキストの読み書きをしてみる

pandasを使う場合
Lambda : s3のcsvファイルをpandasで変換し自動でDynamoへ格納する

【Python】AWS S3バケットにCSVファイルをアップロードしたり、データを読み込んだりする

やりたいこと

フロー

スクレイピングしたデータを、csvデータとして/tmp/フォルダに保存する

/temp/フォルダに保存したcsvデータを、S3バケットにアップロードする

S3バケットのcsvデータを参照する

全体像

確認

参考

スクレイピングしたデータを、csvデータとして`/tmp/`フォルダに保存する

`/temp/`フォルダに保存したcsvデータを、S3バケットにアップロードする