dbis.pm

Perl开发的数据采集工具 基于UNIX

   } 
   
   $rc = $$subdbh->rollback;
   if($rc)
   {
	Trace("Rollbacked Transecation From Database !\n");
	return 1;
   }
   else
   {
	Trace("Err:Rollbacked Transecation From Database Error!\nErr Info:$DBI::errstr\n");
	return 0;
   }
}

###########################################################################
# Function: Exec_Sql
# Description: execute sql statement 
# Input:  
#	1.dbh pointer
#       2.sth_pointer
#	3.sql_statement
#       4.[flag_hold]          --0 not hold and exec finish at end
#       5.flag bind	       --1 bind parameters or not 
#       6.bind_para point            
# Output: None
# Return: undef : Failure
#	  rows	: Successful
###########################################################################
sub Exec_Sql
{
	my($subdbh,$substh,$sub_statement,$subflag,$rows,$sub_rc);
	my($sub_bind,$bind,@bind_paras,$bind_para,@paras,$i);
	$subdbh 	= shift;
	$substh 	= shift;
	$sub_statement 	= shift;
	$subflag 	= shift;
	$bind 		= shift;
	$bind_para	= shift;

	$sub_statement=Replace_SQL_Standard_Type($sub_statement);

	my $currenttime = CFtime('YYYY-MM-DD hh:mm:ss');
	
	$sub_statement =~ s/\bCURRENT\b/\'$currenttime\'/g;
	
	Trace("Exec SQL:\n".$sub_statement);

	if ($DeBug)
	{
	   	Trace("***Programming Debug!! return!!!\n");
	   	return 0;
	} 
	
	if(!($$substh = $$subdbh->prepare($sub_statement)))
	{
		$TraceModes = 'SRC|LOG|REP';		
		Trace("Err:Sql prepare error!\nSQL Statement:$sub_statement\nErr Info:".$DBI::errstr);
		NPM_Send_Message(210);
		NPM_Write_DalLog_bindmod(ERROR_MESSAGE=>"$DBI::errstr");
		OnError($subdbh);
		$TraceModes = 'SRC|LOG';
		return 0;
	}
	if ($bind == 1)
	{
		@bind_paras = $bind_para =~ /\([^\(\)]*\)/g;
		$i = 0;
		foreach $bind_para (@bind_paras)
		{
#			$bind_para = eval($bind_para);
			$bind_para =~ s/\(|\)//g;
			@paras = split(",",$bind_para);
			@paras[2] = eval(@paras[2]);
			Trace("Bind Parameter:$bind_para\n\tParaNo.@paras[0] = @paras[1]");
			$i ++;
			if (@paras[0] != $i)
			{
					Trace("Err:Bind Parameter error!Parameter:@paras");
			}
			else
			{
				$sub_rc = $$substh->bind_param(@paras);
				if (not $sub_rc)
				{
					$TraceModes = 'SRC|LOG|REP';			
					Trace("Err:Execute sql bind error!bind_param\(@paras\)");
					NPM_Send_Message(211);
					NPM_Write_DalLog_bindmod(ERROR_MESSAGE=>"$DBI::errstr");
					OnError($subdbh);
					$TraceModes = 'SRC|LOG';
					return 0;
				}
			}
			
		}
	}
	$sub_rc = $$substh->execute();
	if(not $sub_rc)
	{
		$TraceModes = 'SRC|LOG|REP';			
		Trace("Err:Execute sql error!\nSQL Statement:$sub_statement\nErr Info:".$DBI::errstr);
		NPM_Send_Message(212);
		NPM_Write_DalLog(ERROR_MESSAGE=>"$DBI::errstr");
		OnError($subdbh);
		$TraceModes = 'SRC|LOG';
		return 0;
	}
	
	$rows = $$substh->rows;
	
	if ($subflag == 0)
	{
		$$substh->finish;
		$$substh = $rows;
	}

	Trace("Exec SQL Finished! $rows rows affected!\n");
	return $rows;
}
###########################################################################
# Function: Set_Isdirty_Read
# Description: set isolation dirty to read as informix database
# Input:  
#	1.dbh pointer
#	2.db type
# Output: Null
# Return: Null
###########################################################################
sub Set_Isdirty_Read
{
	my($subdbh,$substh,$db_type,$sql_statement,$sub_rc);
	$subdbh 	= shift;
	$db_type	= shift;
	$sql_statement = 'set isolation dirty read';
	Trace("\n*Set isolation dirty read!\n",0);
	if ($db_type ne 'INFORMIX')
	{
	   	Trace("***This attribute can not support $db_type database type!\n",0);
	   	return 0;
	} 
	if ($DeBug)
	{
	   	Trace("***Programming Debug!! return!!!\n",0);
	   	return 0;
	} 
	
	if(!($substh = $$subdbh->prepare($sql_statement)))
	{
		$TraceModes = 'SRC|LOG|REP';		
		Trace("Warning:Set isolation dirty read error! ".$DBI::errstr);
		$TraceModes = 'SRC|LOG';
		return 0;
	}

	if(!($substh->execute()))
	{
		$TraceModes = 'SRC|LOG|REP';			
		Trace("Warning:Set isolation dirty to read error".$DBI::errstr);
		$TraceModes = 'SRC|LOG';
		return 0;
	}
	if (!$Lock2WaitSec)
	{
		$Lock2WaitSec = 0;
	}
	$sql_statement = "set lock mode to wait $Lock2WaitSec";
	Trace("\n*Set Lock to wait $Lock2WaitSec!\n",0);
	if ($db_type ne 'INFORMIX')
	{
	   	Trace("***This attribute can not support $db_type database type!\n",0);
	   	return 0;
	} 
	
	if(!($substh = $$subdbh->prepare($sql_statement)))
	{
		$TraceModes = 'SRC|LOG|REP';		
		Trace("Warning:Set lock to wait error! ".$DBI::errstr);
		$TraceModes = 'SRC|LOG';
		return 0;
	}

	if(!($substh->execute()))
	{
		$TraceModes = 'SRC|LOG|REP';			
		Trace("Warning:Set lock to wait error! ".$DBI::errstr);
		$TraceModes = 'SRC|LOG';
		return 0;
	}
}
###########################################################################
# Function: Find_Bcp_Oldtab
# Description: 
###########################################################################
sub Find_Bcp_Oldtab
{
	my $mask = shift;
	my $logfile = shift;
	my $tmptable;
	if(!$logfile or !(-e $logfile))
	{
		Trace("Warning:Find_Bcp_Oldtab log file null or not exist!");
		return;
	}
	if(!open(OF,$logfile))
	{
		Trace("Err:Find_Bcp_Oldtab open log file error!");
	}
	
	while(<OF>)
	{
		if(/Bcp\($mask\) to (\w+) SQL:/)
		{
			$tmptable = $1;
			last;
		}
	}
	close(OF);
	return $tmptable;
}
###########################################################################
# Function: Bcp
# Description: execute sql statement 
# Input:  
#\$db_dbh,\$bcp_dbh,\$bcp_tab,$statement,$bind,$bind_para
#	1.dbh pointer
#       2.bcp_dbh pointer
# 	3.bcp_tab
#	4.sql_statement
#       5.flag bind	       --1 bind parameters or not 
#       6.bind_para point            
# Output: Null
# Return: Null
###########################################################################
sub Bcp
{
	use FileHandle;
	my($subdbh,$substh,$subcpdbh,$subcpsth,$subcptab,$sub_statement,$subflag,$rows,$sub_rc);
	my($sub_bind,$bind,@bind_paras,$bind_para,@paras,$i,$errows,$rv);
	my ($i,$counter,$colnum,$coltype,@cols_type,$colname,$counter_sql,$counter_num,
	    $colscale,$colprec,$tabschema,$cols,$cols_noid,$value_cols,$ins_sql,$type,
	    $serial_id,$oldbcptab);
	my (@fetchrow,$ROWS,$line);
	
	$subdbh 	= shift;
	$subcpdbh 	= shift;
	$subcptab	= shift;
	$sub_statement 	= shift;
	$bind 		= shift;
	$bind_para	= shift;
	
	$Bcptab_Serial ++;
	if($Use_Bcp_Oldtab)
	{
		if(!$Bcp_Oldtab_Logfile or !(-e $Bcp_Oldtab_Logfile))
		{
			Trace("Warning: Can not use bcp old table because of old log file null or not exist!");
			$Use_Bcp_Oldtab = 0;
		}
		else
		{
			$oldbcptab = Find_Bcp_Oldtab($Bcptab_Serial,$Bcp_Oldtab_Logfile);
			if(!$oldbcptab)
			{
				Trace("Warning: Can not use bcp old table because of old log file null or not exist!");
			}
			else
			{
				Trace("Bcp($Bcptab_Serial) to $$subcptab SQL:\n".$sub_statement);
				Trace("Use old bcp table $oldbcptab!!\n");
				$$subcptab = $oldbcptab;
				return 0;
			}
		}
	}

	Trace("Bcp($Bcptab_Serial) to $$subcptab SQL:\n".$sub_statement);
	Set_Isdirty_Read($subdbh,$OSSDB_TYPE);
#~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
# count all data number
#~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
	if(!$$ErrorDebug)	#if Error bebug then counter rows
	{
		$counter_sql = $sub_statement;
		
		if(!($substh = $$subdbh->prepare($counter_sql)))
		{
			$TraceModes = 'SRC|LOG|REP';	
			Trace("Bcp SQL Statement:\n$counter_sql\nErr Info:".$DBI::errstr);
			$TraceModes = 'SRC|LOG';
			NPM_Send_Message(210);
			NPM_Write_DalLog_bindmod(ERROR_MESSAGE=>"$DBI::errstr");
			OnError($subdbh);
			return -1;
		}
		$sub_rc = $substh->execute();	
		$counter_num = 0;
		while(@fetchrow = $substh->fetchrow_array)
		{
			$counter_num ++;
		}
		$substh->finish;
	}
	else
	{
		$counter_num = 'unknown';
	}
#~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~	
	if ($DeBug)
	{
	   	Trace("***Programming Debug!! return!!!\n");
	   	return -1;
	} 
	if(!($substh = $$subdbh->prepare($sub_statement)))
	{
		$TraceModes = 'SRC|LOG|REP';			
		Trace("Err:Sql prepare error!\nSQL Statement:$sub_statement\nErr Info:".$DBI::errstr);
		$TraceModes = 'SRC|LOG';
		NPM_Send_Message(210);
		NPM_Write_DalLog_bindmod(ERROR_MESSAGE=>"$DBI::errstr");
		OnError($subdbh);
		return -1;
	}

	if ($bind == 1)
	{
		@bind_paras = $bind_para =~ /\([^\(\)]*\)/g;
		$i = 0;
		foreach $bind_para (@bind_paras)
		{
			$bind_para =~ s/\(|\)//g;
			@paras = split(",",$bind_para);
			@paras[2] = eval(@paras[2]);
			Trace("Bind Parameter:$bind_para\n\tParaNo.@paras[0] = @paras[1]");
			$i ++;
			if (@paras[0] != $i)
			{
					Trace("Err:Bind Parameter error!Parameter:@paras");
			}
			else
			{
				$sub_rc = $substh->bind_param(@paras);
				if (not $sub_rc)
				{
					$TraceModes = 'SRC|LOG|REP';	
					Trace("Err:Execute sql bind error!bind_param\(@paras\)");
					$TraceModes = 'SRC|LOG';
					NPM_Send_Message(211);
					NPM_Write_DalLog_bindmod(ERROR_MESSAGE=>"$DBI::errstr");
					OnError($subdbh);
					return -1;
				}
			}
			
		}
	}
	$sub_rc = $substh->execute();
	if(not $sub_rc)
	{
		$TraceModes = 'SRC|LOG|REP';	
		Trace("Err:Execute sql error!\nSQL Statement:$sub_statement\nErr Info:".$DBI::errstr);
		$TraceModes = 'SRC|LOG';
		NPM_Send_Message(212);
		NPM_Write_DalLog_bindmod(ERROR_MESSAGE=>"$DBI::errstr");
		OnError($subdbh);
		return -1;
	}
	
	Trace ("\nSQL Info:\n",0);
	
	Trace ("Column Name	    TypeName            DataType   Precision  Scale\n",0);
	Trace ("===================================================================\n",0);

	
	$colnum = $substh->{NUM_OF_FIELDS};
	my ($col_infor,$spaces,$infor_tmp);
#~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
#Serial Id
#~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
	$colname = $$subcptab."_ID";
	$cols = $colname.",";
	$coltype = $SQL_TYPE_INFO->{$COOKDB_TYPE}->{4};
	#SQL_INTEGER
	$tabschema 	.= $colname."\t".$coltype.",\n";
	Trace ("$colname  SQL_INTEGER         4          -          -    \n",0);
	$serial_id = 0;
#~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~		
	for($i=0;$i<$colnum;$i++)
	{
#		@typenames 	= map {scalar $dbh->type_info($_)->{TYPE_NAME} } @{ $sth->{TYPE} }
		$type		= $substh->{TYPE}->[$i];
		@cols_type[$i]	= $type;
		#$coltype 	= $$subcpdbh->type_info($type)->{TYPE_NAME};
		$coltype 	= $SQL_TYPE_INFO->{$COOKDB_TYPE}->{$type};
		$colname 	= $substh->{NAME}->[$i];
		$colscale 	= $substh->{SCALE}->[$i];
		$colprec 	= $substh->{PRECISION}->[$i];
		$infor_tmp	= $colname;
		$col_infor      = $infor_tmp;
		$spaces		= ' ';
		$spaces        x= 20 - length($infor_tmp);
		$col_infor     .= $spaces;
		
		$infor_tmp	= $SQL_STANDARD_NUM_TYPE{$type};
		$col_infor     .= $infor_tmp;
		$spaces		= ' ';
		$spaces        x= 20 - length($infor_tmp);
		$col_infor     .= $spaces;
		
		$infor_tmp	= $type;
		$col_infor     .= $infor_tmp;
		$spaces		= ' ';
		$spaces        x= 11 - length($infor_tmp);
		$col_infor     .= $spaces;

		$infor_tmp	= $colprec;
		$col_infor     .= $infor_tmp;
		$spaces		= ' ';
		$spaces        x= 11 - length($infor_tmp);
		$col_infor     .= $spaces;

		$infor_tmp	= $colscale;
		$col_infor     .= $infor_tmp;

		Trace ("$col_infor",0);
		
		if ($colscale <= 0 or $colscale >= $colprec)
		{
			$colscale = 0;
		}
		if($type == 1 or $type == 12 or $type == -1)
		{
			if (($type == 12 or $type == -1) and $colprec > 255)
			{
				$colprec = 255;
			}			
			$coltype = $coltype."(".$colprec.")";
		}
		if($type == 3)
		{
			if ($colprec > 255)
			{
				$colprec = 255;
			}
			$coltype = $coltype."(".$colprec.",".$colscale.")";
		}
		if($type == 2)
		{
			if ($colscale > 0)
			{
				$coltype = $SQL_TYPE_INFO->{$COOKDB_TYPE}->{3};
				$coltype = $coltype."(".$colprec.",".$colscale.")";
			}
		}
		$tabschema 	.= $colname."\t".$coltype.",\n";
		$cols_noid 	.= $colname.","

	}
	chomp($tabschema);
	chop($tabschema);

#	$tabschema .= "PRIMARY KEY (".$$subcptab."_ID)";

	chomp($cols_noid);
	chop($cols_noid);	
	
	$cols = $cols.$cols_noid;
	
	Trace ("===================================================================\n",0);
	$tabschema = "CREATE TABLE $$subcptab\n\(\n".$tabschema."\n\)";
	Trace("\nCREATE TABLE SQL:\n".$tabschema,0);

	if(!($subcpsth = $$subcpdbh->prepare($tabschema)))
	{
		$TraceModes = 'SRC|LOG|REP';	
		Trace("Bcp create table statement:\n$tabschema\nErr Info:".$DBI::errstr);
		$TraceModes = 'SRC|LOG';
		NPM_Send_Message(210);
		NPM_Write_DalLog_bindmod(ERROR_MESSAGE=>"$DBI::errstr");
		OnError($subdbh);
		return -1;
	}
	if(!($subcpsth->execute()))
	{
		$TraceModes = 'SRC|LOG|REP';	
		Trace("Bcp create table statement:\n$tabschema\nErr Info:".$DBI::errstr);
		$TraceModes = 'SRC|LOG';
		NPM_Send_Message(212);
		NPM_Write_DalLog_bindmod(ERROR_MESSAGE=>"$DBI::errstr");
		OnError($subdbh);
		return -1;
	}	
	
	$$subcpdbh->commit;
	
	$TraceModes = 'SRC|LOG';
	Trace("\nBcp all of $counter_num rows to table $$subcptab!!\n",0);
	$TraceModes = "";
#----------------------------INSERT From local database--------------
# This is Not right so can not use
#--------------------------------------------------------------------
	if ($BcpMode == 3)
	{
		$ins_sql = "insert into $$subcptab (".$cols_noid.")\n$sub_statement";
		$TraceModes = 'SRC|LOG';
		Trace("\nBcp from local database(or db link mode)!!\n\n$ins_sql\n",0);
		if(!($rv  = $$subcpdbh->do($ins_sql)))
		{
			$TraceModes = 'SRC|LOG|REP';		
			Trace("Err:Bcp to insert data error!\nSQL Statement:$ins_sql\nErr Info:".$DBI::errstr);
			$TraceModes = 'SRC|LOG';
			NPM_Send_Message(250);
			NPM_Write_DalLog_bindmod(ERROR_MESSAGE=>"$DBI::errstr");
			OnError($subdbh);
			return -1;
		}
		$TraceModes = 'SRC|LOG';
		Trace("Bcp Finished!\n",0);
		return 0;		
	}
#----------------------------INSERT WITH LOAD------------------------
# This is Not right so can not use
#--------------------------------------------------------------------
	if(($COOKDB_TYPE eq 'ORACLE') and ($BcpMode == 2))
	{
		$TraceModes = 'SRC|LOG|REP';
		Trace("Warning:Oracle database can not use bcp load at present!\n");
		$TraceModes = 'SRC|LOG';
		$BcpMode = 0;
	}
#	if ($BcpMode == 2)
#	{
#
#		if (!(-e "$INFORMIXDIR/bin/dbload"))
#		{
#			$TraceModes = 'SRC|LOG|REP';
#			Trace("Warning:$INFORMIXDIR/bin/dbload not exist,can not use bcp load at present!\n");
#			$TraceModes = 'SRC|LOG';
#			$BcpMode = 0;
#		}