Python スクレイピング 更新情報出力

Question

Pythonを用いて、銘柄の適時開示情報ページ(JSON形式)
(https://webapi.yanoshin.jp/webapi/tdnet/list/8818.json2?limit=100)
で、更新情報があれば、会社名とURLを出力したく、以下のコーディングを行いました。
この場合、一つの銘柄の場合には対応していますが、銘柄が複数の場合、最新の更新情報が無限に出力されてしまいます。
複数の銘柄の場合でも、一度出力した最新の更新情報は出力されないようにするにはどのようにすれば良いでしょうか。

import schedule
import time
import json
import requests

old_file="test.txt"

def job():
    tr=""
    resp = requests.get("https://webapi.yanoshin.jp/webapi/tdnet/list/8818.json2?limit=100")
    data = resp.json()
    result = []
    for item in data.get("items", []):
        tdnet = item.get("Tdnet")
        if tdnet is None:
            break
        published_at = tdnet.get("pubdate")
        # 最新の日時の開示のみ取得する。
        if tr == '':
            tr = published_at

        elif published_at != tr:
            break

        result.append(
            dict(
                name = tdnet.get("company_name"),
                published_at = published_at,
                title = tdnet.get("title"),
                url = tdnet.get("document_url")
            )
        )
    global old_file
    if tr == '':
        print("データを取得できませんでした")
    elif tr == old_file:
        pass
        #print("更新なし。　直近の取得日時：", tr)
        
    else:
        print(f"日時： {tr}")
        for r in result:
            data=f"{r['name']} {r['title']} {r['url']}"
            print(data)
            
    old_file=tr

schedule.every(1).seconds.do(job)
while True:
  schedule.run_pending()
  time.sleep(1)

job()

Comments

[t_obara]

仕様を明確にしましょう。複数の銘柄の場合に一度出力した最新の更新情報は出力されない条件を考える必要があります。どのような条件になるとお考えですか？

[hfjdsfh_48]

一度出力した最新の情報を変数として保存し、2回目以降の処理で出力された情報と、保存した変数を比較して、同じ場合は出力しない、違う場合は出力するという条件になると考えます。
また、URLは"https://webapi.yanoshin.jp/webapi/tdnet/list/{コード番号-コード番号}.json?limit=20"　のほうがいいかもしれません。

[t_obara]

その仕様を満たすようにプログラミングをしたのにできないのか、仕様を満たすようにプログラムを記述できないのかどちらですか？

[hfjdsfh_48]

仕様を満たすようなプログラムを記述できません

[t_obara]

複数の銘柄の場合に、「一度出力した最新の情報を変数として保存」ができないということでしょうか？

[hfjdsfh_48]

はい。複数の銘柄の場合は、変数を複数用意すればうまくいくと思うのですが、銘柄が増えたときにまた変数を増やすのは効率的でないと思いました。そもそも変数を複数用意しなくてもよい書き方があれば教えてください。

Answer 1

APIが返すデータの中に「id」というものがあります。
これは、会社を超えて、1つ1つのデータごとに異なる番号です。

したがってデータを取得したときにidを格納しておき、
次にデータを取得した場合には、取得したデータのidを取得済みのidと比較し、
存在しないidであればそれを(新データとして)表示するようにすればよいでしょう。

考え方：

    # 差分データを格納するリスト
    new_records = []

    for record in records:
        id = record["id"]
        # とってきたデータ（record）のidがデータベースの中にない場合は、
        if id not in database:
            # databaseにそのデータを格納する。
            database[id] =  record
            # new_recordsにも追加する。
            new_records.append(record)

    return new_records

　

コード例：　以下db.py　とmain.pyを同じ場所に置いておく。
db.py

# db.py
# 取得したデータ（レコード）を格納するデータベース擬きの辞書
db = dict()

　
main.py

# main.py

import requests
import schedule
import time


import db

def get_data(company_code, limit:int=10):
    '''
    適時開示データを取得する。
    Args:
        company_code (int): 証券コード
        limit (int): 一度に取得するデータの数の上限
    Returns:
        list(dict) or list(None): 適時開示データのリスト
    '''
    resp = requests.get(f"https://webapi.yanoshin.jp/webapi/tdnet/list/{company_code}.json2?limit={limit}")
    data = resp.json()

    items = data.get("items")
    if not items:
        return []
    # データ形式が古い場合に対応
    if items[0].get("Tdnet") is None:
        if items[0].get("id") is None:
            return []
    else:
        return [item.get("Tdnet") for item in items]
    return data.get("items",[])


def select(record, keys):
    '''
    指定されたレコードから、keysに指定された列データをリストにして返す。
    Args:
        record (dict) : 適時開示データ1行
        keys (iterable(str)): キー名
    Returns:
        list(str): 指定されたレコードの、指定されたキー名に該当する列
    '''
    return [record.get(key) for key in keys]


def update_db(records):
    '''
    Args:
        records(list(dict)) : APIからとってきた適時開示データのリスト
    Returns：
        list(dict) or list(None) : 更新前のデータベースになかったレコードのリスト
    '''
    new_records = []

    for record in records:
        id = record["id"]
        if id not in db.db:
            db.db[id] =  record
            new_records.append(record)
    return new_records

def job(codes):
    # 各々の証券コードごとに・・・
    for code in codes:
        # API経由で適時開示データ（レコード）をとってきて・・
        records = get_data(code)
        # データベース内の取得済みのデータと比較し、新しいレコードのみ抽出する。
        new_records = update_db(records)
        # 新しいレコードの内容を出力する。
        for record in new_records:
            pubdate, title, name, url = select(record, ['pubdate', 'title','company_name','document_url'])
            print(f"{pubdate}: {name}  {title} {url}")

# 取得する銘柄の証券コードを指定
target_codes = [7974, 8818]

# サーバー負荷を考慮して10分ごとの取得としています。
# 適時開示はツイッターみたいに頻繁に更新されるものではありませんから、1秒毎の更新間隔は短すぎると思います。
schedule.every(10).minutes.do(job, target_codes)

while True:
  schedule.run_pending()
  time.sleep(1)

Comments

[hfjdsfh_48]

ありがとうございます。Idを格納するリストを作れば思うように動作しました。ご教授いただいたコード例が自分で理解できなかったので、別のやり方をコーディングしました。
import requests
import schedule
import time
import datetime

id_list=[]

def job():
resp = requests.get(f"https://webapi.yanoshin.jp/webapi/tdnet/list/1326-3960-8818.json2?limit=10")
data = resp.json()
result = []
d_today = datetime.date.today()
sdate = d_today.strftime('%Y-%m-%d')

for item in data.get("items", []):
company_id=item.get("id")
pubdate=item.get("pubdate")[:-9]

if sdate!=pubdate:
break

elif company_id in id_list:
break

else:
id_list.append(company_id)

result.append(
dict(
pubdate=item.get("pubdate"),
name = item.get("company_name"),
title = item.get("title"),
url = item.get("document_url")
)
)

for r in result:
data=f"{r['pubdate']} {r['name']} {r['title']} {r['url']}"
print(data)
#以下Lineで通知する処理
schedule.every(1).seconds.do(job)
while True:
schedule.run_pending()
time.sleep(1)

job()

Answer 2

リストを利用すると良いです。以下のような感じです。

  previous_text = []

  for ...:

    if brand_text in previous_text:
      # exist
    else
      # new
      previous_text.append(brand_text)