DBテーブル日毎に履歴を監視する

###・要件
DBテールは日々にバッチまたオンラインで更新されているとき、何を更新されているかを監視する場合、データを集めることにPythonのPandasを使って実装しました。
・日々更新を監視
・テーブルのキー情報（複数可）、更新日時がある前提です。
###・対応案
PythonのPandasを使う
・日々の差分を比較する（更新日時を除く、全項目を比較する）
・実行モジュールとSQL設定を分ける設定する


###・設定
  tb-monitor-config.ini
  ・今回セクションがT001だけですが、複数のテーブルの場合、Txxxでよい
　・Int型のNULLは0に変更して薦める（Pandasは勝手に型解釈してから、比較しにくい）

```
[DEFAULT]
host = xxxx.cu44c0jm9fqj.ap-northeast-1.rds.amazonaws.com
port = 5432
dbname = xxx_cirus_personal_db
user = xxx
password = xxx

[T001]
csv_all_file = c:\\tmp\\t001_all_mnt.csv.gz

#int field, make the null value
sql_str = SELECT COSTOMER_ID,DATA_SRC,SFID,COALESCE(LVL , 0) as LVL,KGY_ID,JGS_ID,KJB_ID,LBC_CD,FULL_NM,LAST_NM,FIRST_NM,MAIL_ADDR,COMP_NM,
    TEL_BIZ,TEL_MOBILE,ZIP_CD,ADDRESS,DEPARTMENT,Etl_Reg_Date,Etl_Upd_Date,'' as Del_Flag 
    From T001_CUSTOMER 
    where etl_upd_date > '#date_from#'
#   where etl_upd_date >= '2021-12-09 21:31:25.568'
#   and etl_upd_date <= '2021-12-09 21:31:30.000'


keys = COSTOMER_ID
date_stamp = etl_upd_date

```

###・実行モジュール
  csv_out_t001_gz.py
  ・CSVは圧縮
　・Logフォルダを事前に作成

```
import psycopg2
import pandas as pd
from datetime import datetime
import sys
import configparser
import os
import csv
import gzip
import logging
 
sth = logging.StreamHandler()
flh = logging.FileHandler('.\\log\\db_monitor.log')
 
logging.basicConfig(format='%(asctime)s - %(levelname)s - %(message)s', level=logging.INFO,
                    handlers=[sth, flh])
logger = logging.getLogger(__name__)
logger.info("処理を開始します。")


#SectionID
if len(sys.argv) < 2:
   logger.error("SQL Section is neccessary!")
   exit(1)

sect = sys.argv[1]



# --------------------------------------------------
# configparserの宣言とiniファイルの読み込み
# --------------------------------------------------
config_ini = configparser.ConfigParser()
config_ini.read('tb-monitor-config.ini', encoding='utf-8')  


#host info
host = config_ini['DEFAULT']['host']
port = config_ini['DEFAULT']['port']
dbname = config_ini['DEFAULT']['dbname']
user = config_ini['DEFAULT']['user']
password = config_ini['DEFAULT']['password']
con_str = 'host={0} port={1} dbname={2} user={3} password={4}'.format(host,port,dbname,user,password)
logger.info("connect string:{0}".format(con_str))

# merge task
csv_all_file = config_ini[sect]['csv_all_file']
sql_str = config_ini[sect]['sql_str']
keys = config_ini[sect]['keys'].lower().split(',')
date_stamp = config_ini[sect]['date_stamp'].lower()

#  date from 
date_from = '1900-01-01 00:00:00.000'
if os.path.exists(csv_all_file) == True:
    df_all = pd.read_csv(csv_all_file, dtype=str,compression='gzip')
    date_from = max(df_all[date_stamp])
logger.info("dateFrom:{0}".format( date_from))


sql_str = sql_str.replace('#date_from#',date_from)

logger.info("csv file:{0}".format(csv_all_file))
logger.info("sql str:\n{0}".format(sql_str))
logger.info("keys: {0}".format(keys))


#start
start =datetime.now()
logger.info("Query Start")

conn =  psycopg2.connect(con_str)
dff_cur= pd.read_sql_query(sql_str, conn)
#print(df_cur.info())
cnt = dff_cur[date_stamp].count()
logger.info("--Db Counts:{0}".format( cnt ))

if dff_cur[date_stamp].count() == 0 :
    exit(0)

df_cur = dff_cur


#all records file exists
logger.info("Merge all - Start ")
if os.path.exists(csv_all_file) == True:

    logger.info("InFile:{0} Counts:{1}".format(csv_all_file, df_all[date_stamp].count()))

    #set of latest image
    df_max_stamp = df_all[ keys + [ date_stamp ]].fillna('').groupby(keys).max()
    dsh = df_all.fillna('').set_index(keys)
    df_dsh = dsh.loc[df_max_stamp.index,:]\
        .query('{0} in {1}'.format(date_stamp, list(df_max_stamp[date_stamp].values)))
    
    #get ride of duplicate keys
    df_dsh = df_dsh[~df_dsh.index.duplicated()]


    #set index for cur
    df_cur_ind = set(df_cur.fillna('').set_index(keys).index)
    df_dsh_ind = set(df_dsh.index)
    #print(df_cur_ind)
    #print(df_dsh_ind)
    #set del flag for dsh only
    #df_all.set_index(keys, inplace = True)
    #df_all.loc[df_dsh_ind - df_cur_ind,'del_flag'] = '1'

    #current only
    df_cur = df_cur.fillna('').set_index(keys)
    df_cur_only = df_cur.loc[df_cur_ind - df_dsh_ind,:].reset_index()
    logger.info("DbRecords-Add  Counts:{0}".format(df_cur_only[date_stamp].count()) )


    #diff of the common except time stamp
    df_com_idx = df_cur_ind & df_dsh_ind
    df_dsh_com = df_dsh.fillna('').loc[df_com_idx,:].drop(columns=date_stamp)
    df_cur_com = df_cur.fillna('').loc[df_com_idx,:].drop(columns=date_stamp)
    df_dsh_com = df_dsh_com.astype(df_cur_com.dtypes)

    #print(df_cur_com.info())
    #print(df_dsh_com.info())
    #print(df_cur_com == df_dsh_com)
    df_csv_diff = df_cur.loc[df_cur_com[(df_cur_com == df_dsh_com).all(axis=1) == False].index,:].reset_index()
  
    logger.info("DbRecords-Diff Counts:{0}".format( df_csv_diff[date_stamp].count()))

    #new all
    df_all_new = pd.concat([df_all, df_cur_only, df_csv_diff]).sort_values(keys)

else:
    df_all_new = df_cur

startFetch =datetime.now()
logger.info("Merger Over")
df_all_new.to_csv(csv_all_file,
      sep=',',
      header=True,
      index=False,
      quoting=csv.QUOTE_ALL,
      compression='gzip',
      quotechar='"',
      doublequote=True,
      line_terminator='\n')
exec_time = datetime.now()
logger.info("処理終了。Merged Counts:{0}".format(df_all_new[date_stamp].count()))

```

###・実行
  python.exe csv_out_t001_gz.py T001