課題

シェアモビリティのステーションは、日々増加しておりキャッチアップするのが大変。
気が付かないうちに、ステーションが新しくできていたといったことがある。

やりたいこと

シェアモビリティデータの定時取得と新設ステーションを特定。そしてXに自動ポストする。

こちらのXのアカウントでツイートする予定です。

インフォメーション

XはBotの一掃を掲げていますが、きちんと申告すれば大丈夫なようで、このアカウントも「自動アカウント」である申告を行いました。

環境

AWS S3, Lambda

言語

Python 3.9

使用するデータ

データについてはこちら↓

GBFSを公開した際の背景や想いなどはこちら↓

各事業者のGBFSのURLはこちら

Stepを整理

シェアモビリティデータの定時取得
新設ステーションの特定
Xに自動ポスト

それではやっていきましょう。

シェアモビリティデータの定時取得

まずは、一日一回、ステーションのデータを取得し、蓄積していきます。

条件

毎朝6時をトリガーにLambdaのプログラムを実行
取得したデータをS3のフォルダに格納

getgbfs.py

import json
import http.client
from urllib.parse import urlparse
import boto3
import datetime

def lambda_handler(event, context):

    # URLからデータを取得
    url = "https://api-public.odpt.org/api/v4/gbfs/hellocycling/station_information.json"
    parsed_url = urlparse(url)
    conn = http.client.HTTPSConnection(parsed_url.netloc)
    conn.request("GET", parsed_url.path)
    response = conn.getresponse()
    data = response.read()
    conn.close()

    # JSONデータの読み込み
    data_json = json.loads(data)

    # 'data' キーの中身を展開
    data_items = data_json['data']['stations']

    # CSV形式の文字列に変換
    csv_str = "station_id, name, lat, lon\n"
    for item in data_items:
        csv_str += f"{item['station_id']}, {item['name']}, {item['lat']}, {item['lon']}\n"

    # S3にアップロード
    bucket_name = 'gbfsdata'
    t_delta = datetime.timedelta(hours=9)
    JST = datetime.timezone(t_delta, 'JST')
    now = datetime.datetime.now(JST)
    file_name = f"hello-cycling/station-information/{now.strftime('%Y%m%d_%H%M')}.csv"

    s3 = boto3.resource('s3')
    s3.Bucket(bucket_name).put_object(Key=file_name, Body=csv_str, ContentType='text/csv')

    return {"status": "Success"}

新設ステーションの特定

前日のデータと、当日のデータを突き合わせて、
station の id に差分があるところがステーションの増加 or 減少があったところと考えます。
今回は、新設のステーションをツイートしたいので、「前日までなかったが、当日のデータにはあるステーション」を抽出します。

new_station.py

Xに自動ポスト

特定したステーションの情報をXに投稿します。

注意
Lambda上での自動投稿に少し躓いているので、ローカルでの処理について記載します。

auto_tweet.py

エラー対処

Response
{
"errorMessage": "Unable to import module 'lambda_function': cannot import name 'DEFAULT_CIPHERS' from 'urllib3.util.ssl_' (/opt/python/urllib3/util/ssl_.py)",
"errorType": "Runtime.ImportModuleError",
"requestId": "xxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxxxxx",
"stackTrace": []
}

原因は「urllib3」が、つい2週間前(2023年6月前半)にLatestが、
1系から2系になったことによるものらしい。
https://qiita.com/maechi/items/6d5107bc526784a026b3

参考にしたブログ