EqualumのFS機能を試す！（道具編1）

####前回は、Equalumを使ってファイルシステム上に送り込まれてくる複数のCSVファイルを、自動的に指定されたデータベースのテーブルに処理する為に必要な関係ノード上の基本的な設定を行いました。今回から2回に分けて、前回構成した環境に自動的にCSVファイルを送り込むツールをPythonで作成していく作業を行います。

##出来上がりのイメージはこんな感じです。
毎度いつものやっつけ仕事的デザインですが、取り急ぎ検証に必要（そう）なデータを想定してサクッと！作ってみたいと思います。

##まずは冒頭部分から・・・
冒頭部分は、いつものお約束的インポート祭りになります。

##################################################
#
# 「物販っぽい・なんちゃってデータ」をCSV形式のファイルに出力・転送
#
# (1) 出力結果はローカルの指定されたディレクトリに格納
# (2) 自動的にリモート側の指定ディレクトリに転送
#　　　(3) ローカルに出力されたデータは転送後に自動的に消去
#
##################################################
import sys
stdout = sys.stdout
sys.stdout = stdout
import os
import glob
import re
import datetime
import time
import hashlib
#
import tkinter as tk
from tkinter import ttk
#
import pymysql.cursors
import csv
import paramiko

##処理に必要な変数を定義します。
今回は、このツールでローカル（MBPで作業していますので・・・）に出力されるCSVファイル（事前に用意したフォルダー内に出力されます）を、自動的にNFSサーバとして暫定利用するEqualumの上流側MySQLノードに転送する処理をPythonのparamikoを使って書きます（冒頭部分の情報はその際に利用します）。

後は、いつものパターンで書いていますので、恒例のNDA（ノン・ダメ出し・アグリーメント）を条件に適宜書き換えてご活用ください。

#
# 転送先のサーバー情報（MySQLノード側）
host='xxx.xxx.xxx.xxx'
host_username='xxxxx'
host_password='xxxxx'
#
# FLOW処理で使用するターゲット側のMySQL情報
TGT_Host = "yyy.yyy.yyy.yyy"
TGT_Port = 3306
TGT_User = "yyyyy"
TGT_Pass = "yyyyy"
TGT_DB = "yyyyy"
TGT_Char = "utf8mb4"
#
# テーブル初期化SQL
Table_Init = "DROP TABLE IF EXISTS "
#
# EQのターゲット側テーブル定義 （Equalumで購入総計と消費税、ポイント、配送センター、 地域を追加するのでここで準備しておく）
DC0 = "id_TGT INT AUTO_INCREMENT, ts_TGT DATETIME(6) DEFAULT CURRENT_TIMESTAMP(6), PRIMARY KEY(id_TGT, ts_TGT), id_CDC INT, ts_CDC DATETIME(6), "
DC1 = "Category VARCHAR(20), Product VARCHAR(20), Price INT, Units INT, Point INT, Logistics VARCHAR(20), "
DC2 = "Card VARCHAR(40), Number VARCHAR(30), Payment INT, Tax INT, "
DC3 = "User VARCHAR(20), Zip VARCHAR(10), Prefecture VARCHAR(10), Area VARCHAR(10), Address VARCHAR(60), Tel VARCHAR(15), Email VARCHAR(40)"
########################################
# グローバル変数の定義
ON  = 0
OFF = 1
Delay = 500   # プログレスバーの表示調整
Wait_Time = 2 # 転送時の時間調整
########################################
#
# 処理で試用するメタデータ
#
# 全部のデータをカラム出力する場合の1行目データ
CSV_Column1 = ["ID", "ts", "Category", "Product", "Price", "Units", "Point", "Payment", "Tax", "Card", "Number", "User", "Zip", "Prefecture", "Address", "Tel", "Email", "Area", "Logistics"]
# EQ連携させる際の1行目カラムデータ
CSV_Column2 = ["ID", "ts", "Category", "Product", "Price", "Units", "Card", "Number", "User", "Zip", "Prefecture", "Address", "Tel", "Email"]
#
# 購入ポイント情報（カテゴリ名の順番に設定
Point_List = [0.02, 0.1, 0.03, 0.02, 0.05]
# 消費税率の設定
Tax_Data = 0.1
# カテゴリ名
Category_List = ["酒類","家電","書籍","DVD/CD","雑貨"]
# 酒類の商品情報
Product_Name0 = ["日本酒","バーボン","ビール","芋焼酎","赤ワイン","白ワイン","スコッチ","ブランデー","泡盛","テキーラ"]
Product_Price0 = [1980, 2500, 490, 2000, 3000, 2500, 3500, 5000, 1980, 2000] 
# 家電の商品情報
Product_Name1 = ["テレビ","洗濯機","ラジオ","ステレオ","電子レンジ","パソコン","電池","エアコン","乾燥機","掃除機"]
Product_Price1 = [49800, 39800, 2980, 88000, 29800, 64800, 198, 64800, 35800, 24800]  
# 書籍の商品情報
Product_Name2 = ["週刊誌","歴史","写真集","漫画","参考書","フィクション","経済","自己啓発","月刊誌","新刊"]
Product_Price2 = [280, 1500, 2500, 570, 1480, 1400, 1800, 1540, 980, 1980]    
# DVD/CDの商品情報
Product_Name3 = ["洋楽","演歌","Jポップ","洋画","アイドル","クラッシック","邦画","連続ドラマ","企画","アニメ"]
Product_Price3 = [1980, 2200, 2500, 3500, 2980, 1980, 3800, 2690, 1980, 2400]
# 雑貨の商品情報
Product_Name4 = ["洗剤","電球","贈答品","医薬部外品","ペットフード","乾電池","文房具","男性用品","女性用品","季節用品"]
Product_Price4 = [498, 198, 1980, 398, 980, 248, 398, 2980, 3580, 1980]
#　地域名ルックアップ情報（キーは都道府県名）
Area_Dict={'北海道':'北海道','青森県':'東北','岩手県':'東北','宮城県':'東北','秋田県':'東北','山形県':'東北','福島県':'東北',
           '茨城県':'関東','栃木県':'関東','群馬県':'関東','埼玉県':'関東','千葉県':'関東','東京都':'関東','神奈川県':'関東',
           '新潟県':'中部','富山県':'中部','石川県':'中部','福井県':'中部','山梨県':'中部','長野県':'中部','岐阜県':'中部','静岡県':'中部','愛知県':'中部',
           '三重県':'近畿','滋賀県':'近畿','京都府':'近畿','大阪府':'近畿','兵庫県':'近畿','奈良県':'近畿','和歌山県':'近畿',
           '鳥取県':'中国','島根県':'中国','岡山県':'中国','広島県':'中国','山口県':'中国',
           '徳島県':'四国','香川県':'四国','愛媛県':'四国','高知県':'四国',
           '福岡県':'九州・沖縄','佐賀県':'九州・沖縄','長崎県':'九州・沖縄','熊本県':'九州・沖縄','大分県':'九州・沖縄','宮崎県':'九州・沖縄','鹿児島県':'九州・沖縄','沖縄県':'九州・沖縄'}
# 物流センタールックアップ情報（キーは地域名）
Logi_Dict = {'北海道':'道央物流センター','東北':'東北物流センター','関東':'関東中央物流センター',
             '中部':'甲州物流センター','近畿':'伊丹物流センター','中国':'広島臨港物流センター','四国':'讃岐物流センター','九州・沖縄':'平戸物流センター'}
#
TGT_Message = "ターゲット側にテーブルを作成しました。"
GEN_Message = "データの生成・挿入処理を終了しました。"

###具体的な処理を記述します・・・・
まずは露払い的な処理を書いていきます。此処もいつもの定型的な記述なので特に難しい所は無いかと。

#
print("----------------------------------------------------")
print("********************* 処理開始 *********************")
print(datetime.datetime.now().strftime("%Y/%m/%d %H:%M:%S") + " : 検証用のCSVファイルを作ります")
print("----------------------------------------------------")
#
# Fakerの初期化
from faker import Faker
fakegen = Faker('ja_JP')    
Faker.seed(fakegen.random_digit())
#
#　GUI部分の作成
root = tk.Tk()
root.title("CSVデータ生成ツール")
root.geometry("420x470")
#
# GUIスタイルの設定（この設定の場合、プログレスバーが太く表示される）
style = ttk.Style()
style.theme_use('default')
#
# SQL処理状況のカラー設定
style.configure("Orange.Horizontal.TProgressbar", foreground='orange', background='orange')
#
# アプリの終了とウインドウの消去
def Exit_Tool(): root.destroy()
#
# SQL処理プログレスバーの更新
def var_start1(value_bar): progressbar1.configure(value=value_bar)
#
# MySQL上にターゲット用のテーブルを作成する
def TGT_Table():
        
        # テーブル名を取得
        Table_Name = TBL_Name.get("1.0", "end").replace('\n','')
   
        # デモ用のテーブルの作成SQL 
        Table_Create = "CREATE TABLE IF NOT EXISTS " + Table_Name + " (" + DC0 + DC1 + DC2 + DC3 + ")"  
    
        # MySQLとの接続（此処では、TGT設定されている側のMySQLを選択）
        db = pymysql.connect(host = TGT_Host,
                         port = TGT_Port,
                         user = TGT_User,
                         password = TGT_Pass,
                         db = TGT_DB,
                         charset = TGT_Char,
                         cursorclass=pymysql.cursors.DictCursor)

        # ターゲット側に格納テーブルを作成                
        with db.cursor() as cursor:
       
                # 既存テーブルの初期化
                cursor.execute(Table_Init + Table_Name)
                db.commit()
       
                #　新規にテーブルを作成
                cursor.execute(Table_Create)    
                db.commit()

        # データベースを切断
        db.close()

        # ステータスバーに処理終了を表示
        statusbar["text"] = " " + TGT_Message

###次に物販系なんちゃってデータ生成部分
此処はある意味で「今回の最初のヤマ場・・」ですが、基本的にはFakerに頼り切った形の力押し的な記述の連続になります。

#
# 商材系のデータを生成
def Data_Gen_1():

        # EQ連携の有無を確認
        EQ_Test= int(combo_dict1[cb1.get()])

        # 作業用のリストを初期化
        Data_List = []

        # ランダムに書き込む商材の種類と商品IDを選択
        Category_ID = fakegen.random_digit()
        if Category_ID > 4: Category_ID = Category_ID - 5              
        Product_ID = fakegen.random_digit()

        # カテゴリ名の設定  
        Category = Category_List[Category_ID]

        # カラム情報の設定
        if Category_ID == 0: # 酒類
                Product = Product_Name0[Product_ID]
                Price   = Product_Price0[Product_ID]
                Units   = fakegen.random_digit() + 1  # リアルっぽく調整しています
                Point   = int(Price * Units * Point_List[Category_ID])  

        elif Category_ID == 1: # 家電
                Product = Product_Name1[Product_ID]
                Price   = Product_Price1[Product_ID]
                Units   = 1  # リアルっぽく調整しています
                Point   = int(Price * Units * Point_List[Category_ID])

        elif Category_ID == 2: # 書籍
                Product = Product_Name2[Product_ID]
                Price   = Product_Price2[Product_ID]
                Units   = fakegen.random_digit() + 1
                if Units >3: Units = 3  # リアルっぽく調整しています  
                Point   = int(Price * Units * Point_List[Category_ID])

        elif Category_ID == 3: # DVD/CD
                Product = Product_Name3[Product_ID]
                Price   = Product_Price3[Product_ID]
                Units   = fakegen.random_digit() + 1
                if Units >2: Units = 2  # リアルっぽく調整しています
                Point   = int(Price * Units * Point_List[Category_ID])    

        else: # 雑貨
                Product = Product_Name4[Product_ID]
                Price   = Product_Price4[Product_ID]
                Units   = fakegen.random_digit() + 1
                if Units >4: Units = 4  # リアルっぽく調整しています
                Point   = int(Price * Units * Point_List[Category_ID])

        # 支払い総額と消費税
        Payment = Units * Price
        Tax     = int(Payment * Tax_Data)

        # リスト形式でデータを戻す
        if (EQ_Test == OFF): Data_List = [Category, Product, Price, Units, Point, Payment,  Tax]
        else:                Data_List = [Category, Product, Price, Units]

        # リストデータを戻す
        return(Data_List)
#
# 支払い系のデータを生成
def Data_Gen_2():

        # 結果を返すリスト領域を初期化
        Data_List = []

        # 支払い情報の生成
        if str(fakegen.pybool()) == "True": Card = "現金"
        else:                               Card = fakegen.credit_card_provider()

        # カード番号の生成
        Number = fakegen.credit_card_number()              
        if Card == "現金":                  Number = "N/A"

        # リスト形式でデータを返す
        Data_List = [Card, Number]

        # リストデータを戻す
        return(Data_List)
#
# 購入者系のデータを生成
def Data_Gen_3(Area_Dict, Logi_Dict):

        # EQ連携の有無を確認
        EQ_Test = int(combo_dict1[cb1.get()])

        # 作業用のデータリストを初期化
        Data_List = []

        # 購入者情報の生成
        User    = fakegen.name()
        Zip     = fakegen.zipcode()
        Address = fakegen.address()
        Tel     = fakegen.phone_number()
        Email   = fakegen.ascii_email()

        # 都道府県名の抽出
        pattern = u"東京都|北海道|(?:京都|大阪)府|.{2,3}県"
        m = re.match(pattern , Address)
        if m: Prefecture = m.group()

        # 地域名と物流センター名を取得           
        Area = Area_Dict[Prefecture]
        Logistics = Logi_Dict[Area]

        # リスト形式でデータを戻す
        if (EQ_Test == OFF): Data_List = [User, Zip, Prefecture, Address, Tel, str(Email), Area, Logistics]
        else:                Data_List = [User, Zip, Prefecture, Address, Tel, str(Email)]

        # リストデータを戻す
        return(Data_List)

##今回のまとめ
今回は、Pythonを使った検証用のCSVファイル生成ツールの導入分を紹介させて頂きました。

次回は、引き続き残りの部分（実際に設定条件に合わせたCSVファイルを作成する）のコードを紹介させて頂きます。