[Python] Sanity Test on Cash Flow Statement

import logging
import os
import sys

import logger

class SanityCheckerTxt():
   
    def __init__(self):
        self.__logger = logging.getLogger()
        self.white_msg = [
            '資料庫中查無需求資料',
            '無應編製合併財報之子公司',
            '外國發行人免申報個別財務報表資訊，請至合併財務報表查詢',
        ]
        self.check_list_prefixes = {
            'operating_activity' : [
                '營業活動',
            ],
            'investing_activity' : [
                '投資活動',
            ],
            'financing_activity' : [
                '融資活動',
                '理財活動',
                '不影響現金流量之融資活動'
            ],
            'net_income' : [
                '本年度淨利',
                '本期母公司淨利',
                '本期合併淨利',
                '本期合併淨損',
                '本期合併總利益',
                '本期合併總損益',
                '本期純利',
                '本期純益',
                '本期純損',
                '本期純(損)益',
                '本期淨利',
                '本期淨純益',
                '本期淨益',
                '本期淨損',
                '本期淨(損)',
                '本期稅後純益',
                '本期稅後淨利',
                '本期損益',
                '本期總損益',
                '本期(損)純益',
                '合併(損)',
                '合併純益',
                '合併純(損)益',
                '合併淨利',
                '合併淨損益',
                '合併淨損',
                '合併總利益',
                '合併總純益',
                '合併總純損',
                '合併總純(損)益',
                '合併總淨利',
                '合併總淨損',
                '合併總損失',
                '合併總損益',
                '合併總(損)益',
                '純益',
                '純損',
                '純(損)益',
                '淨利',
                '淨損',
                '歸屬於母公司股東之合併純益',
                '歸屬於母公司股東之合併淨利',
                '歸屬於母公司股東之純益',
                '歸屬於母公司股東之淨利',
                '繼續營業單位稅後淨利',
                'A10000', # t05st36_2384_99_02 and t05st36_2384_99_03
            ]
        }

    def check_batch(self, src_dir):
        self.__logger.debug('''Check directory: %s''' % src_dir)
       
        for file in os.listdir(src_dir):
            self.check(os.path.join(src_dir, file))
            #return # for debugging
           
    def check(self, file):
        self.__logger.debug(file)
        assert os.path.isfile(file)

        fd = open(file, 'rb')
        content = fd.read()
        fd.close()

        lines = content.decode('utf-8').split('\n')

        # No record - Test if white message
        if len(lines) is 1:
            msg = lines[0]
            if not msg in self.white_msg:
                self.__logger.info('''%s => %s''' % (file, msg))
            return
       
        # Has record - Test if startswith certain prefixes
        check_list = {
            'operating_activity' : False,
            'investing_activity' : False,
            'financing_activity' : False,
            'net_income' : False,
        }
        for line in lines:
            normed_line = ''.join(line.strip().split())
            normed_line = normed_line.replace('│', '').replace('予', '於')
            normed_line = normed_line.replace('（', '(').replace('）', ')')
            for key in check_list:
                for prefix in self.check_list_prefixes[key]:
                    if normed_line.startswith(prefix):
                        check_list[key] = True
                        break
        for key in check_list:
            if not check_list[key]:
                self.__logger.info('''%s => No %s''' % (file, key))
                self.__logger.debug('\n'.join(lines))


       
def main():
    logger.config_root(level=logging.INFO)
    c = SanityCheckerTxt()
    c.check_batch('./txt/2498');

if __name__ == '__main__':
    sys.exit(main())