第2章：GCPを用いた住宅情報をスクレイピングと、賃貸物件価格予測を自動化

概要

目的

summoのwebサイトの物件から、最新でお得な賃貸物件を抽出したい。第2章

背景

不動産のサイトのみでは、条件に対する家賃が高いのか安いのか感覚でしか判断出来ないと、日々感じていました。そこで、様々な賃貸情報から家賃の相場を予測することで、数値的にその物件の価値を図ることが出来ると思いました。

内容

suumoのWebサイトから、住宅情報をスクレイピングして、DataFrameとしてデータを収集する。そのうえで、よりお得な物件を抽出すること。

今回は記事を3つに分けて、投稿します。
第1章：suumoのWebサイトから住宅情報をスクレイピング
第2章：GCPを用いた住宅情報をスクレイピングと、賃貸物件価格予測を自動化
第3章：Straemlitを用いた賃貸物件のWebアプリAPI作成

上記のような構成で考えています。

第1章ではプロジェクトで住宅情報のまとめサイトであるsuumoの賃貸の情報をjupiterのローカル上で、スクレイピングした。
その上で、取得したデータを用いて、家賃の相場を予測して、その相場と実際の家賃価格の差を求めることで、よりお得な物件を抽出した。

今回の第2章では、以上のことをGCPを用いて、日々の更新される物件情報を含めて自動化し、毎日行えるように設定する。

作成した分析基盤について

① Cloud Schedulerが毎日am09:00にPub/Subへ通知
↓
② 通知を受け取ったPub/SubがCloud Functionsを起動
↓
③ Cloud Functionsはsuumoから指定した駅周辺の「本日の新着」物件の情報をスクレイピング
↓
④ スクレイピングされたデータはCloud Functions上で最低限の前処理 (不要な文字列の削除など)を行い、BigQueryのデーブルへ挿入。
↓
⑤ ④で得たデータをCloud Storageにcsvファイルを保存して、それを用いて新たにCloud Functionsで関数を作成し家賃の相場を予測

以上のような流れで、データを取得、分析、ファイル保存を行う。

実際の操作

実際のGCPにおける具体的な操作は下記の記事を参考にしています。
具体的な内容もほぼ同じです。異なる部分は、「もう一つCloud Functionsで関数を作って、家賃の相場との比較の特徴量を作っている点です。」

以上の記事を参考にBigQuery,Cloud Storage,Cloud Functionsを設定する。
今回は、Cloud Functionsで作成した関数のコードを記述します。
以下は、「function-1」という名前の関数の「main.py」のコードです。
エントリポイント：main

!クリック! 関数「function-1」の「main.py」のソースコード

import requests
from bs4 import BeautifulSoup
import pandas as pd
import time
import re
import datetime
import logging
from google.cloud import storage as gcs # gcsへデータを送るのに必要
from google.cloud import bigquery as gbq # BigQueryのテーブルにデータを挿入するのに必要
# スクレイピングの開始点のURL
url = "https://suumo.jp/jj/chintai/ichiran/FR301FC005/?ar=030&bs=040&ra=013&rn=0045&ek=004506820&cb=0.0&ct=9999999&mb=0&mt=9999999&et=9999999&cn=9999999&shkr1=03&shkr2=03&shkr3=03&shkr4=03&sngz=&po1=09&po2=99&pc=100"

# スクレイピングを実行する関数
def suumo_scraping():
    name = []
    station = []
    price = []
    sikikinreikin = []
    room = []
    age = []
    address = []
    count_new_list = [] 

    urls=requests.get(url)
    # 連続してアクセスするのを防ぐために3秒待つ
    time.sleep(3)
    urls.encoding = urls.apparent_encoding 
    soup=BeautifulSoup()
    soup=BeautifulSoup(urls.content,"html.parser")
    get_url = soup.find("ol",class_="pagination-parts")
    # 物件のページ数を取得
    num_list =[]
    for i in get_url.find_all("li"):
        num_list.append(i.text)
    num = int(num_list[10]) + 1

    # ページ数分だけスクレイピングを実行
    for p in range(1,num):
        page=str(p)
        url_2="https://suumo.jp/jj/chintai/ichiran/FR301FC005/?ar=030&bs=040&ra=013&rn=0045&ek=004506820&cb=0.0&ct=9999999&mb=0&mt=9999999&et=9999999&cn=9999999&shkr1=03&shkr2=03&shkr3=03&shkr4=03&sngz=&po1=09&po2=99&pc=100" + "&page=" + page
        urls=requests.get(url_2)
        # 連続してアクセスするのを防ぐために3秒待つ
        time.sleep(3)
        soup=BeautifulSoup(urls.content,"html.parser")
        # 「本日の新着」「新着」のタグを検索
        house_info = soup.find_all("span",class_="ellipse_pct ellipse_pct--red")
        # 「本日の新着」「新着」のタグの中で「本日の新着」の数をカウント
        count_new = 0
        for i in house_info:
            if i.text == "本日の新着":
                count_new += 1
        count_new_list.append(count_new)
        # ページ内の「本日の新着」の数が0でないならスクレイピングを実行
        if count_new != 0:
            house_name = soup.find_all("a",class_="js-cassetLinkHref")
            station_name = soup.find_all("div",style="font-weight:bold")
            table_data = pd.read_html(urls.content)
            for h in house_name:
                name.append(h.text)
            for s in station_name:
                station.append(s.text)
        
            for i in range(0,count_new):
                table = table_data[i]
                price.append(table.iloc[0,0])
                room.append(table.iloc[0,2])
                age.append(table.iloc[0,3])
                address.append(table.iloc[0,4])
                sikikinreikin.append(table.iloc[0,1])
        # ページ内の「本日の新着」が0ならスクレイピングをやめる
        else:
            break
    
    return name, station, price, sikikinreikin, room, age, address, count_new_list

# 前処理を実行する関数 スクレイピングの際に余計な空白等がついてくるのでここで除去する
# このコードだけだとイメージしにくい。手元の環境でデータを見ながらのほうがイメージしやすい
def preprocessing(name, station, price, sikikinreikin, room, age, address, count_new_list):
    station_line_list = []
    time_move_list = []

    for i in station:
        station_line_list.append(i.split(" ")[0])
        time_move_list.append(i.split(" ")[1])

    station_list = []
    line_list = []

    for i in station_line_list:
        station_list.append(i.split("/")[1])
        line_list.append(i.split("/")[0])

    move_list =[]
    time_list = []

    for i in time_move_list:
        if "歩" in i:
            move_list.append("徒歩")
            time_list.append(re.sub(r"\D", "", i))
        else:
            move_list.append("バス")
            time_list.append(re.sub(r"\D", "", i))

    df=pd.DataFrame()        

    df["station"] = pd.Series(station_list)
    df["line"] = pd.Series(line_list)
    df["move"] = pd.Series(move_list)
    df["time_to_station_min"] = pd.Series(time_list).astype(float)


    price_list = []
    admin_list = []

    for i in price:
        price_list.append(i.split(" ")[0].replace("万円",""))
        admin_list.append(i.split(" ")[3].replace("円",""))

    admin_fee_list = []

    for i in admin_list:
        if i == "-":
            admin_fee_list.append("0")
        else:
            admin_fee_list.append(i)

    df["price_10k"] = pd.Series(price_list).astype(float)
    df["admin_fee"] = pd.Series(admin_fee_list).astype(float)

    sikikin_pre = []
    reikin_pre = []
    deposit_pre = []
    sikibiki_pre = []

    for i in sikikinreikin:
        sikikin_pre.append(i.split(" ")[0])
        reikin_pre.append(i.split(" ")[2])
        deposit_pre.append(i.split(" ")[4])
        sikibiki_pre.append(i.split(" ")[6])

    sikikin_list = []
    reikin_list = []
    deposit_list = []
    sikibiki_list = []

    for i in sikikin_pre:
        a = i.split("敷")[1]
        if "万円" in a:
            sikikin_list.append(a.replace("万円",""))
        else:
            sikikin_list.append(0)

    for i in reikin_pre:
        a = i.split('礼')[1]
        if "万円" in a:
            reikin_list.append(a.replace("万円",""))
        else:
            reikin_list.append(0)

    for i in deposit_pre:
        a = i.split('\xa0')[1]
        if "万円" in a:
            deposit_list.append(a.replace("万円",""))
        else:
            deposit_list.append(0)

    for i in sikibiki_pre:
        a = i.split("\xa0")[1]
        if "万円" in a:
            sikibiki_list.append(a.replace("万円",""))
        elif "-" in a:
            sikibiki_list.append(0)
        else:
            sikibiki_list.append("実費")

    df["sikikin_10k"] = pd.Series(sikikin_list).astype(float)
    df["reikin_10k"] = pd.Series(reikin_list).astype(float)
    df["deposit_10k"] = pd.Series(deposit_list).astype(float)
    df["sikibiki_10k"] = pd.Series(sikibiki_list)

    room_list = []
    area_pre = []
    direction_pre = []

    for i in room:
        room_list.append(i.split(" ")[0])
        area_pre.append(i.split(" ")[2])
        direction_pre.append(i.split(" ")[4])

    area_list = []

    for i in area_pre:
        area_list.append(i.replace("m2",""))

    df["room"] = pd.Series(room_list)
    df["area_m2"] = pd.Series(area_list)
    df["direction"] = pd.Series(direction_pre)

    type_list = []
    age_pre = []

    for i in age:
        type_list.append(i.split(" ")[0])
        age_pre.append(i.split(" ")[2])

    age_list = []

    for i in age_pre:
        if i == "新築":
            age_list.append(0)
        else:
            age_list.append(re.sub(r"\D", "", i))

    df["type"] = pd.Series(type_list)
    df["age_year"] = pd.Series(age_list).astype(float)

    df["scraping_date"] = datetime.date.today()
    df["name"] = pd.Series(name)
    df["address"] = pd.Series(address)

    id_range = 0
    for i in count_new_list:
        id_range = i + id_range


    id_list = []
    to_day = str(datetime.date.today())
    to_day = to_day.replace("-","")
    for i in range(0,id_range):
        if i < 10:
            i = str(i)
            id_list.append(to_day + "000"  + i)            
        elif i >= 10 and i < 100:
            i = str(i)
            id_list.append(to_day + "00" + i)
        elif i >= 100 and i < 1000:
            i = str(i)
            id_list.append(to_day + "0" + i)            
        else:
            i = str(i)
            id_list.append(to_day + i) 

    df["scraping_id"] = pd.Series(id_list)

    df = df.reindex(columns=["scraping_id","scraping_date","name","price_10k","age_year","admin_fee","sikikin_10k","reikin_10k","deposit_10k","sikibiki_10k","line","station","move","time_to_station_min","room","area_m2","direction","type","address"])
    # 最後のページのスクレイピングでは「本日の新着」以外のデータも混ざっている
    # pd.read_html(urls.content)で取得した賃貸価格など
    # それ以外で取得した物件名などは「本日の新着」分のデータしか取得していないのでNULLの部分が出てくる
    # NULLが入っているデータは余分なデータなので削除する
    df.dropna(inplace=True)
    
    return df

# データをGCSへcsvファイルで保存する関数
def send_storage(df):
    #Bigqueryで作成したデータセットの「プロジェクトID」の部分
    project_id = "rapid-stage-380007"  

    #Cloud Storageで作ったバケット名
    bucket_name = "suumo_bucket01" 

    #「上記のバケットの中のファイル名/自分で格納したいファイル名(今回はtestとした)」
    gcs_path_1 = "suumo_data/test" # バケットのファイルまでのパス
    # ファイル名でスクレイピングした日付がわかるようにする
    to_day = str(datetime.date.today())
    to_day = to_day.replace("-","")
    
    client = gcs.Client(project_id)
    bucket = client.get_bucket(bucket_name)
    blob_gcs_1 = bucket.blob(gcs_path_1 + "_" + to_day + ".csv")

    blob_gcs_1.upload_from_string(data=df.to_csv(index=False))

# BigQueryのテーブルにデータをインサートする関数
def bigquery_insert(df):
    client = gbq.Client()
    
#Bigqueryのプロジェクト内のテーブルを開いて、「詳細」を開く。そこのテーブル情報のテーブルIDの部分
    table = client.get_table("rapid-stage-380007.suumo01.suumo01_table")
    client.insert_rows(table, df.values.tolist())
#project.dataset.suumo01_table
#suumo01_table
# Cloud Functionで実行する関数
def main(event, context):
    name, station, price, sikikinreikin, room, age, address, count_new_list = suumo_scraping()
    df = preprocessing(name, station, price, sikikinreikin, room, age, address, count_new_list)
    send_storage(df)
    bigquery_insert(df)

上記のコードにおいて、私が沼ったポイントは「# データをGCSへcsvファイルで保存する関数」以下のコードの部分です。
Cloud Functions、Bigquery、Cloud Storageのそれぞれのパケット名、IDなど対応を把握する必要があります。
それぞれどこに対応しているのか、コード内のテキストに詳しく載せておくので、参考にしてください。

ちなみに、「requirements.txt」は下記の通りです。

# Function dependencies, for example:
# package>=version

beautifulsoup4>=4.10.0
requests>=2.27.1
pandas>=1.4.1
google-cloud-storage>=2.1.0
lxml>=4.8.0
google-cloud-bigquery>=2.20.0

以上で、「デプロイ」、「関数のテスト」を行い、うまくいったら掲載した記事を参考に
Cloud Schedulerの設定を行ってください。
︙ をクリックし、ジョブを強制実行するを押すと先ほどunix-cron形式で設定した時間に関わらずジョブを実行できます。
正常に動作すれば、GCSにcsvファイルが保存され、BigQueryのテーブルにデータが挿入されているはずです。

データ分析の自動化

スクレイピングして作ったファイルを用いて、前回の第1章で行った家賃の相場を予測するデータ分析基盤を作る。

【ざっくりした手順は第1章と同じ】

① スクレイピングして作成したDFから、家賃を予測するモデルを作り、「家賃の相場」として新たに特徴量を作成。
↓
② 前処理はobject型のものをラベルエンコーディングをして、最低限の前処理を行った。
↓
③ ロジスティック回帰を用いて、モデル構築。
↓
④ ( 「実際の家賃」-「家賃の相場」)をした求めることで相場よりも安い家賃、お得と思われる物件を抽出。

【具体的な手順】

Cloud Functionsに新たに「mp_function-1」という名の関数を作成。
下記に関数内のコードを書きます。

エントリポイント：mp_main

!クリック! 関数「mp_function-1」の「main.py」のソースコード

import base64

from google.cloud import storage
import pandas as pd
import io
from sklearn.linear_model import LinearRegression
import datetime
from sklearn.preprocessing import LabelEncoder


def read_csv_from_gcs(bucket_name, file_path):
    #Google Cloud StorageからCSVファイルを読み込み、PandasのDataFrameを返す
    client = storage.Client()
    bucket = client.get_bucket(bucket_name)
    blob = bucket.blob(file_path)
    data = blob.download_as_string()
    df = pd.read_csv(io.BytesIO(data),dtype=str)
    return df

def write_csv_to_gcs(df, bucket_name, file_path):
    #PandasのDataFrameをGoogle Cloud Storage上のCSVファイルに書き込み
    client = storage.Client()
    bucket = client.get_bucket(bucket_name)
    blob = bucket.blob(file_path)
    blob.upload_from_string(df.to_csv(index=False), 'text/csv')

def mp_main(event, context):
    """Cloud Pub/Subトピック上のメッセージによってトリガー。
引数:
- event (dict): イベントペイロード。
- context (google.cloud.functions.Context): イベントのメタデータ。
   """
    
    df1 = read_csv_from_gcs('suumo_bucket01', 'suumo_data/test_20230411.csv')

    #カテゴリ変数のカラムを指定して、まとめて処理
    def encode_categorical(df,cols):
        for col in cols:
            le = LabelEncoder()
            df[col] = pd.Series(le.fit_transform(df[col]))
    
        return df
    df1 = encode_categorical(df1,cols=["scraping_date","name","line","station","move","room","direction","type","address"])

    X = df1.drop('price_10k', axis=1)
    y = df1['price_10k']

    model = LinearRegression()
    model.fit(X, y)

    df2 = df1.copy()
    df2['market_rent'] = model.predict(X)
    df2['price_10k'] = df2['price_10k'].astype(float)
    df2['market_rent'] = df2['market_rent'].astype(float)

    df2['rent_difference'] = df2['price_10k'] - df2['market_rent']

    df2_sorted = df2.sort_values('rent_difference', ascending=True)
    write_csv_to_gcs(df2_sorted, 'suumo_bucket01', 'suumo_mp_data/mp_file01.csv')

# データをGCSへcsvファイルで保存する関数
def send_storage(df2_sorted):
    project_id = "rapid-stage-380007" 
    bucket_name = "suumo_bucket01" 
    gcs_path_2 = "suumo_mp_data/mp_file01" # バケットのファイルまでのパス
    # ファイル名でスクレイピングした日付がわかるようにする
    to_day = str(datetime.date.today())
    to_day = to_day.replace("-","")
    
    client = gcs.Client(project_id)
    bucket = client.get_bucket(bucket_name)
    blob_gcs_2 = bucket.blob(gcs_path_2 + "_" + to_day + ".csv")

    blob_gcs_2.upload_from_string(data=df2_sorted.to_csv(index=False))

下記は、requirements.txt

# Function dependencies, for example:
# package>=version

pandas>=1.4.1
google-cloud-storage>=2.1.0

scikit-learn>=1.0.2
lightGBM>=3.3.3

以上で、1つ目の関数と同じように、デプロイ、テストをしてみてください。

押上地区など場所をある程度絞ったら、データ量もそんなに多くないので、通して実装を行うときは地区を絞ってみても良いかもしれません。

まとめ

今回はChatGPTにかなり助けられました。
GCP初心者の人は、「エラーなどまずどこを確認すれば良いのかわからない。」という状況になるかと思います。特に、参考記事のCloud Functionsの関数のデプロイ、テストでつっかかったら、「ログ」を確認して赤くなっている警告、エラーを見てください。見て自分でわからなかったら、エラー情報を全てコピーして、ChatGPTに張り付けて聞いてみてください。ヒントをくれるはずです。

初めはGCFでログの情報の多さにエラー文を読むことに時間がかかっていましたが、丁寧に読んで適切に対処することで着実に作業を進めることが出来ました。
今回はGCFのコードが複雑になりすぎないように、スクレイピングの関数と、前処理、データ分析の関数を分けて作成したが、今後はまとめて作った方が早そうだし、まとまりが良いと思いました。
実際に毎朝情報が自動で更新されていて、便利さを実感しました。

次回は第3章「Straemlitを用いた賃貸物件のWebアプリAPI作成」です。お得な物件を抽出できたので、条件で絞る機能やお得な物件を可視化できるAPIを作成します。