main.py

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)