/*
 * Copyright (c) 2009 The openGion Project.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
 * either express or implied. See the License for the specific language
 * governing permissions and limitations under the License.
 */
package org.opengion.fukurou.process;

import org.opengion.fukurou.system.OgRuntimeException ;		// 6.4.2.0 (2016/01/29)
import org.opengion.fukurou.util.Argument;
import org.opengion.fukurou.util.SystemParameter;
import org.opengion.fukurou.system.LogWriter;

import org.opengion.fukurou.util.HybsEntry ;
import org.opengion.fukurou.db.ConnectionFactory;

import java.util.Set ;
import java.util.HashSet ;
import java.util.Map ;
import java.util.LinkedHashMap ;

import java.sql.Connection;
import java.sql.Statement;
import java.sql.ResultSet;
import java.sql.SQLException;

/**
 * Process_BulkQueryは、データベースから読み取った内容を、一括処理する、
 * FirstProcess と、ChainProcess のインターフェースを両方持った、実装クラスです。
 * ParamProcess のサブクラス(Process_DBParam)にセットしたり、加工したりします。
 *
 * このクラスは、上流から、下流への処理は、１度しか実行されません。
 * FirstProcess の検索結果は、Set オブジェクトとして、Process_DBParam に渡します。
 * ChainProcess は、その結果を取り出し、自分自身の処理結果と合せて加工します。
 *
 * FirstProcess では、-action は、query のみです。
 *   query は、指定のSQL文を実行し、結果のSetをParamProcessに設定します。
 * ChainProcess では、-action は、query、bulkSet、minus、intersect が指定できます。
 *   query     は、上記と同じです。
 *   minus     は、先のSetから、SQL文の実行結果を引き算し、結果Setを再設定します。
 *   intersect は、先のSetから、SQL文の実行結果と重複する結果Setを再設定します。
 *   bulkSet   は、先のSetを取り出し、SQL文に加味して処理します。
 * 流れ的には、query で検索し、minusまたはintersect でSetオブジェクトを加工し、bulkSet で
 * 利用します。例えば、ORACLEから、ユニークキーのSetを作成し、SQLServerのユニークキーを
 * minusした結果を、ORACLEからDELETEすれば、不要なデータを削除するなどの処理が実行可能になります。
 * また、単純に、query だけを、チェインすれば、単発のUPDATE文を実行することが可能です。
 *
 * データベース接続先等は、ParamProcess のサブクラス(Process_DBParam)に
 * 設定された接続(Connection)を使用します。
 * DBID は、Process_DBParam の -configFile で指定する DBConfig.xml ファイルを使用します。
 *
 * 引数文字列中にスペースを含む場合は、ダブルコーテーション("") で括って下さい。
 * 引数文字列の 『=』の前後には、スペースは挟めません。必ず、-key=value の様に
 * 繋げてください。
 *
 * SQL文には、{&#064;DATE.YMDH}等のシステム変数が使用できます。
 *
 * @og.formSample
 *  Process_BulkQuery -action=query -dbid=DBGE -sql="select KEY from TABLE_X"
 *
 *     -action=処理方法(必須)      ： 実行する処理方法を指定します
 *                                       -action=query     単なるSQL文を実行します。
 *                                       -action=bulkSet   実行したSQL文の結果を、Set&lt;String&gt; オブジェクトに設定します。
 *                                       -action=minus     Set&lt;String&gt; オブジェクトと、ここでの実行結果の差分をとります。
 *                                       -action=plus      Set&lt;String&gt; オブジェクトと、ここでの実行結果の加算をします。
 *                                       -action=intersect Set&lt;String&gt; オブジェクトと、ここでの実行結果の積分をとります。
 *   [ -dbid=DB接続ID             ] ： -dbid=DBGE (例: Process_DBParam の -configFile で指定する DBConfig.xml ファイルで規定)
 *   [ -sql=検索SQL文             ] ： -sql="select * from GEA08"
 *   [ -sqlFile=検索SQLファイル   ] ： -sqlFile=select.sql
 *                                       -sql= を指定しない場合は、ファイルで必ず指定してください。
 *   [ -sql_XXXX=固定値           ] ： -sql_SYSTEM_ID=GE
 *                                       SQL文中の{&#064;XXXX}文字列を指定の固定値で置き換えます。
 *                                       WHERE SYSTEM_ID='{&#064;SYSTEM_ID}' ⇒ WHERE SYSTEM_ID='GE'
 *   [ -bulkKey=XXXX              ] ： -bulkKey=XXXX
 *                                       SQL文中の{&#064;XXXX}文字列をProcess_BulkQuery等で取得した値で置き換えます。
 *                                       WHERE SYSTEM_ID IN ( {&#064;XXXX} ) ⇒ WHERE SYSTEM_ID IN ( 'AA','BB','CC' )
 *   [ -bulkType=NUM|STR          ] ： -bulkType=STR
 *                                     Bulkの値を文字列に変換する場合に、数字型か、文字型を指定します。
 *                                     数字型では、AA,BB,CC とし、文字型では、'AA','BB','CC' に変換します(初期値:STR)。
 *   [ -fetchSize=100             ] ：フェッチする行数(初期値:100)
 *   [ -display=[false/true]      ] ：結果を標準出力に表示する(true)かしない(false)か(初期値:false[表示しない])
 *   [ -debug=[false/true]        ] ：デバッグ情報を標準出力に表示する(true)かしない(false)か(初期値:false[表示しない])
 *
 * @og.rev 5.3.4.0 (2011/04/01) 新規追加
 * @version  4.0
 * @author   Kazuhiko Hasegawa
 * @since    JDK5.0,
 */
public class Process_BulkQuery extends AbstractProcess implements FirstProcess , ChainProcess {
	private static final int    MAX_BULK_SET	= 500 ;		// ORACLE の制約が 1000 なので。

	private static final String ACT_QUERY		= "query" ;
	private static final String ACT_BULKSET		= "bulkSet" ;
	private static final String ACT_MINUS		= "minus" ;
	private static final String ACT_PLUS		= "plus" ;			// 6.3.1.1 (2015/07/10) 追加
	private static final String ACT_INTERSECT	= "intersect" ;

	private static final String[] ACTION_LST = { ACT_QUERY,ACT_BULKSET,ACT_MINUS,ACT_PLUS,ACT_INTERSECT };

	private String		actionCmd	;			// SQL結果を加工(query:実行、minus:引き算、intersect:重複分)
	private String		dbid		;			// メインDB接続ID

	private String		bulkKey		;
	private boolean		bulkType	= true;		// true:STR , false:NUM

	private int			sqlCount	;			// SQL文の処理件数
	private int			setCount	;			// 取り出したSetの件数
	private int			outCount	;			// マージ後のSetの件数

	private int			fetchSize	= 100;
	private boolean		display		;			// 表示しない
	private boolean		debug		;			// デバッグ情報
	private boolean		firstTime	= true;		// 最初の一回目

	/** staticイニシャライザ後、読み取り専用にするので、ConcurrentHashMap を使用しません。 */
	private static final Map<String,String> MUST_PROPARTY   ;		// ［プロパティ］必須チェック用 Map
	/** staticイニシャライザ後、読み取り専用にするので、ConcurrentHashMap を使用しません。 */
	private static final Map<String,String> USABLE_PROPARTY ;		// ［プロパティ］整合性チェック Map

	static {
		MUST_PROPARTY = new LinkedHashMap<>();
		MUST_PROPARTY.put( "action",	"実行する処理方法を指定します。(query|minus|plus|intersect)" );

		USABLE_PROPARTY = new LinkedHashMap<>();
		USABLE_PROPARTY.put( "dbid",		"Process_DBParam の -configFile で指定する DBConfig.xml ファイルで規定" );
		USABLE_PROPARTY.put( "sql",		"検索SQL文(sql or sqlFile 必須)例: \"select * from GEA08\"" );
		USABLE_PROPARTY.put( "sqlFile",	"検索SQLファイル(sql or sqlFile 必須)例: select.sql" );
		USABLE_PROPARTY.put( "sql_",		"SQL文中の{&#064;XXXX}文字列を指定の固定値で置き換えます。" +
									CR + "WHERE SYSTEM_ID='{&#064;SYSTEM_ID}' ⇒ WHERE SYSTEM_ID='GE'" );
		USABLE_PROPARTY.put( "dbid2",	"DB接続ID2 例: Process_DBParam の -configFile で指定する DBConfig.xml ファイルで規定" );
		USABLE_PROPARTY.put( "sql2",		"検索SQL文2(sql or sqlFile 必須)例: \"select * from GEA08\"" );
		USABLE_PROPARTY.put( "sqlFile2",	"検索SQLファイル2(sql or sqlFile 必須)例: select.sql" );
		USABLE_PROPARTY.put( "sql2_",	"SQL文2中の{&#064;XXXX}文字列を指定の固定値で置き換えます。" +
									CR + "WHERE SYSTEM_ID='{&#064;SYSTEM_ID}' ⇒ WHERE SYSTEM_ID='GE'" );
		USABLE_PROPARTY.put( "bulkKey",		"SQL文中の{&#064;XXXX}文字列をProcess_BulkQuery等で取得した値で置き換えます。" +
									CR + "WHERE SYSTEM_ID IN ( {&#064;XXXX} ) ⇒ WHERE SYSTEM_ID IN ( 'AA','BB','CC' )" );
		USABLE_PROPARTY.put( "bulkType",		"Bulkの値を文字列に変換する場合に、文字型か、数字型を指定します。" +
									CR + "数字型では、AA,BB,CC とし、文字型では、'AA','BB','CC' に変換します。(初期値:STR)" );
		USABLE_PROPARTY.put( "fetchSize","フェッチする行数 (初期値:100)" );
		USABLE_PROPARTY.put( "display",	"結果を標準出力に表示する(true)かしない(false)か" +
										CR + "(初期値:false:表示しない)" );
		USABLE_PROPARTY.put( "debug",	"デバッグ情報を標準出力に表示する(true)かしない(false)か" +
										CR + "(初期値:false:表示しない)" );
	}

	/**
	 * デフォルトコンストラクター。
	 * このクラスは、動的作成されます。デフォルトコンストラクターで、
	 * super クラスに対して、必要な初期化を行っておきます。
	 *
	 */
	public Process_BulkQuery() {
		super( "org.opengion.fukurou.process.Process_BulkQuery",MUST_PROPARTY,USABLE_PROPARTY );
	}

	/**
	 * プロセスの初期化を行います。初めに一度だけ、呼び出されます。
	 * 初期処理(ファイルオープン、ＤＢオープン等)に使用します。
	 *
	 * @og.rev 5.3.9.0 (2011/09/01) 1000件を超えた場合の処理を追加
	 * @og.rev 6.3.1.1 (2015/07/10) plus アクションの追加
	 *
	 * @param   paramProcess データベースの接続先情報などを持っているオブジェクト
	 */
	public void init( final ParamProcess paramProcess ) {
		final Argument arg = getArgument();

		actionCmd	= arg.getProparty( "action"		, null , ACTION_LST );

		fetchSize	= arg.getProparty( "fetchSize"	, fetchSize );
		display		= arg.getProparty( "display"	, display );
		debug		= arg.getProparty( "debug"		, debug );

		dbid		= arg.getProparty( "dbid");
		String sql	= arg.getFileProparty( "sql"	, "sqlFile",true );
		if( debug ) { println( "入力SQL:" + sql ); }

		final HybsEntry[] entry	=arg.getEntrys( "sql_" );		//配列
		final SystemParameter sysParam = new SystemParameter( sql );
		sql = sysParam.replace( entry );
		if( debug ) { println( "変換SQL:" + sql ); }

		if( ACT_BULKSET.equalsIgnoreCase( actionCmd ) ) {
			bulkKey		= arg.getProparty( "bulkKey" );
			final String bkType = arg.getProparty( "bulkType" );
			if( bkType != null ) { bulkType = "STR".equalsIgnoreCase( bkType ); }	// 初期値が true なので、null チャックは外せません。

			final Set<String> setData = paramProcess.getBulkData();
			if( debug ) { println( setData.toString() ); }
			setCount = setData.size();

			if( setCount > 0 ) {
				// 5.3.9.0 (2011/09/01) 1000件を超えた場合の処理を追加
				final String[] sqls = makeBulkQuery( sql,bulkKey,bulkType,setData );
				for( int i=0; i<sqls.length; i++ ) {
					if( debug ) { println( "BulkSQL:" + sqls[i] ); }
					createSetData( paramProcess, dbid, sqls[i] );
				}
			}
		}
		else if( ACT_QUERY.equalsIgnoreCase( actionCmd ) ) {
			final Set<String> setData2 = createSetData( paramProcess, dbid, sql );
			if( debug ) { println( setData2.toString() ); }
			setCount = setData2.size();
			outCount = setCount;
			paramProcess.setBulkData( setData2 );
		}
		else {
			final Set<String> setData  = paramProcess.getBulkData();
			final Set<String> setData2 = createSetData( paramProcess, dbid, sql );
			setCount = setData2.size();

			if( ACT_MINUS.equalsIgnoreCase( actionCmd ) ) {
				setData.removeAll( setData2 );
			}
			else if( ACT_PLUS.equalsIgnoreCase( actionCmd ) ) {			// 6.3.1.1 (2015/07/10)
				setData.addAll( setData2 );
			}
			else if( ACT_INTERSECT.equalsIgnoreCase( actionCmd ) ) {
				setData.retainAll( setData2 );
			}
			outCount = setData.size();
			if( debug ) { println( setData.toString() ); }
			paramProcess.setBulkData( setData );
		}
	}

	/**
	 * プロセスの終了を行います。最後に一度だけ、呼び出されます。
	 * 終了処理(ファイルクローズ、ＤＢクローズ等)に使用します。
	 *
	 * @param   isOK トータルで、OKだったかどうか [true:成功/false:失敗]
	 */
	public void end( final boolean isOK ) {
		// 何もありません。
	}

	/**
	 * このデータの処理において、次の処理が出来るかどうかを問い合わせます。
	 * この呼び出し１回毎に、次のデータを取得する準備を行います。
	 *
	 * @return	処理できる:true / 処理できない:false
	 */
	public boolean next() {
		return firstTime;
	}

	/**
	 * 引数の LineModel を処理するメソッドです。
	 * 変換処理後の LineModel を返します。
	 * 後続処理を行わない場合(データのフィルタリングを行う場合)は、
	 * null データを返します。つまり、null データは、後続処理を行わない
	 * フラグの代わりにも使用しています。
	 * なお、変換処理後の LineModel と、オリジナルの LineModel が、
	 * 同一か、コピー(クローン)かは、各処理メソッド内で決めています。
	 * ドキュメントに明記されていない場合は、副作用が問題になる場合は、
	 * 各処理ごとに自分でコピー(クローン)して下さい。
	 *
	 * @param	data	オリジナルのLineModel
	 *
	 * @return	処理変換後のLineModel
	 */
	@SuppressWarnings(value={"unchecked"})
	public LineModel action( final LineModel data ) {
		return data ;
	}

	/**
	 * 最初に、 行データである LineModel を作成します
	 * FirstProcess は、次々と処理をチェインしていく最初の行データを
	 * 作成して、後続の ChainProcess クラスに処理データを渡します。
	 *
	 * @param	rowNo	処理中の行番号
	 *
	 * @return	処理変換後のLineModel
	 */
	public LineModel makeLineModel( final int rowNo ) {
		firstTime = false;		// 一度しか処理しないため、false を設定する。

		final LineModel model = new LineModel();

		model.setRowNo( rowNo );

		return model;
	}

	/**
	 * 内部で使用する Set オブジェクトを作成します。
	 * Exception 以外では、必ず Set<String> オブジェクトを返します。
	 *
	 * @og.rev 5.3.9.0 (2011/09/01) 1000件を超えた場合の処理を追加
	 * @og.rev 6.4.2.1 (2016/02/05) try-with-resources 文で記述。
	 *
	 * @param   paramProcess	データベースの接続先情報などを持っているオブジェクト
	 * @param   dbid			接続先ID
	 * @param   sql				実行するSQL文(検索系)
	 *
	 * @return	実行結果から取り出した、最初のカラムのみを集めた Setオブジェクト
	 * @throws RuntimeException データベース処理ができなかった場合。
	 */
	private Set<String> createSetData( final ParamProcess paramProcess, final String dbid, final String sql ) {
		final Set<String> data = new HashSet<>();

		Connection connection	= null;

		try {
			connection = paramProcess.getConnection( dbid );
			// 6.4.2.1 (2016/02/05) try-with-resources 文
			try( final Statement stmt = connection.createStatement() ) {
				if( fetchSize > 0 ) { stmt.setFetchSize( fetchSize ); }
				if( stmt.execute( sql ) ) {			// true:検索系 , false:更新系
					try( final ResultSet resultSet = stmt.getResultSet() ) {
						while( resultSet.next() ) {
							sqlCount++ ;
							final String str = resultSet.getString(1);
							if( display ) { println( str ); }
							data.add( str );
						}
					}
				}
				else {
					sqlCount += stmt.getUpdateCount();
				}
			}
		}
		catch( final SQLException ex) {
			final String errMsg = "SQL を実行できませんでした。" + CR
						+ "errMsg=[" + ex.getMessage() + "]" + CR
						+ "errorCode=[" + ex.getErrorCode() + "] State=[" + ex.getSQLState() + "]" + CR
						+ "DBID=" + dbid + CR
						+ "SQL =" + sql ;

			throw new OgRuntimeException( errMsg,ex );
		}
		finally {

			ConnectionFactory.remove( connection,dbid );
		}
		return data;
	}

	/**
	 * 内部で使用する Set オブジェクトを作成します。
	 * Exception 以外では、必ず Set<String[]> オブジェクトを返します。
	 *
	 * @og.rev 5.3.9.0 (2011/09/01) 1000件を超えた場合の処理を追加
	 *
	 * @param	sql			オリジナルのSQL文
	 * @param	bulkKey		一括処理で置き換えるキー文字列
	 * @param	bulkType	文字型(true)か、数字型(false)を指定
	 * @param   setData		一括処理の元となるSetオブジェクト
	 *
	 * @return	オリジナルのSQL文 に 一括処理の文字列と置換したSQL文の配列
	 */
	private String[] makeBulkQuery( final String sql, final String bulkKey, final boolean bulkType,final Set<String> setData ) {
		String[] sqls = new String[ setData.size()/MAX_BULK_SET + 1 ];
		int idx = 0;
		int cnt = 0;

		final StringBuilder buf = new StringBuilder( BUFFER_MIDDLE );
		String bulkVal = null;
		if( bulkType ) {			// 文字列の場合
			for( final String key : setData ) {
				cnt++;
				buf.append( ",'" ).append( key ).append( '\'' );
				if( cnt >= MAX_BULK_SET ) {
					bulkVal = buf.substring( 1 );		// 先頭のコロンをはずす
					sqls[idx++] = sql.replace( "{@" + bulkKey + "}" ,bulkVal );
					cnt = 0;
					buf.setLength(0);					// 6.1.0.0 (2014/12/26) refactoring
				}
			}
			if( cnt > 0 ) {			// きっちりで終わらない場合
				bulkVal = buf.substring( 1 );	// 先頭のコロンをはずす
				sqls[idx] = sql.replace( "{@" + bulkKey + "}" ,bulkVal );
			}
		}
		else {						// 数字の場合
			for( final String key : setData ) {
				cnt++;
				buf.append( ',' ).append( key );		// 6.0.2.5 (2014/10/31) char を append する。
				if( cnt >= MAX_BULK_SET ) {
					bulkVal = buf.substring( 1 );		// 先頭のコロンをはずす
					sqls[idx++] = sql.replace( "{@" + bulkKey + "}" ,bulkVal );
					cnt = 0;
					buf.setLength(0);					// 6.1.0.0 (2014/12/26) refactoring
				}
			}
			if( cnt > 0 ) {			// きっちりで終わらない場合
				bulkVal = buf.substring( 1 );			// 先頭のコロンをはずす
				sqls[idx] = sql.replace( "{@" + bulkKey + "}" ,bulkVal );
			}
		}

		return sqls;
	}

	/**
	 * プロセスの処理結果のレポート表現を返します。
	 * 処理プログラム名、入力件数、出力件数などの情報です。
	 * この文字列をそのまま、標準出力に出すことで、結果レポートと出来るような
	 * 形式で出してください。
	 *
	 * @return   処理結果のレポート
	 */
	public String report() {
		final String report = "[" + getClass().getName() + "]" + CR
				+ TAB + "Action    : " + actionCmd	+ CR
				+ TAB + "DBID      : " + dbid		+ CR
				+ TAB + "sqlCount  : " + sqlCount	+ CR
				+ TAB + "setCount  : " + setCount	+ CR
				+ TAB + "outCount  : " + outCount ;

		return report ;
	}

	/**
	 * このクラスの使用方法を返します。
	 *
	 * @return	このクラスの使用方法
	 * @og.rtnNotNull
	 */
	public String usage() {
		final StringBuilder buf = new StringBuilder( 1200 )
			.append( "Process_BulkQueryは、データベースから読み取った内容を、一括処理するために、"		).append( CR )
			.append( "ParamProcess のサブクラス(Process_DBParam)にセットしたり、加工したりする"			).append( CR )
			.append( "FirstProcess と、ChainProcess のインターフェースを両方持った、実装クラスです。"	).append( CR )
			.append( CR )
			.append( "このクラスは、上流から、下流への処理は、１度しか実行されません。"					).append( CR )
			.append( "FirstProcess の検索結果は、Set オブジェクトとして、Process_DBParam に渡します。"	).append( CR )
			.append( "ChainProcess は、その結果を取り出し、自分自身の処理結果と合せて加工します。"		).append( CR )
			.append( CR )
			.append( "FirstProcess では、-action は、query のみです。"									).append( CR )
			.append( "  query は、指定のSQL文を実行し、結果のSetをParamProcessに設定します。"			).append( CR )
			.append( "ChainProcess では、-action は、query、bulkSet、minus、intersect が指定できます。"	).append( CR )
			.append( "  query     は、上記と同じです。"													).append( CR )
			.append( "  minus     は、先のSetから、SQL文の実行結果を引き算し、結果Setを再設定します。"	).append( CR )
			.append( "  intersect は、先のSetから、SQL文の実行結果と重複する結果Setを再設定します。"	).append( CR )
			.append( "  bulkSet   は、先のSetを取り出し、SQL文に加味して処理します。"					).append( CR )
			.append( CR )
			.append( "流れ的には、query で検索し、minusまたはintersect でSetオブジェクトを加工し、"		).append( CR )
			.append( "bulkSet で利用します。例えば、ORACLEから、ユニークキーのSetを作成し、"			).append( CR )
			.append( "SQLServerのユニークキーをminusした結果を、ORACLEからDELETEすれば、不要な"			).append( CR )
			.append( "データを削除するなどの処理が実行可能になります。また、単純に、query だけを、"		).append( CR )
			.append( "チェインすれば、単発のUPDATE文を実行することが可能です。"							).append( CR )
			.append( CR )
			.append( "データベース接続先等は、ParamProcess のサブクラス(Process_DBParam)に"			).append( CR )
			.append( "設定された接続(Connection)を使用します。"										).append( CR )
			.append( CR )
			.append( "引数文字列中に空白を含む場合は、ダブルコーテーション(\"\") で括って下さい。"	).append( CR )
			.append( "引数文字列の 『=』の前後には、空白は挟めません。必ず、-key=value の様に"		).append( CR )
			.append( "繋げてください。"																).append( CR )
			.append( CR )
			.append( "SQL文には、{@DATE.YMDH}等のシステム変数が使用できます。"						).append( CR )
			.append( CR ).append( CR )
			.append( getArgument().usage() ).append( CR );

		return buf.toString();
	}

	/**
	 * このクラスは、main メソッドから実行できません。
	 *
	 * @param	args	コマンド引数配列
	 */
	public static void main( final String[] args ) {
		LogWriter.log( new Process_BulkQuery().usage() );
	}
}
