トップ > Python > 楽天モバイル（奈良県）の包括免許の差分をツイート

楽天モバイル（奈良県）の包括免許の差分をツイート

Python 楽天モバイル pandas

インストール

pip install pandas
pip install requests
pip install tweepy

プログラム

# -*- coding: utf-8 -*-

import csv
import datetime
import pathlib
import urllib.parse

import pandas as pd
import requests
import tweepy

# Twitter
consumer_key = ""
consumer_secret = ""
access_token = ""
access_token_secret = ""

# API

d = {
    # 1:免許情報検索  2: 登録情報検索
    "ST": 1,
    # 詳細情報付加 0:なし 1:あり
    "DA": 1,
    # スタートカウント
    "SC": 1,
    # 取得件数
    "DC": 1,
    # 出力形式 1:CSV 2:JSON 3:XML
    "OF": 1,
    # 無線局の種別
    "OW": "FB_H",
    # 所轄総合通信局
    "IT": "E",
    # 免許人名称/登録人名称
    "NA": "楽天モバイル",
}

parm = urllib.parse.urlencode(d, encoding="shift-jis")

r = requests.get("https://www.tele.soumu.go.jp/musen/list", parm)
r.raise_for_status()

# データラングリング

cr = csv.reader(r.text.splitlines(), delimiter=",")
data = list(cr)

df0 = pd.DataFrame(data[1:]).dropna(how="all")

update = datetime.datetime.strptime(data[0][0], "%Y-%m-%d").date()

# レコード数確認
len(df0) == int(data[0][1])

df1 = df0[25].str.strip().str.split(r"\\n", 2, expand=True)

se = df1.loc[df1[0].str.contains("携帯電話（その他基地局等"), 2]

df2 = (
    se.str.strip()
    .str.replace(r"\\n", "")
    .str.extractall("(.+?)\(([0-9,]+?)\)")
    .rename(columns={0: "市区町村名", 1: "開設局数"})
    .reset_index(drop=True)
)

df2["市区町村名"] = df2["市区町村名"].str.strip()

df2["開設局数"] = df2["開設局数"].str.strip().str.replace(",", "").astype(int)

flag = df2["市区町村名"].str.endswith(("都", "道", "府", "県"))

df2["都道府県名"] = df2["市区町村名"].where(flag).fillna(method="ffill")

df2

# 都道府県

df_prefs = df2[flag].reset_index(drop=True)
df_prefs

df_prefs.to_csv("prefs.csv", encoding="utf_8_sig")

# 市区町村

df_cities = (
    df2[~flag].reset_index(drop=True).reindex(columns=["都道府県名", "市区町村名", "開設局数"])
)
df_cities

df_cities.to_csv("cities.csv", encoding="utf_8_sig")

# 奈良県

df_nara = df_cities[df_cities["都道府県名"] == "奈良県"].copy()

df_nara["市区町村名"] = df_nara["市区町村名"].str.replace(
    "^(添上郡|山辺郡|生駒郡|磯城郡|宇陀郡|高市郡|北葛城郡|吉野郡)", "", regex=True
)

df_nara.set_index("市区町村名", inplace=True)

df_nara.drop("都道府県名", axis=1, inplace=True)

idx = (
    "奈良市",
    "大和高田市",
    "大和郡山市",
    "天理市",
    "橿原市",
    "桜井市",
    "五條市",
    "御所市",
    "生駒市",
    "香芝市",
    "葛城市",
    "宇陀市",
    "山添村",
    "平群町",
    "三郷町",
    "斑鳩町",
    "安堵町",
    "川西町",
    "三宅町",
    "田原本町",
    "曽爾村",
    "御杖村",
    "高取町",
    "明日香村",
    "上牧町",
    "王寺町",
    "広陵町",
    "河合町",
    "吉野町",
    "大淀町",
    "下市町",
    "黒滝村",
    "天川村",
    "野迫川村",
    "十津川村",
    "下北山村",
    "上北山村",
    "川上村",
    "東吉野村",
)

df_nara = df_nara.reindex(idx, fill_value=0)

df_nara["開設局数"].sum()

df_nara.loc["計"] = df_nara["開設局数"].sum()


# 更新日
print(update.isoformat())

dt_3d = datetime.date.today() - datetime.timedelta(days=3)

if update > dt_3d:
    print("3日以内に更新されました")

# 差分

p = pathlib.Path("latest.csv")

df_nara.rename(columns={"開設局数": update.isoformat()}, inplace=True)

if p.exists():

    print("latest.csvを更新します")

    df_last = pd.read_csv(p, index_col=0)
    dt_last = datetime.date.fromisoformat(df_last.columns[0])

    if update > dt_last:

        print("データがアップデートされました")

        df_last.to_csv(f"{df_last.columns[0]}.csv", encoding="utf_8_sig")

        df_diff = df_last.join(df_nara)
        df_diff["差分"] = df_diff.diff(axis=1).iloc[:, -1].astype(int)

        df_diff.to_csv("diff.csv", encoding="utf_8_sig")

        # 差分表示

        df3 = df_diff[df_diff["差分"] != 0]

        mes = []

        mes.append("総務省免状更新 #楽天モバイル #奈良県")
        mes.append("https://www.tele.soumu.go.jp/musen/SearchServlet?pageID=4&IT=E&DFCD=0001745556&DD=2&styleNumber=00")
        mes.append("https://docs.google.com/spreadsheets/d/1HojKLdBi3DP_qP6_y0gHEYCztmwazzWDc1iDng-nmtU/edit?usp=sharing\n")

        for i, row in df3.iterrows():
            mes.append(f"{i} {row[1]}({row[2]:+})")

        twit = "\n".join(mes)

        print(twit)

        # ツイート

        auth = tweepy.OAuthHandler(consumer_key, consumer_secret)
        auth.set_access_token(access_token, access_token_secret)

        api = tweepy.API(auth)
        api.update_status(twit)

else:

    print("latest.csvを作成しました")

df_nara.to_csv("latest.csv", encoding="utf_8_sig")