fc_collector.pm

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

#############################################################

sub read_colcfg{
	my ($pkg,$colcfg)=@_;

	my ($col_inf,$cnt_inf,$cmd_inf,$err_inf,$err_msg);

	if(!(-r $colcfg)){
		$err_msg="Can't read file $colcfg .\n File:".__FILE__."\t Line:".__LINE__."\n";
		return (-1,$err_msg,'','');	
	}

	$col_inf=eval{XMLin("$colcfg")};
	if($@){
		$err_msg="$@ File:".__FILE__."\t Line:".__LINE__."\n";
		return (-1,$err_msg,'','');
	}
	$cnt_inf=$col_inf->{'Connect_Region'};	
	$cmd_inf=$col_inf->{'Command_Region'};
	$err_inf=$col_inf->{'Error_Region'};
	
#	print "Connect Region is:\n",Dumper($cnt_inf),"\n";
#	print "Command Region is:\n",Dumper($cmd_inf),"\n";
	return (1,"Read $colcfg success.\n" ,$cnt_inf,$cmd_inf,$err_inf);
}

##############################################################
#
#  Function:collect_data
#
#  Input:collection config file
#
#  Output:data file collected.
#
#  Return value:1 success, -1 fail.
#
#  Description: called by sub FC_TaskProcessor::execute_workflow.
#		control each step of collection.
#
#
#############################################################

sub collect_data{
	my ($pkg,$colcfg)=@_;

	DBIs::Trace("Begin collecting  data...\n");
	DBIs::Trace("Collector Config_file is: $colcfg\n");

	#-------------------------------------------
	# read collection config file,get connection 
	# information and command information
	#-------------------------------------------
	my($returncode,$return_Msg);

	($returncode,$return_msg,$pkg->{'connect_func'},$pkg->{'command_func'})=creat_method_function();
	if ($return_code == -1) {
		DBIs::Trace("Can't Read Work Method! \n");
		return -1;
	}
	
	#---------------------------------------------
	# Read Col_cfg , Split it to 3 parts
	#---------------------------------------------
	my ($return_code,$return_msg,$cnt_inf,$cmd_inf,$err_inf)=$pkg->read_colcfg($colcfg);
	DBIs::Trace("$return_msg");
	if ($return_code == -1) {
		return $pkg->move_colcfg(-1,$colcfg);
	}
	DBIs::Trace("Finished read col_cfg\n");

	#-------------------------------------------
	# connect net element
	#-------------------------------------------
	($return_code,$return_msg)=$pkg->execute_connect($cnt_inf);
	DBIs::Trace("$return_msg");
	if ($return_code == -1) {
		return $pkg->move_colcfg(-1,$colcfg);
	}
	DBIs::Trace("finished connect\n");

	#-------------------------------------------
	#send command and collect data
	#-------------------------------------------
	($return_code,$return_msg)=$pkg->execute_cmdlist($cmd_inf,$err_inf);
	DBIs::Trace("$return_msg");
	if ($return_code==-1) {
		return $pkg->move_colcfg(-1,$colcfg);
	}

	#-------------------------------------------
	# success.
	#-------------------------------------------
	DBIs::Trace("finish data collection.\n");
	return $pkg->move_colcfg(1,$colcfg);
}

##############################################################
#
#  Function:move_colcfg
#
#  Input:return_code,collection config file
#
#  Output:
#
#  Return value:return_code
#
#  Description: called by sub collect_data.
#		move collection config file($colcfg) to
#		different path according to return_code.
#
#
#############################################################

sub move_colcfg{
	my ($pkg,$return_code,$colcfg)=@_;

	#----------------------------------------
	# get colcfg's upper path.
	#----------------------------------------
	$colcfg=~/(.*)\/new/;
	my $upper_path=$1;

	if($return_code==1){
		DBIs::Trace("mv $colcfg $upper_path/success/ \n");
	        if(!(-d "$upper_path/success/")) {
                        system("mkdir $upper_path/success/");
                }
		system ("mv $colcfg $upper_path/success/");}
	if($return_code==-1){
		DBIs::Trace("mv $colcfg $upper_path/fail/ \n");
                if(!(-d "$upper_path/fail/")) {
                        system("mkdir $upper_path/fail/");
                }
		system ("mv $colcfg $upper_path/fail/");}

	return $return_code;
}
##############################################################
#
#  Function:check_return_code
#
#  Input:$err_inf
#
#  Output:
#
#  Return value:
#
#  Description:
#
#
#############################################################

sub check_return_code{
    my $return_code=shift;
    my $err_full_inf=shift;
    my $err_msg=shift;

    $err_inf=$err_full_inf->{'error'};
    foreach $err_value(values %$err_inf){;
	    if ($err_msg=~/$err_value->{'error_des'}/){;
	        if ($err_value->{'action'} eq "halt"){
	            return -1;
                }
	        return 1;
	     } 
    }
    if ($err_full_inf->{'default'} eq "next")
       { return 1;} 
    return -1; 
}


##############################################################
#
#  Function: WriteHash2Col_cfg
#
#  Input: 
#
#  Output:
#
#  Return value:
#
#  Description: Write Col_config file for Collect
#
##############################################################

sub WriteHash2Col_cfg {
        my ($template_inf,$outputfilename,$task_id) = @_;

        #---------------------------------------
        # Deal With Command Region
        #---------------------------------------

        my $command = $template_inf->{'Command_Region'}{'cmd'};
        delete $template_inf->{'Command_Region'}{'cmd'};

        if (exists $command->{'id'}) {
                delete $command->{'id'};
                $command = {"1"=>$command};
        } elsif (exists $command->{'name'}) {
                delete $command->{'name'};
                $command = {"1"=>$command};
        }
        my $Command_Region = "<Command_Region>\n";
        foreach my $id (keys %$command) {
                my $cmd_line ="\t<cmd id=\"$id\" ";
                my $cmd = $command->{$id};
                foreach my $value (keys %$cmd) {
                        $cmd_line .= "$value=\"".$cmd->{$value}."\" ";
}
                chop($cmd_line);
                $cmd_line .= "/>\n";
                $Command_Region .= $cmd_line;
        }
        $Command_Region .= "</Command_Region>\n";

        #--------------------------------------
        # Deal with Connect & Error Region
        #--------------------------------------

        my $tmpfile ="/tmp/$task_id"."_tmpfile";
        eval{XMLout($template_inf,outputfile=>"$tmpfile",rootname=>"")};
        if ($@) {
                print $@."File ".__FILE__."Line ".__LINE__."\n";
                exit;
        }
        open (READ,$tmpfile) || die "Can't open file: $tmpfile\n";
        my $Other_Region;
        while (<READ>) {
                $Other_Region .= $_;
        }
	#--------------------------------------
        # Write OutPutConfigFile
        #--------------------------------------

        open (WRITE,">$outputfilename") || die "Can't open file: $outputfilename
\n";
        print WRITE "<Collect_Config>\n";
        print WRITE $Command_Region;
        print WRITE $Other_Region;
        print WRITE "</Collect_Config>";

        close (WRITE);
        return ($outputfilename);
}

##############################################################
#
#  Function: Format_HashStruc
#
#  Input:
#
#  Output:
#
#  Return value:
#
#  Description:
#
#############################################################

sub Format_HashStruc {
        my ($pkg,$hash_ref) = @_;
        my $hash;

        if(exists $hash_ref->{'id'}){
		my $key = $hash_ref->{'id'};
                $hash={"$key"=>$hash_ref};
                delete $hash_ref->{'id'};
        }
        else{
                if(exists $hash_ref->{'name'}){
			my $key = $hash_ref->{'name'};
                        $hash={"$key"=>$hash_ref};
                        delete $hash_ref->{'name'};
                }
                else{
                        #-----------------------------------------------
                        # there are more than one steps in the work flow.
                        #-----------------------------------------------
                        $hash=$hash_ref;
                }
        }
        return ($hash);
}


##############################################################
#
#  Function: Data_filters
#
#  Input: (\@remote_file,\@local_file,$task_id,$db_para)
#	$db_para is a hash ref and it has following structure:
#	{	
#		'db_name'=> database name,
#		'server_name'=> db server name,
#		'db_user'=> user name,
#		'db_pwd' => user password
#	}
#
#  Output: (\@remote_file,\@local_file)
#
#  Return value:
#
#  Description: this fuction is used by 2 work type: MMI and FTP
#  in FTP case , $remote_file and $local_file must be defined.
#  in MMI case , the $local_file is null , and $remote_file is vendor_cmd.
#
###############################################################

sub Data_filters {
	my ($pkg,$remote_file,$local_file,$task_id,$db_para) = @_;
	my ($r,$str);
        my $db_name  = $db_para->{'db_name'};
        my $db_server  = $db_para->{'server_name'};
        my $db_usr  = $db_para->{'db_user'};
        my $db_passwd  = $db_para->{'db_pwd'};
	my $connect_str = "dbi:Informix:$db_name"."\@$db_server";
        my $db_hdl = DBI->connect($connect_str, $db_usr, $db_passwd);
	my $lock_mode = "set lock mode to wait 60";
        $db_hdl->do($lock_mode);

        my $sql = "select distinct vendor_data from dal_log where vendor_data= ? and task_id= ? and program_name=\"filecollector\"";
        my $sth =$db_hdl->prepare($sql)|| (($r=$DBI::err)&&($str=$DBI::errstr));
        if ($r) {
                $db_hdl->rollback;
                print " $r \n$str";
                exit(-1);
        }
	my (@remote_tmp,@local_tmp,@new_remotefile,@new_localfile);
	@remote_tmp = @$remote_file;
	if ($local_file) {
		@local_tmp = @$local_file;
	}
        for (my $i=0;$i<@remote_tmp;$i++) {
                my $datafile = $remote_tmp[$i];
                $sth->bind_param(1,$datafile);
                $sth->bind_param(2,$task_id);
                $sth->execute || (($r=$DBI::err) &&($str=$DBI::errstr));
                my @row_ary  = $sth->fetchrow_array;
                if ($row_ary[0]) {
                        push @new_remotefile, $datafile;
			if ($local_file) {
                        	push @new_localfile, $local_tmp[$i];
			}
                }
        }

        $db_hdl->disconnect();

	if ($local_file) {
		return (\@new_remotefile,\@new_localfile);
	}else {
		return (\@new_remotefile);
	}
}

#######################################################################
#
# Function: write_dallog
#
# Input:
#
# Output:
#
# Description:
#
#
#######################################################################

sub write_dallog {
	my ($pkg,$error_message,$col_tag,$message_class) = @_;
	
	my $v2c_map = $pkg->{'Assistant_Var'}{'v2c_map'};
	my $error_hdl = $pkg->{'ErrMessage_hdl'};
	my $db_para = $pkg->{'db_para'};
        my $common_tmp = $error_hdl->{'common'};

        my ($v2c_tmp,$dallog_msg);
        foreach my $id (keys %$v2c_map) {
                my $vendor_data = $v2c_map->{$id}{'vendor_data'};
                if ($col_tag =~ /$vendor_data/i) {
                        $v2c_tmp = $v2c_map->{$id};
                }
        }
	foreach my $key (keys %$common_tmp) {
                $dallog_msg->{$key} = $common_tmp->{$key};
                if (!($dallog_msg->{$key})) {
                        $dallog_msg->{$key} = '';
                }
        }
        foreach my $key (keys %$v2c_tmp) {
                $dallog_msg->{$key} = $v2c_tmp->{$key};
                if (!($dallog_msg->{$key})) {
                        $dallog_msg->{$key} = '';
                }
        }
        $dallog_msg->{'error_message'} = $error_message;
	$dallog_msg->{'message_class'} = $message_class;

        my $db_dbname= $db_para->{'db_name'};
        my $db_server= $db_para->{'server_name'};
        my $db_pwd= $db_para->{'db_pwd'};
        my $db_user= $db_para->{'db_user'};

        my ($return_code,$return_msg) =$error_hdl->connect_DB($db_type,$db_server,$db_dbname,$db_user,$db_pwd);
        if ($return_code < 0) {
                DBIs::Trace("Can't Connect to DB Write Dal_log !!\n$return_msg");
                return $return_code;
        }

        ($return_code,$return_msg) = $error_hdl->WritelogTbl($dallog_msg);
        if ($return_code < 0) {
                DBIs::Trace("Can't Write Dal_log !!\n$return_msg");
                return $return_code;
        }
        my $db_hdl = $error_hdl->{'db_handle'};
        $db_hdl->disconnect();
        undef($db_hdl);
        delete $error_hdl->{'db_handle'};
        return(1);
}

1;