dbis.pm

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

#	}
	if ($BcpMode == 2)
	{
		if(!$Data_Load_TmpDir)
		{
			$Data_Load_TmpDir = "$DAL_HOME/tmp/dbload";
		}
		my $dbloaddatfile 	= "$Data_Load_TmpDir/$$subcptab.unl";
		my $dbloadcmdfile 	= "$Data_Load_TmpDir/$$subcptab.cmd";
		my $dbloadlogdfile 	= "$Data_Load_TmpDir/$$subcptab.log";
		my $dbloadsql		= "$Data_Load_TmpDir/$$subcptab.sql";
		my $dbloadcmd 		= "$INFORMIXDIR/bin/dbload -d $Wnms_Database -c $dbloadcmdfile -l $dbloadlogdfile";

#----------------------------------------------------------------------------		
#dbload [-d dbname] [-c cfilname] [-l logfile] [-e errnum] [-n nnum]
#        [-i inum] [-s] [-p] [-r | -k] [-X]
#
#        -d      database name
#        -c      command file name
#        -l      bad row(s) log file
#        -e      bad row(s) # before abort
#        -s      syntax error check only
#        -n      # of row(s) before commit
#        -p      prompt to commit or not on abort
#        -i      # or row(s) to ignore before starting
#        -r      loading without locking table
#        -X      recognize HEX escapes in character fields
#        -k      loading with exclusive lock on table(s)
#----------------------------------------------------------------------------
		
		if (!(-e "$Data_Load_TmpDir") or !(-d "$Data_Load_TmpDir"))
		{
			&Touch_Dir("$Data_Load_TmpDir");
		}
		
		if(!(open(CMDFILE,">".$dbloadcmdfile)))
		{
			$TraceModes = 'SRC|LOG|REP';
			Trace("Err:Create dbload cmd file $dbloadcmdfile error!\n");
			$TraceModes = 'SRC|LOG';
			OnError($subcpdbh);
			return -1;			
		}
		
		print CMDFILE "FILE $Data_Load_TmpDir/$$subcptab.unl DELIMITER \'|\' ".($colnum+1).";\n";
		print CMDFILE "INSERT INTO $$subcptab;";	
		close(CMDFILE);

		$TraceModes = 'SRC|LOG';
		Trace("\nBcp with Load to File $dbloaddatfile!!Waiting...\n",0);
		$TraceModes = "";
		my $dfile = new FileHandle(">$dbloaddatfile");
		if(!defined $dfile)
		{
			$TraceModes = 'SRC|LOG|REP';
			Trace("Err:Open data file $dbloaddatfile error!\n");
			$TraceModes = 'SRC|LOG';
			OnError($subcpdbh);
			return -1;			
		}

		$serial_id = 0;
		my $file_buf;
		while(@fetchrow = $substh->fetchrow_array)
		{
			$line = "";
			$serial_id ++;
			$line = $serial_id.'|';
			for($i=0;$i<@fetchrow;$i++)
			{
				@fetchrow[$i] = Trim(@fetchrow[$i]);
				if ($substh->{TYPE}->[$i] == 11)
				{
					if (@fetchrow[$i] =~ /^\s*(\d{4}-\d{2}-\d{2}\s\d{2}:\d{2}:\d{2}).*$/)
					{
						@fetchrow[$i] = $1;
					}
				}
				$line .= @fetchrow[$i]."|";
				
			}
			chop($line);
			$file_buf .= $line."\n";
			$rows ++;
			if (($rows % 1000) == 0)
			{
				print $dfile $file_buf;
				undef $file_buf;
				printf ("\n\t%7d of $counter_num rows have finished!",$rows);
				
			}
		}
		print $dfile $file_buf;
		undef $file_buf;
		close($dfile);
		
		$TraceModes = 'SRC|LOG';
		Trace("\nBcp with Load To Table $$subcptab!!Waiting...\n\n",0);
		
		if(!(open(CMDSQL,">".$dbloadsql)))
		{
			$TraceModes = 'SRC|LOG|REP';
			Trace("Err:Create dbload cmd file $dbloadcmdfile error!\n");
			$TraceModes = 'SRC|LOG';
			OnError($subcpdbh);
			return -1;			
		}
		print CMDSQL "Load from $dbloaddatfile delimiter \'|\' insert into $$subcptab";
		close(CMDSQL);
		$net->cmd("dbaccess mddb $dbloadsql");
		$net->close;
		
		
		#system($dbloadcmd);
		if($Del_Bcpfile)
		{
			unlink($dbloaddatfile);
			unlink($dbloadcmdfile);
			unlink($dbloadlogdfile);
		}
		Trace("Bcp Finished!\n",0);
		return 0;
	}
#----------------------------INSERT NOT BIND------------------------
  
	if ($BcpMode == 0)
	{
		if ($BcpBuffer)
		{
			$TraceModes = 'SRC|LOG';
			Trace("\nBcp not Bind to Buffer!!Waiting...\n",0);
			$TraceModes = "";
			my (@fetchrow,$ary_ref,$rows,$row_ref);
			while(@fetchrow = $substh->fetchrow_array)
			{
				push(@$ary_ref,[@fetchrow]);
				$rows ++;
				if (($rows % 5000) == 0)
				{
					printf ("\n\t%7d of $counter_num rows have finished!",$rows);
				}
			}
			$counter_num = $counte;
			$TraceModes = 'SRC|LOG';	
			Trace("\nBcp not Bind to Table $$subcptab!!Waiting...\n\n",0);
			$TraceModes = "";
			$serial_id = 0;
			foreach $row_ref (@$ary_ref)
			{
				$value_cols = "";
				$counter ++;
				$serial_id ++;
				$value_cols = "\'".$serial_id."\',";
#-------------optimize not use Time (2002-01-01 00:00:00.000=>2002-01-01 00:00:00)------------
				for($i=0;$i<@$row_ref;$i++)
				{
					if ($substh->{TYPE}->[$i] == 11)
					{
						if (@$row_ref[$i] =~ /^\s*(\d{4}-\d{2}-\d{2}\s\d{2}:\d{2}:\d{2}).*$/)
						{
							@$row_ref[$i] = $1;
						}
					}
					$value_cols .= "\'".@$row_ref[$i]."\',";
					
				}
				chomp($value_cols);
				chop($value_cols);
#---------------------------------------------------------------------------------------------
#				$value_cols = join("\',\'",@$row_ref);
#				$value_cols = "\'".$value_cols."\'";
#---------------------------------------------------------------------------------------------				
				$ins_sql = "insert into $$subcptab (".$cols.")values(".$value_cols.");";
#				&Exec_Sql($subcpdbh,\$subcpsth,$ins_sql);
#--------------------------------optimize not use Exec_Sql--------------------------------
				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);
					Trace("\nIgnore...\n",0);
					$TraceModes = 'SRC|LOG';
#					OnError($subdbh);
				}
#--------------------------------------------------------------------------------				
				if (($counter % 1000) == 0)
				{
					printf ("\n\t%7d of $counter_num rows have finished!",$counter);
#--------------------------------------------------------------------------------
					if ($COOKDB_TYPE eq 'ORACLE')
					{
						Commit_DB($subcpdbh);
					}
#--------------------------------------------------------------------------------
				}
			}
			$counter_num = $counter;
			undef($ary_ref);
			$TraceModes = 'SRC|LOG';
		
			Trace("\nBcp Finished! $counter rows!\n",0);
			$substh->finish;			
		}
		else
		{
			$TraceModes = 'SRC|LOG';			
			Trace("\nBcp not Bind To Table $$subcptab!!Waiting...\n\n",0);
			$TraceModes = "";
			my @fetchrow;
			$serial_id = 0;
			while(@fetchrow = $substh->fetchrow_array)
			{
				$value_cols = "";
				$counter ++;
				$serial_id ++;
				$value_cols = "\'".$serial_id."\',";
#-------------optimize not use Time (2002-01-01 00:00:00.000=>2002-01-01 00:00:00)------------
				for($i=0;$i<@fetchrow;$i++)
				{
					if ($substh->{TYPE}->[$i] == 11)
					{
						if (@fetchrow[$i] =~ /^\s*(\d{4}-\d{2}-\d{2}\s\d{2}:\d{2}:\d{2}).*$/)
						{
							@fetchrow[$i] = $1;
						}
					}
					$value_cols .= "\'".@fetchrow[$i]."\',";
					
				}
				chomp($value_cols);
				chop($value_cols);
#---------------------------------------------------------------------------------------------
#				$value_cols = join("\',\'",@fetchrow);
#				$value_cols = "\'".$value_cols."\'";
#---------------------------------------------------------------------------------------------
				$ins_sql = "insert into $$subcptab (".$cols.")values(".$value_cols.");";
#--------------------------------optimize not use Exec_Sql--------------------------------
#				&Exec_Sql($subcpdbh,\$subcpsth,$ins_sql);
#---------------------------------------------------------------------------------------------
				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);
					Trace("\nIgnore...\n",0);
#					OnError($subdbh);
					$TraceModes = 'SRC|LOG';
				}
#--------------------------------------------------------------------------------				

				if (($counter % 1000) == 0)
				{
					printf ("\n\t%7d of $counter_num rows have finished!",$counter);
#--------------------------------------------------------------------------------
					if ($COOKDB_TYPE eq 'ORACLE')
					{
						Commit_DB($subcpdbh);
					}
#--------------------------------------------------------------------------------
				}
			}
			$TraceModes = 'SRC|LOG';
		
			Trace("Bcp Finished! $counter rows!\n",0);
			$substh->finish;
		}
		Commit_DB($subcpdbh);
		return $counter;
	}
#----------------------------INSERT WITH BIND------------------------
	if ($BcpMode == 1)
	{
		$value_cols = "?";
		for($i=0;$i<$colnum;$i++)
		{
			if ($value_cols)
			{
				$value_cols .= ',?';
			}
			else
			{
				$value_cols = '?';
			}
		}
		$ins_sql = "insert into $$subcptab (".$cols.")values(".$value_cols.");";
	
		if(!($subcpsth = $$subcpdbh->prepare($ins_sql)))
		{
			$TraceModes = 'SRC|LOG|REP';
			Trace("Err:Bcp insert bind sql prepare error!\nSQL Statement:$ins_sql\nErr Info:".$DBI::errstr);
			NPM_Send_Message(210);
			NPM_Write_DalLog(ERROR_MESSAGE=>"$DBI::errstr");
			OnError($subcpdbh);
			return -1;
		}
		if ($BcpBuffer)
		{
			$TraceModes = 'SRC|LOG';
			Trace("\nBcp with Bind to Buffer!!Waiting...\n",0);
			$TraceModes = "";
			my (@fetchrow,$ary_ref,$rows,$row_ref);
			while(@fetchrow = $substh->fetchrow_array)
			{
				push(@$ary_ref,[@fetchrow]);
				$rows ++;
				if (($rows % 5000) == 0)
				{
					printf ("\n\t%7d of $counter_num rows have finished!",$rows);
				}
			}
			$counter_num = $counte;
			$TraceModes = 'SRC|LOG';			
			Trace("\nBcp with Bind To Table $$subcptab!!Waiting...\n\n",0);
			$TraceModes = "";
			$serial_id = 0;
			foreach $row_ref (@$ary_ref)
			{
				$value_cols = "";
				$counter ++;
				$serial_id ++;
				for($i=0;$i<@$row_ref;$i++)
				{
					if ($substh->{TYPE}->[$i] == 11)
					{
						if (@$row_ref[$i] =~ /^\s*(\d{4}-\d{2}-\d{2}\s\d{2}:\d{2}:\d{2}).*$/)
						{
							@$row_ref[$i] = $1;
						}
					}
				}
				unshift(@$row_ref,$serial_id);
				if(!($subcpsth -> execute(@$row_ref)))
				{
					$TraceModes = 'SRC|LOG|REP';
					$errows = join(",",@$row_ref);
					Trace("Err:Bcp insert date error!\nInsert values:$errows\nErr Info:".$DBI::errstr);
					Trace("\nIgnore...\n",0);
#					OnError($subdbh);
					$TraceModes = 'SRC|LOG';
				}
			
				if (($counter % 1000) == 0)
				{
					printf ("\n\t%7d of $counter_num rows have finished!",$counter);
#--------------------------------------------------------------------------------
					if ($COOKDB_TYPE eq 'ORACLE')
					{
						Commit_DB($subcpdbh);
					}
#--------------------------------------------------------------------------------
				}
			}
			undef($ary_ref);		
			$TraceModes = 'SRC|LOG';
		
			Trace("Bcp Finished! $counter rows!\n",0);
			$substh->finish;		
		}
		else
		{
			
	
			$TraceModes = 'SRC|LOG';			
			Trace("\nBcp with Bind To Table $$subcptab!!Waiting...\n\n",0);
			$TraceModes = "";	
			my @fetchrow ;	
			$serial_id = 0;
			while(@fetchrow = $substh->fetchrow_array)
			{
				$value_cols = "";
				$counter ++;
				$serial_id ++ ;
				for($i=0;$i<@fetchrow;$i++)
				{
					if ($substh->{TYPE}->[$i] == 11)
					{
						if (@fetchrow[$i] =~ /^\s*(\d{4}-\d{2}-\d{2}\s\d{2}:\d{2}:\d{2}).*$/)
						{
							@fetchrow[$i] = $1;
						}
					}
				}
				unshift(@fetchrow,$serial_id);
				if(!($subcpsth -> execute(@fetchrow)))
				{
					$TraceModes = 'SRC|LOG|REP';
					$errows = join(",",@fetchrow);
					Trace("Err:Bcp insert date error!\nInsert values:$errows\nErr Info:".$DBI::errstr);
					Trace("\nIgnore...\n",0);
#					OnError($subdbh);
					$TraceModes = 'SRC|LOG';
				}
			
				if (($counter % 1000) == 0)
				{
					printf ("\n\t%7d of $counter_num rows have finished!",$counter);
#--------------------------------------------------------------------------------
					if ($COOKDB_TYPE eq 'ORACLE')
					{
						Commit_DB($subcpdbh);
					}
#--------------------------------------------------------------------------------
	
				}
			}
			$TraceModes =  "SRC|LOG";
		
			Trace("Bcp Finished! $counter rows!\n",0);
			$substh->finish;
		}
		Commit_DB($subcpdbh);
		return $counter;
	}
}

###########################################################################
# Function: Exec_Finish
# Description: execute sql statement finish 
# Input:  
#	1.sth pointer
# Output: Null
# Return: Null
###########################################################################
sub Exec_Finish
{
	my ($substh,$subrc);

	$substh = shift;
	if ($DeBug)
	{
		Trace("Execute SQL Statement finished!\n");
	   	Trace("***Programming Debug!! return!!!\n");
	   	return 0;
	} 
	$subrc = $$substh->finish;
	Trace("Execute SQL Statement finished!\n");
}
###########################################################################
# Function: SetError
# Description: execute sql statement 
# Input:  
#	1.$ErrorRollBack	Default 0
#	2.$ErrorCommit		Default 0
#	3.$ErrorExit		Default 0
#	4.$ErrorDebug		Default 1
# Output: Null
# Return: Null
###########################################################################
sub SetError
{
	$ErrorRollBack = shift;
	$ErrorCommit = shift;
	$ErrorExit = shift;
	$ErrorDebug = shift;
}
###########################################################################
# Function: Write_Dchk_Err
# Description: Write_Dchk_Err
# Input:  
#	error info
# Output: Null
# Return: Null
###########################################################################
sub Write_Dchk_Err
{
	my $err_info = shift;
	my ($sub_statement);
	my ($sub_dbh,$sub_sth,$sub_rc,$dc_ctime);
	my ($subconn_dbn,$subconn_dbd,$subconn_uid,$subconn_pwd);

	($subconn_dbd,$subconn_dbn,$subconn_uid,$subconn_pwd)=Get_ODBMS();
	
	if(!($subconn_dbd and $subconn_dbn))
	{
		($subconn_dbd,$subconn_dbn,$subconn_uid,$subconn_pwd)=Get_DBMS();
	}
	if(!($subconn_dbd and $subconn_dbn))
	{
		Trace("Warning: Not get cookdb dbms,so not write data check error information!");
		return 0;
	}

	$dc_ctime = CFtime('YYYY-MM-DD hh:mm:ss');
	Trace("Connecting Database! DBI:$subconn_dbd:$subconn_dbn $subconn_uid/*****\n");
	
	$sub_rc = $sub_dbh = DBI->connect("DBI:$subconn_dbd:$subconn_dbn",$subconn_uid,$subconn_pwd,
			{RaiseError => 0,PrintError => 0, AutoCommit => 1});
	if(!$sub_rc)
	{
		Trace("Err:Connect Database error!\nErr Info:$DBI::errstr\n");
		return 0;
	}
	
	$sub_statement =<<EOF;
INSERT INTO mapi_dchk_err
(
omc_id,			data_collect_time,
scan_start_time,	scan_stop_time,
perf_name,		err_info
)
VALUES(?,?,?,?,?,?);
EOF
	
	$sub_rc = $sub_sth = $sub_dbh->prepare($sub_statement);
	if(!$sub_rc)
	{
		Trace("Err:SQL Prepare error!\nSQL:\n".$sub_statement."Err Info:$DBI::errstr\n");
		$sub_dbh->disconnect;
		return 0;
	}
	$sub_rc = $sub_sth->execute($OMC_ID,$dc_ctime,$PM_DCHK_STIME,$PM_DCHK_ETIME,$PM_DCHK_PNAME,$err_info);
	if(!$sub_rc)
	{
		Trace("Err:Write Data Check Information Error!Err Info:".$DBI::errstr);
		$sub_dbh->disconnect;
		return 0;
	}

	$sub_sth->finish;

	$sub_rc = $sub_dbh->disconnect;
	
	Trace("Write data check information end!\n");
}
###########################################################################
# Function: Write_OlogDB
# Description: Write OlogDb
# Input:  
#	0/1/2	# start_time/end_time/error
# Output: Null
# Return: Null
###########################################################################
sub Write_OlogDB
{
	my $ss_flag = shift;
	my ($sub_statement);
	my ($sub_dbh,$sub_sth,$sub_rc);
	my ($subconn_dbn,$subconn_dbd,$subconn_uid,$s