【Python】YahooショッピングAPIを使用して、カテゴリ別の売り上げランキング取得作業を自動化してみた

Book名:第1階層カテゴリID_カテゴリ名.xlsx
┗Sheet名:第1階層カテゴリ名
┗Sheet名:第2階層カテゴリ名
・
・
・

.
┗result
┗config.py
┗make_category_json.py
┗make_ranking_files.py

PATH_JSON_CATE = r'./category.json'

'''
api使用に必要な下記要素をそれぞれ記述してください
    ・アプリケーションID
    ・application_secret
    ・クリエイターProと紐づけしてあるyahooアカウントのid及びpw
    ・ショップコード
'''
MY_INFO = {
    'YAPI_ID':'*',
    'YAPI_SEC':'*',
    'Y_ID':'*',
    'Y_PW':'*',
    'SHOP_ID':'*'
}

import requests,json,time
from config import MY_INFO,PATH_JSON_CATE
REQ_URL_CATE = 'https://shopping.yahooapis.jp/ShoppingWebService/V1/json/categorySearch'

###カテゴリ取得処理メイン関数
def main():
    
    #親カテゴリ取得
    dict_cate = {}    
    cate_params = {
        'appid':MY_INFO['YAPI_ID'],
        'output':'json',
        'category_id':1
    }

    dict_cate = fetch_category_info(cate_params)

    #カテゴリ辞書が空の場合は処理終了
    if len(dict_cate) == 0:
        print("###Fetch parent category Failed.Exit processing###")
        return False

    #子カテゴリ取得
    for id,val in dict_cate.items():
        print(f"Now category --> {val['name']}")
        cate_params['category_id'] = id

        tmp_dict = fetch_category_info(cate_params)

        if len(tmp_dict) == 0:
            print(f"Fetch category Failed --> {val['name']}")
        else:
            dict_cate[id]['children'] = tmp_dict
            print(f"Fetch category Succeeded --> {val['name']}")

    #jsonファイル作成
    with open(PATH_JSON_CATE,mode='w',encoding='utf-8') as f:
        json.dump(dict_cate,f,indent=4,ensure_ascii=False)
    
    return True

#カテゴリ取得関数
def fetch_category_info(arg_params):
    
    func_dict = {}
    res = requests.get(url=REQ_URL_CATE,params=arg_params)
    
    if res.status_code == 200:
        
        res = json.loads(res.text)
        for elm in res['ResultSet']['0']['Result']['Categories']['Children'].values():
            try:
                func_dict[elm['Id']] = {'name':elm['Title']['Short']}
            except Exception as e:
                print(f"##Detected Error --> {e}##\n##value --> {elm}##")
        
    else:
        print(f"API request failed --> {res.status_code}:{res.text}")
    
    time.sleep(1)    
    return func_dict
    
if __name__ == '__main__':
    main()

{
    "13457": {
        "name": "ファッション",
        "children": {
            "2494": {
                "name": "レディースファッション"
            },
            "2495": {
                "name": "メンズファッション"
            },
            "2496": {
                "name": "腕時計、アクセサリー"
            }
        }
    },
    "2498": {
        "name": "食品",
        "children": {
            "2499": {
                "name": "ドリンク、水、お酒"
            },
・
・
・

import os,sys,re,json,time,datetime
import requests,openpyxl
import make_category_json
import pandas as pd
from config import MY_INFO,PATH_JSON_CATE

REQ_URL_RANK = 'https://shopping.yahooapis.jp/ShoppingWebService/V2/categoryRanking'
DIR_RESULT = r'./result'
ranking_params = {
    'appid' : MY_INFO['YAPI_ID'],
    'category_id' : 0,
    'generation' : 'all',
    'period' : 'weekly',
    'offset' : 1 #指定した順位を20位毎
}
re_itemcode = re.compile('(?<=_).+$') #itemcode取得用正規表現
re_sheetname = re.compile(r'(:|\\|\?|\[|\]|\/|\*)') #シート名編集用正規表現

def main():

    sta_time = datetime.datetime.now()
    print(f"処理開始 --> {sta_time}")

    ##カテゴリーjsonファイル存在判定関数呼び出し
    dict_cate = is_category_json()
    if len(dict_cate) == 0:
        print('###Fetch parent category Failed.Exit processing###')
        sys.exit()

    ##結果フォルダ作成関数呼び出し
    path_result = make_result_dir()
    
    for p_id,p_val in dict_cate.items():
        
        cate_name = is_suitble_sheet_name(p_val['name'])
        path_result_file = f"{path_result}/{p_id}_{cate_name}.xlsx"
        
        wb = openpyxl.Workbook()
        wb.save(path_result_file)

        #親ジャンルランキング取得
        print(f"{'#'*5}\nNow category --> {p_id}:{cate_name}")
        fetch_category_ranking(p_id,cate_name,path_result_file)
        
        #子ジャンルランキング取得
        for c_id,c_val in p_val['children'].items():
    
            cate_name = is_suitble_sheet_name(c_val['name'])
            
            print(f"{'#'*5}\nNow category --> {c_id}:{cate_name}")
            fetch_category_ranking(c_id,cate_name,path_result_file)
    
        #初期シート削除
        remove_initial_sheet(path_result_file)
    
    fin_time = datetime.datetime.now()
    print(f"{'#'*5}\n処理時間 --> {fin_time - sta_time}\n{'#'*5}")

'''
初期シート削除関数
'''
def remove_initial_sheet(arg_path):

    wb_load = openpyxl.load_workbook(arg_path)
    ws_load = wb_load.worksheets[0]
    if len(wb_load.worksheets) != 1:
        wb_load.remove(ws_load)
    else:
        ws_load.title = '取得失敗'
        ws_load['A1'] = '取得失敗'
    wb_load.save(arg_path)
    wb_load.close()

'''
カテゴリーjsonファイル存在判定関数
存在する -> カテゴリーの連想配列を返す
存在しない -> カテゴリー取得モジュール呼び出し、失敗の場合は空の連想配列を返す
'''
def is_category_json():

    ###jsonファイル存在判定
    is_exists = os.path.isfile(PATH_JSON_CATE)
    if is_exists == False:
        
        ##存在しない場合、カテゴリー取得モジュールを呼び出し
        is_success = make_category_json.main()
        
        ##カテゴリー取得に失敗した場合、空の連想配列を返す
        if is_success == False:
            func_dict = {}
            print(f'###Fetch category --> Failed')
            return func_dict

    with open(PATH_JSON_CATE,mode='r',encoding='utf-8') as f:
        func_dict = json.load(f)
    
    print(f'###Fetch category --> Succeeded')        
    return func_dict

'''
結果フォルダ作成関数
存在する -> resultフォルダ内に本日日付のフォルダを作成しそのファイルパスを返す
存在しない -> resultフォルダを作成し、上記と同じ処理を行う
'''
def make_result_dir():

    ##結果フォルダ存在判定
    is_exists = os.path.isdir(DIR_RESULT)
    if  is_exists == False:
        os.mkdir(DIR_RESULT)
    
    func_path = rf"{DIR_RESULT}/{format(datetime.datetime.today(),'%Y%m%d')}"
    os.mkdir(func_path)

    return func_path

'''
シート名編集関数
以下条件を満たさない場合、編集対象
・不適切な文字
・不適切な文字数
'''
def is_suitble_sheet_name(arg_name:str):
    
    #文字判定(不適切文字はアンダーバーに置き換え)
    match_result = re_sheetname.findall(arg_name)
    if match_result:            
        arg_name = re_sheetname.sub('_',arg_name)
    
    #文字数判定
    if len(arg_name) >= 31:
        arg_name = arg_name[0:31]
    
    return arg_name

'''
ランキングデータ作成関数
'''
def fetch_category_ranking(arg_id:int,arg_name:str,arg_path:str):
    
    ranking_params['category_id'] = arg_id

    res = requests.get(REQ_URL_RANK,params=ranking_params)
    time.sleep(1)

    if res.status_code != 200:
        print(f"Api request Failed --> {res.status_code}:{res.text}")
        return False
    else:
        res = json.loads(res.text)
        ranking_data = res['category_ranking']['ranking_data']
                
        ##各列データseries作成
        d_rank = pd.Series(data=[elm['rank'] for elm in ranking_data],name='rank')
        d_icode = pd.Series(data=[re_itemcode.search(elm['image']['id']).group(0) for elm in ranking_data],name='ItemCode')
        d_iname = pd.Series(data=[elm['name'] for elm in ranking_data],name='ItemName')
        d_review_r = pd.Series(data=[elm['review']['rate'] for elm in ranking_data],name='ReviewRate')
        d_review_c = pd.Series(data=[elm['review']['count'] for elm in ranking_data],name='ReviewCount')
        d_scode = pd.Series(data=[elm['store']['id'] for elm in ranking_data],name='ShopCode')
        d_sname = pd.Series(data=[elm['store']['name'] for elm in ranking_data],name='ShopName')
        d_iurl = pd.Series(data=[elm['url'] for elm in ranking_data],name='ItemUrl')
        d_surl = pd.Series(data=[elm['store']['url'] for elm in ranking_data],name='ShopUrl')

        ##series -> dataframe
        d_rank = pd.DataFrame(d_rank)
        d_icode = pd.DataFrame(d_icode)
        d_iname = pd.DataFrame(d_iname)
        d_review_r = pd.DataFrame(d_review_r)
        d_review_c = pd.DataFrame(d_review_c)
        d_scode = pd.DataFrame(d_scode)
        d_sname = pd.DataFrame(d_sname)
        d_iurl = pd.DataFrame(d_iurl)
        d_surl = pd.DataFrame(d_surl)

        #各要素のdataframe結合
        df = pd.concat([d_rank,d_icode,d_iname,d_review_r,d_review_c,d_scode,d_sname,d_iurl,d_surl],axis=1)

        #dataframe -> excel
        with pd.ExcelWriter(arg_path, mode='a') as writer:
            df.to_excel(writer, sheet_name=arg_name,index=False)            

        print(f"Make ranking sheet --> Succeeded")
        return True

if __name__ == '__main__':
    main()