from src.get_doc_id import GetDocID from datetime import date, timedelta import pandas as pd import os import warnings from src.get_stock_code import StockCode_getter from src import group_classification from src.myLogger import set_logger from answers import search, group2_classification import csv import openpyxl from src import group1_scoring, group2_scoring, group3_scoring, group4_scoring # ログ設定を初期化 logger = set_logger("my_app", level="INFO") edinet_api_key = os.getenv("EDINET_API_KEY") # 警告を特定のものに限定する warnings.filterwarnings('ignore', category=DeprecationWarning) def find_docID(security_code): df = pd.read_csv('./data/csv/edinet_df.csv') filtered_df = df[df['証券コード'] == security_code] # Check if any rows were found if not filtered_df.empty: # Get the 'docID' from the filtered DataFrame docID = filtered_df.iloc[0]['docID'] return docID else: return -1 def find_ir(security_code): docID = find_docID(security_code) if docID != -1 or docID is not None: PDF = "https://disclosure2dl.edinet-fsa.go.jp/searchdocument/pdf/" + str(docID) + ".pdf" return PDF def ph2_group_classification(company_name): group, related_url_list, unrelated_url_list, other_url_list = search(company_name) logger.info(f'{company_name}: {group}') if group == 'Group 1-1': scoring_result, url = group1_scoring.scoring(company_name) logger.info(f'Scoring result: {scoring_result}') elif group == 'Group 3': scoring_result, url = group3_scoring.scoring(company_name) logger.info(f'Scoring result: {scoring_result}') elif group == 'Group 1-2': path = 'data/Group1-2_result.csv' scoring_result = [] with open(path, mode='a', newline='') as file: row = [company_name, group, related_url_list] writer = csv.writer(file) writer.writerow(row) else: path = 'data/Group5_result.csv' scoring_result = [] with open(path, mode='a', newline='') as file: row = [company_name, group, related_url_list] writer = csv.writer(file) writer.writerow(row) return group, related_url_list, unrelated_url_list, other_url_list, scoring_result def init_csv(): path1 = 'data/Classification_result.csv' header = ['企業名', 'Group', '関連URL', '関連無しURL', '開けなかったURL'] with open(path1, mode='w', newline='') as file: writer = csv.writer(file) writer.writerow(header) # scoring_tier1-1.csv 初期化 path = 'data/Group1-1_result.csv' header1_1 = ['企業名', "グループ名", '脱炭素言及内容', '関連URL'] with open(path, mode='w', newline='') as file: writer = csv.writer(file) writer.writerow(header1_1) # scoring_tier1-2.csv 初期化 path = 'data/Group1-2_result.csv' header1_2 = ['企業名', "グループ名", '関連URL'] with open(path, mode='w', newline='') as file: writer = csv.writer(file) writer.writerow(header1_2) # scoring_tier2.csv 初期化 path = 'data/Group2_result.csv' scoringList_Group2 = [ '再生可能エネルギーを使用している', '太陽光発電を行っている', '風力発電を行っている', 'バイオマスエネルギーの使用をしている', 'その他の再生可能エネルギーを使用している', 'その他の再生可能エネルギーの例', 'エネルギー使用量の削減を行っている', '電気自動車の利用', '暖房や給湯などでヒートポンプを利用している', 'その他の方法でエネルギー使用量の削減を行っている', 'その他のエネルギー使用量の削減の例' ] header2 = ['企業名', 'グループ分類'] + scoringList_Group2 + ['スコア', '関連ありURL'] with open(path, mode='w', newline='') as file: writer = csv.writer(file) writer.writerow(header2) # scoring_tier3.csv 初期化 path = 'data/Group3_result.csv' scoringList_Group3 = [ '再生可能エネルギーを使用している', '太陽光発電を行っている', '風力発電を行っている', 'バイオマスエネルギーの使用をしている', 'その他の再生可能エネルギーを使用している', 'その他の再生可能エネルギーの例', 'エネルギー使用量の削減を行っている', '電気自動車の利用', '暖房や給湯などでヒートポンプを利用している', 'その他の方法でエネルギー使用量の削減を行っている', 'その他のエネルギー使用量の削減の例' ] header3 = ['企業名', 'グループ分類'] + scoringList_Group3 + ['スコア', '関連ありURL'] with open(path, mode='w', newline='') as file: writer = csv.writer(file) writer.writerow(header3) path = 'data/Group4_result.csv' scoringList_Group4 = [ '再生可能エネルギーを使用している', '太陽光発電を行っている', '風力発電を行っている', 'バイオマスエネルギーの使用をしている', 'その他の再生可能エネルギーを使用している', 'その他の再生可能エネルギーの例', 'エネルギー使用量の削減を行っている', '電気自動車の利用', '暖房や給湯などでヒートポンプを利用している', 'その他の方法でエネルギー使用量の削減を行っている', 'その他のエネルギー使用量の削減の例' ] header4 = ['企業名', 'グループ分類'] + scoringList_Group4 + ['スコア', '関連ありURL'] with open(path, mode='w', newline='') as file: writer = csv.writer(file) writer.writerow(header4) # scoring_group5.csv 初期化 path = 'data/Group5_result.csv' header5 = ['企業名', "グループ名", '関連URL'] with open(path, mode='w', newline='') as file: writer = csv.writer(file) writer.writerow(header5) return (header1_1, header1_2, header2, header3, header4, header5) def main(company_name, docID): # # =====初回のみ実行(有価証券報告書の各情報取得)==== # if not os.path.exists("./data/csv/edinet_df.csv"): # today = date.today() # one_year_ago = today - timedelta(days=365) # gd = GetDocID(start_date=one_year_ago, end_date=today, edinet_api_key=edinet_api_key) # edinet_df = gd.create_docid_df() # edinet_df.to_csv("./data/csv/edinet_df.csv",index=False) # # ========================================== logger.info(f"{company_name} docID: {docID}") if docID != -1: # docID += "0" # PDF = find_ir(docID) PDF = "https://disclosure2dl.edinet-fsa.go.jp/searchdocument/pdf/" + str(docID) + ".pdf" logger.info(f'{company_name} 有価証券報告書: {PDF}') result = group_classification.main(PDF) if result == '(1, 1)': group = 'Group 4' logger.info(f'{company_name}: Group 4') related_url_list = PDF scoring_result, urls = group4_scoring.scoring(company_name) unrelated_url_list = ' ' other_url_list = ' ' return group, related_url_list, unrelated_url_list, other_url_list, scoring_result elif result == '(1, 0)': url = [PDF] group, related_url_list, unrelated_url_list, other_url_list = group2_classification(company_name) logger.info(f"related_url_list: {related_url_list}") if isinstance(related_url_list, list): related_url_list += url else: related_url_list = [PDF] logger.info(f'{company_name}: {group}') if group == 'Group 2': scoring_result, urls = group2_scoring.scoring(company_name) logger.info(f'Scoring result: {result}') else: scoring_result = [] path = 'data/Group1-2_result.csv' with open(path, mode='a', newline='') as file: row = [company_name, group, related_url_list] writer = csv.writer(file) writer.writerow(row) return group, related_url_list, unrelated_url_list, other_url_list, scoring_result else: return ph2_group_classification(company_name) else: return ph2_group_classification(company_name) if __name__ == "__main__": # 企業リストを分類し、コンソールに分類結果を出力 # company_list = ["日本石油販売株式会社（D&Dstudio）"] # for company_name in company_list: # logger.info(main(company_name)) # print("==========\n") # data/会社情報.xlsxが企業リストで企業リストを分類し、csvファイルに出力 stock_code_getter = StockCode_getter() _, _, _, _, _, _ = init_csv() path = 'data/Classification_result.csv' workbook = openpyxl.load_workbook('data/test_data.xlsx') sheet = workbook.active for row in sheet.iter_rows(min_row=2, max_row=sheet.max_row, min_col=2, max_col=2): for cell in row: company_name = cell.value docID = stock_code_getter.get_stock_code(company_name) group, related_url, unrelated_url, other_url, scoring_result = main(company_name, docID) data = [company_name, group, related_url, unrelated_url, other_url] with open(path, mode='a', newline='') as file: writer = csv.writer(file) writer.writerow(data) print("==========\n") # 一時保存用pdfの削除 file_path = 'copy.pdf' if os.path.exists(file_path): os.remove(file_path)