运行数据质量检查规则

#!/usr/bin/perl

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

# 规则检查主要包括以下步骤：

#1.查询出所有的规则组，进行第一次循环

#　 2.查询当前规则组下的所有规则（一般都是针对同一个字段），循环每一个规则，从中提取规则sql

#   3.一组规则的每个规则都会在前一个规则的执行基础之上计算通过率，

#       有效检查记录总数，计算的粒度细分到机构，规则编号

#说明：

#调用形式 ：

#perl ruleCheck-concurrent.pl DQC_BHYWXXB#BHQSRQ_20160601.dir（单个字段）

#perl ruleCheck-concurrent.pl DQC_BHYWXXB#XDHTH-CJRQ_20160601.dir（多个字段）

#执行数据库环境：DB210.5 DPF

# Writer  : zengst

# Version : 2.0

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

use strict;#对语法做严格限制，如变量必须先用my声明

use DBI;

use Time::Local;

 

 

my $dbh;#与数据库的连接

my $AUTO_HOME = $ENV{"AUTO_HOME"};

#通过ETL的加密方式得到登录信息

my $LOGON_DB2_FILE=$ENV{"LOAD_LOGON_DB2_FILE"};

my $DB2LOGON_FILE = "${AUTO_HOME}/etc/${LOGON_DB2_FILE}";

my $ODBCDriver = "AHNXODS64";

my $LOGONSTR;

my $DB2user;

my $DB2pwd;

my $RUN_DATE;#质量检测日期

my $Topic_Area;#检查的专题

my $Sys_Id;#检查的系统id

my $Map_Id;#规则组id

my $table_Name;#检查的表名

my $Check_Column_Name;#检查的字段名称，一个规则组，检测一个字段

my $PK_Column_Name;#检查的表的主键字段

my $Org_Column_Name;#机构字段名称，注意不是值，因为检查的业务系统的机构字段可能不一致

my $Rule_Name;#规则说明，即检查的是什么

my $checkID;  #检查编号

my $SQLClause;#当前规则检查的sql

my $Rule_Info;#规则信息

my $Rule_Group;#规则组信息

my $whereClause;

my $modifydate;

my $CheckDimension;#规则维度

my $CheckSegmentation;

my $logFile;

my $RuleTable='EASTDQM.Rule';#规则信息表

my $RuleGroupTable='EASTDQM.RuleGroup';#规则组信息表

my $check_Result = 'EASTDQM.Check_Result';#轻度结果汇总表

my $check_detail='EASTDQM.Check_Result_Detail';#检查结果明细表

my $check_detail_sample = 'EASTDQM.check_detail_sample';

 

#日志文件目录

#一个大于号，表示覆盖原有文件的内容

#两个大于号表示在原有文件内容之后添加

#@日志文件因为并发，可能发生读写冲突，最好加上表名和字段名的区分，即规则组的粒度

 

 

# 判断是否有信号文件信息

if ( $#ARGV < 0 ) {

  print "\n";

  print "Usage: ****.pl CONTROL_FILE  \n";

  print "Usage: Use parameters \n";

  print "CONTROL_FILE  --  SYS_JOBNAME_YYYYMMDD.dir \n";

  exit(1);

}

 

my $job_params = $ARGV[0]; #传入的参数

my $tableName="";#参数中的表名

my $column="";#参数中的字段名

my $TX_DATE=substr($job_params,length($job_params)-12,8);#业务数据日期

my $_date = substr($TX_DATE,0,4)."-".substr($TX_DATE,4,2)."-".substr($TX_DATE,6,2);

#my $TX_DATE="20160621";#数据日期

my $jobName = substr($job_params,4,length($job_params)-17);

sub getTableAndColumn{

if($jobName =~ /#/){

my @job_split = split("#",$jobName);

$tableName = @job_split[0];

#可能有多个列,列之间通过中杠分割 

$column = @job_split[1];

if(split("-",$column)){

my @columnArr = split("-",$column);

my $columnLength = $#columnArr;

foreach my $k(0..$columnLength){

if($k eq 0){

$column = @columnArr[$k];

}else{

$column = $column.','.@columnArr[$k];

}

}

}

} 

else{

print "The para format is wrong\n";

return 0;

}

}

 

sub getDB2LogonStr{

#定义文件的句柄

my $LOGONFILEREF;

#打开文件

    open($LOGONFILEREF, "${DB2LOGON_FILE}");

    my $LOGONSTR = <$LOGONFILEREF>;

    close(LOGONFILEREF); 

     $LOGONSTR = `${AUTO_HOME}/bin/IceCode.exe "$LOGONSTR"`;

    my ($UserID, $UserPwd) = split(' ',$LOGONSTR);

    chop($UserPwd);  

($DB2user,$DB2pwd) = split(',' , $UserPwd);

}

 

#初始化数据库的连接信息

sub DBconnect{

getDB2LogonStr();

$dbh = DBI->connect("dbi:ODBC:${ODBCDriver}","$DB2user","$DB2pwd",{RaiseError => 1, AutoCommit => 0 })

|| die "Couldn't connect to database: ".DBI->errstr;

print (GetNowTime()." get DB2 connection success! \n ");

return $dbh;

}

 

#得到规则组信息。因为每一个规则组里面的规则需要进行单独的

#权重评分，一次性搜索全部的规则时处理逻辑会相当复杂

sub getRuleGroupInfo{

#得到所有正在运行的规则组

getTableAndColumn();

my $runFreq  =isRunCondition();

   #根据表名+字段名的粒度筛选规则组，并发时并发度做到一定程度的最大化

my $groupInfoSQL = "select Map_Id,Topic_Area from ${RuleGroupTable} where Status= 1 and freq_flag in($runFreq) and table_name='$tableName' and check_column_name='$column' order by Map_Id";

#my $groupInfoSQL = "select Map_Id,Topic_Area from ${RuleGroupTable} where Status= 1 and freq_flag in('01') and table_name='dacorpinfo' and check_column_name='basicbankname' order by Map_Id";

#my $groupInfoSQL = "select Map_Id,Topic_Area from ${RuleGroupTable} where Status= 1 and table_name not in('CZXX','GRDQCKFHZ','GRHQCKFHZMXJL','GRJCXX','JJKXX','JYLS','NBFHZMXJL') AND MAP_ID NOT IN(SELECT MAP_ID FROM $check_Result group by map_id) order by Map_Id";

my $sth = $dbh->prepare($groupInfoSQL) or return -1;

my $ret = $sth->execute() or return -1;  

${Rule_Group} = $sth->fetchall_arrayref();

$sth->finish();

return ${Rule_Group};

}

 

#得到数据库配置表的信息

sub getRuleInfo{

my ($groupId)=@_;#参数传入规则组Map_id

 

#一个规则组下有多个规则，rule_prior小的先执行

my $ruleConfigSql = 

"select check_id,Topic_Area,r.sys_id,r.table_Name,r.Check_Column_Name,".

"r.PK_Column_Name,r.Org_Column_Name,rule_name,modifydate,r.SQLClause,map_id,WhereClause,CHECKDIMENSION ".

"from ${RuleTable} r".

" where r.status='1' and map_id=$groupId ".

" order by Rule_Prior asc";

 

my $sth = $dbh->prepare($ruleConfigSql) or return -1;

my $ret = $sth->execute() or return -1;  

${Rule_Info} = $sth->fetchall_arrayref();

$sth->finish();

return 0;

}

#执行规则SQL

sub exeRuleSql{

 

#存储已经排好序的待执行的规则sql

my @exeCheckSql=();

 

 

for my $i(0..$#${Rule_Info}){

 

$checkID=${Rule_Info}->[$i]->[0];  #检查编号

print (GetNowTime()." Begin To Run Rule[$i]:$checkID \n");

$Topic_Area = ${Rule_Info}->[$i]->[1];#Rule_Info是一个二维数组，${Rule_Info}->[$i]表示是第i行

$Sys_Id=${Rule_Info}->[$i]->[2];

$table_Name=${Rule_Info}->[$i]->[3];

$Check_Column_Name=${Rule_Info}->[$i]->[4];

$PK_Column_Name=${Rule_Info}->[$i]->[5];

$Org_Column_Name=${Rule_Info}->[$i]->[6];

$Rule_Name=${Rule_Info}->[$i]->[7];

$modifydate=${Rule_Info}->[$i]->[8];

$SQLClause=${Rule_Info}->[$i]->[9];

$SQLClause =~ s/(\$[\w_]+)/eval $1/ge;

$SQLClause =~ s/(\$\{[^\}]+\})/eval $1/ge;

$Map_Id=${Rule_Info}->[$i]->[10];

$whereClause = ${Rule_Info}->[$i]->[11];

$CheckDimension = ${Rule_Info}->[$i]->[12];

 

push(@exeCheckSql,$SQLClause);

 

#为了防止多次执行，需要删除轻度汇总表和和检查明细表

#运行时要删除全部的机构数据

 

print (GetNowTime() . " **************************Clear The Result Table:**************************\n");

my $DelSql = "Delete From ${check_Result} Where Check_ID = ${checkID} and rundate='${_date}'";

my $sth = $dbh->prepare($DelSql) or return -1;

#返回影响的行数,select也可以影响行数

my $ret = $sth->execute() or return -1; 

$sth->finish();

print (GetNowTime() . " **************************Clear The Result Table:Succeed[Rows:$ret] **************************\n");

 

 

 

print (GetNowTime() . " **************************Clear The Detail Table:**************************\n");

my $DelSql = "Delete From ${check_detail} Where Check_ID = ${checkID} and RUNDATE='${_date}'";

$sth = $dbh->prepare($DelSql) or return -1;

#返回影响的行数，也包括select语句

$ret = $sth->execute() or return -1;

$sth->finish();#释放结果集

print (GetNowTime() . " **************************Clear The Detail Table:Succeed[Rows:$ret]**************************\n");

 

 

#字符串的相等是eq，第一次执行的规则的记录总数是待检测表的记录总数

if($i==0){

 

my $fromposi = index($SQLClause,'FROM');

my $fromclause = substr($SQLClause,$fromposi+4);

 

#主键字段是多列，这里将它拼成一列

my $pkselect = '';

my @PKcon = split(",",$PK_Column_Name);

my $PKlen = $#PKcon;#获得数组的下标，数组为空时，返回-1

if($PKlen eq 0){

$pkselect = $PK_Column_Name;

}else{

foreach my $k(0..$PKlen){

if($k eq 0){

$pkselect = @PKcon[$k];

}else{

$pkselect = $pkselect.'||'.'\'#\''.'||'.(@PKcon[$k]);

}

}

}

 

#对检查的字段是多列情况的处理,将多列的数据拼接成一列

my $ckselect= ''; 

my @CKcon = split(",",$Check_Column_Name);

my $CKlen = $#CKcon;

if($CKlen eq 0){

$ckselect = $Check_Column_Name;

}else{

foreach my $k(0..$CKlen){

if($k eq 0){

$ckselect = @CKcon[$k];

}else{

$ckselect = $ckselect.'||'.'\'#\''.'||'.(@CKcon[$k]);

}

}

}

 

my $now1 = time;

print (GetNowTime() ." ************************** insert into detail begin **************************\n");

 

#将规则sql的检查结果插入detail表，因为最终需要进行轻度汇总

my $detailinsertSql = "Insert Into ${check_detail}" .

" select ${checkID},'${_date}', ${Map_Id}," . 

" '${Topic_Area}','${Sys_Id}','${table_Name}','${Check_Column_Name}','${CheckDimension}','${CheckSegmentation}',".

" ${pkselect},${Org_Column_Name},${ckselect} from ".$fromclause;

 

my $detailsth = $dbh->prepare($detailinsertSql) or return -1;

$detailsth->execute() or return -1;

$detailsth->finish();

print (GetNowTime() ." ************************** insert into detail end  **************************\n");

 

 

print (GetNowTime() ." ************************** insert into sample data begin **************************\n");

#从全量表按照规则、机构和运行日期的维度取得样本数据，插入到样例表

#partition by 可以是多个字段

my $sampInsert = "insert into ${check_detail_sample}".

" select ${checkID},'${_date}',${Map_Id},'${Topic_Area}','${Sys_Id}','${table_Name}','${Check_Column_Name}',".

"'${CheckDimension}','${CheckSegmentation}',pk_column_value,org_id,check_column_value from ".

"(select crt.*,ROW_NUMBER() over( partition by org_id,check_id) as seq from ${check_detail} crt where rundate='${_date}') detailtmp ".

" where detailtmp.seq<6";

my $inssth = $dbh->prepare($sampInsert) or return -1;

$inssth->execute() or return -1;

$inssth->finish();

#将日志信息打印到文件当中

print (GetNowTime() ." ************************** insert into sample data end  **************************\n");

 

 

print (GetNowTime() ." ************************** begin insert check_result**************************\n");

#将detail数据进行汇总插入到result表,直接通过sql进行数据的循环插入，方式insert into table select ******

my $InsSql = " Insert Into ${check_Result}".

" select ${checkID},'${_date}', '${Rule_Name}',${Map_Id},'${Topic_Area}','${Sys_Id}','${table_Name}','${Check_Column_Name}','${CheckDimension}','${CheckSegmentation}',".

" mainorg ,effectCount,effectCount-decode(FailCount,null,0,FailCount), decode(FailCount,null,0,FailCount),".

" cast(decode(FailCount,null,0,FailCount) as double)/effectCount,cast((effectCount-decode(FailCount,null,0,FailCount)) as double)/effectCount from ".

"(select count(*) as effectCount, $Org_Column_Name as mainorg from  ${Sys_Id}.${table_Name} group by ${Org_Column_Name}) tmain ".

" left join ".

"(select count(*) as FailCount, org_id,check_id from ${check_detail} detail where detail.rundate = '${_date}'and check_id=${checkID} group by org_id,check_id) tdetail".

" on tmain.mainorg = tdetail.org_id";

 

my $collsth = $dbh->prepare($InsSql) or return -1;

$collsth->execute() or return -1;

$collsth->finish();

print (GetNowTime() ." ************************** end insert check_result**************************\n");

my $now2 = time;

my $diff = $now2 - $now1;

if($diff > 1800){#大于30分钟

print ("execute time is : $diff second,checkId is :【$checkID】 ! \n");

}

 

$dbh->commit();

#没有commit时，就不会插入到数据库

 

 

}else{

 

#上一次执行的sql,一个规则组下的第n个规则执行时，检查的有效记录数=全表-前面n-1次检查出的记录数

my $preWhereCluse = '';

my $arrIndex = $#exeCheckSql;

 

#当前是第i个规则sql，那么需要得到前面i-1个规则sql检测的sql

for(my $j=0;$j<=($arrIndex-1);$j++){

 

#这里要求规则sql都是有where条件的,截取的都是where之后的内容

#这里要求执行的规则sql只有where子句

my $whereposi = index(@exeCheckSql[$j],'WHERE');

my $whereclause = substr(@exeCheckSql[$j],$whereposi+5);

 

if($j eq 0){

$preWhereCluse = ' and not('.$whereclause.')';

}else{

$preWhereCluse =  $preWhereCluse.' and not('.$whereclause.')';

}

}

 

#本次检查在上一次检查的基础上，排除上次规则检查有问题的记录

my $currSqlCluse = $SQLClause.$preWhereCluse;

 

my $fromposi = index($currSqlCluse,'FROM');

my $fromclause = substr($currSqlCluse,$fromposi+4);

 

my $pkselect = '';

my @PKcon = split(",",$PK_Column_Name);

my $PKlen = $#PKcon;#获得数组的下标，从0开始，数组为空时，返回-1

if($PKlen eq 0){

$pkselect = $PK_Column_Name;

}else{

foreach my $k(0..$PKlen){

if($k eq 0){

$pkselect = @PKcon[$k];

}else{

$pkselect = $pkselect.'||'.'\'#\''.'||'.(@PKcon[$k]);

}

}

}

 

my $ckselect= ''; 

my @CKcon = split(",",$Check_Column_Name);

my $CKlen = $#CKcon;

if($CKlen eq 0){

$ckselect = $Check_Column_Name;

}else{

foreach my $k(0..$CKlen){

if($k eq 0){

$ckselect = @CKcon[$k];

}else{

$ckselect = $ckselect.'||'.'\'#\''.'||'.(@CKcon[$k]);

}

}

 

}

my $now1 = time;

print (GetNowTime() ." ************************** insert into detail begin **************************\n");

#将规则sql的检查结果插入明细表

my $detailInsertSql = "Insert Into ${check_detail}" .

" select ${checkID},'${_date}', ${Map_Id}," . 

" '${Topic_Area}','${Sys_Id}','${table_Name}','${Check_Column_Name}','${CheckDimension}','${CheckSegmentation}',".

" ${pkselect},${Org_Column_Name},${ckselect} from ".$fromclause;

 

my $detailsth = $dbh->prepare($detailInsertSql) or return -1;

$detailsth->execute() or return -1;

$detailsth->finish();

print (GetNowTime() ." ************************** insert into detail end  **************************\n");

 

 

print (GetNowTime() ." ************************** insert into sample data begin **************************\n");

#从临时表按照规则、机构和运行日期的维度取得样本数据，插入明细表

#partition by 可以是多个字段

my $sampInsert = "insert into ${check_detail_sample}".

" select ${checkID},'${_date}',${Map_Id},'${Topic_Area}','${Sys_Id}','${table_Name}','${Check_Column_Name}',".

"'${CheckDimension}','${CheckSegmentation}',pk_column_value,org_id,check_column_value from ".

"(select crt.*,ROW_NUMBER() over( partition by org_id,check_id) as seq from ${check_detail} crt where rundate='${_date}') detailtmp ".

" where detailtmp.seq<6";

my $inssth = $dbh->prepare($sampInsert) or return -1;

$inssth->execute() or return -1;

$inssth->finish();

print (GetNowTime() ." ************************** insert into sample data end  **************************\n");

 

 

 

print (GetNowTime() ." ************************** begin insert check_result**************************\n");

#将detail数据插入到result表,直接通过sql进行数据的循环插入，方式insert into table select ******

my $InsSql = " Insert Into ${check_Result}".

" select ${checkID},'${_date}', '${Rule_Name}',${Map_Id},'${Topic_Area}','${Sys_Id}','${table_Name}','${Check_Column_Name}','${CheckDimension}','${CheckSegmentation}',".

" mainorg ,effectCount,effectCount-decode(FailCount,null,0,FailCount), decode(FailCount,null,0,FailCount),".

" cast(decode(FailCount,null,0,FailCount) as double)/effectCount,cast((effectCount-decode(FailCount,null,0,FailCount)) as double)/effectCount from ".

"(select count(*) as effectCount, $Org_Column_Name as mainorg from  ${Sys_Id}.${table_Name} where 1=1".$preWhereCluse." group by ${Org_Column_Name}) tmain ".

" left join ".

"(select count(*) as FailCount, org_id,check_id from ${check_detail} detail where detail.rundate = '${_date}'and check_id=${checkID} group by org_id,check_id) tdetail".

" on tmain.mainorg = tdetail.org_id";

 

my $collsth = $dbh->prepare($InsSql) or return -1;

$collsth->execute() or return -1; 

$collsth->finish();

print (GetNowTime() ." ************************** end insert check_result**************************\n");

my $now2 = time;

my $diff = $now2 - $now1;

if($diff > 1800){#大于30分钟

print ("execute time is : $diff second,checkId is :【$checkID】 ! \n");

}

$dbh->commit();

 

}#else block

 

}#rule_info for block

 

return 0;

}

 

sub main{

print "main begin \n";

#得到数据库连接

$dbh = DBconnect();

 

#初始化规则组信息

print (GetNowTime() . " *********************init ruleGroup info ! ***************************\n");

getRuleGroupInfo();

 

#如果查询没有数据，返回接收的数组取值后，下标的最大值就是-1

my $ruleGrouplength = $#${Rule_Group}+1;

 

print (GetNowTime() . " *********************ruleGroup length: $ruleGrouplength ***************************\n");

 

#判断执行结果

if ($ruleGrouplength = 0 ){

print (GetNowTime() . " *********************No Rule_Group Info !*********************\n");

} else {

print (GetNowTime() . " *********************Get Rule_Group:Succeed !********************* \n");

}

 

#循环规则组，根据每一个规则组，找到该规则组下的规则（注意规则状态必须是1，即是可运行的）

my $groupTemp = '';

for my $ii(0..$#${Rule_Group}){

$groupTemp = ${Rule_Group}->[$ii]->[0];

print (GetNowTime() . "********************* Begin To Run Rule_Group[$ii+1],groupId:${groupTemp}********************* \n");

#传入了规则组参数Mapid

print (GetNowTime() . " *********************Begin To get Rule Info! ***************************\n");

getRuleInfo($groupTemp);

print (GetNowTime() . " *********************end  To get Rule Info! ***************************\n");

print (GetNowTime() . " *********************Begin To execute Rule check ! ***************************\n");

exeRuleSql();

 

#将错误明细表的记录导出成txt文件，这里是规则组的粒度

#导出明细后，删除明细数据，防止数据量膨胀

 

print (GetNowTime() . " *********************end To execute Rule check ! ***************************\n");

}

 

print (GetNowTime() . " ********************* end  ***************************\n");

 

return 0;

}

 

sub GetNowTimeNospechar{

my ($sec, $min, $hour, $day, $mon, $year, $wday, $yday, $isdst) = localtime(); 

$hour = sprintf("%02d", $hour);

$min  = sprintf("%02d", $min);

$sec  = sprintf("%02d", $sec);

my $stime = GetNowDate().$hour.$min.$sec;

return $stime;

}

 

sub GetNowTime{

my ($sec, $min, $hour, $day, $mon, $year, $wday, $yday, $isdst) = localtime(); 

$hour = sprintf("%02d", $hour);

$min  = sprintf("%02d", $min);

$sec  = sprintf("%02d", $sec);

my $stime = GetNowDate()." ".$hour.":".$min.":".$sec;

return $stime;

}

 

#找到当前的日期，月份要加1，年份是从1900开始算的

sub GetNowDate{

my ($sec, $min, $hour, $day, $mon, $year, $wday, $yday, $isdst) = localtime(); 

$mon = sprintf("%02d", $mon+1);

$day = sprintf("%02d", $day);

my $stime = ($year+1900)."-".$mon."-".$day;

return $stime;

}

sub GetNowDate8{

my ($sec, $min, $hour, $day, $mon, $year, $wday, $yday, $isdst) = localtime(); 

$mon = sprintf("%02d", $mon+1);

$day = sprintf("%02d", $day);

my $stime = ($year+1900).$mon.$day;

return $stime;

}

 

#判断当前日期所属的频率（每天，每星期，每半月，每月，每季度，每年）

sub isRunCondition{

 

my $FLAG ="'01'";

my $year = substr($TX_DATE,0,4);

my $month = substr($TX_DATE,4,2);

my $day = substr($TX_DATE,6,2);

 

#判断是否闰年

my $isRunYear;

if(($year%4==0 && $year%100!=0) || $year%400==0){

$isRunYear = 1;

} else {

$isRunYear = 0;

} 

  

   #判断TX_DATE所在周的周末

 my $timenum = Time::Local::timelocal(0, 0, 0, $day, $month-1, $year);

 my @timestr = localtime($timenum);

 my $wday = $timestr[6];

 if ($wday==6){

 $FLAG=$FLAG.",'02'";

 } 

 

  #判断TX_DATE所在月的月末

  my $Month_End;

  if($month eq "01" || $month eq "03" || $month eq "05" || $month eq "07" || $month eq "08" || $month eq "10" || $month eq "12"){

  $Month_End = "31";

  } elsif($month eq "04" || $month eq "06" || $month eq "09" || $month eq "11"){

  $Month_End = "30";

  } elsif($month eq "02" && $isRunYear==1) {

  $Month_End = "29";

  } else {

  $Month_End = "28";

  }

 

if($Month_End eq $day){

$FLAG=$FLAG.",'03'";

}

 

#判断TX_DATE所在季的季末

my $Quarter_Month;

my $Quarter_End;

if($month eq "01" or $month eq "02" or $month eq "03") {

$Quarter_Month="03";

}elsif($month eq "04" or $month eq "05" or $month eq "06") {

$Quarter_Month="06";

}elsif($month eq "07" or $month eq "08" or $month eq "09") {

$Quarter_Month="09";

}elsif($month eq "10" or $month eq "11" or $month eq "12") {

$Quarter_Month="12";

}

 

if($Quarter_Month eq "03" || $Quarter_Month eq "12" ){

$Quarter_End ="31";

} elsif($Quarter_Month eq "09" || $Quarter_Month eq "06" ){

$Quarter_End ="30";

}

 

if($Quarter_Month.$Quarter_End eq $month.$day){

$FLAG=$FLAG.",'04'";

}

 

#判断TX_DATE所在半年末

my $Half_Month;

my $Half_End;

if($month eq "01" or $month eq "02" or $month eq "03" or $month eq "04" or $month eq "05" or $month eq "06") {

$Half_Month="06";

}elsif($month eq "07" or $month eq "08" or $month eq "09" or $month eq "10" or $month eq "11" or $month eq "12") {

$Half_Month="12";

}

 

if($Half_Month eq "12" ){

$Half_End ="31";

} elsif($Half_Month eq "06" ){

$Half_End ="30";

}

 

if($Half_Month.$Half_End eq $month.$day){

$FLAG=$FLAG.",'05'";

}

 

#年末

my $Year_End="1231";

if($Year_End eq $month.$day){

$FLAG=$FLAG.",'06'";

}

 

return $FLAG;

}

#方法调用，程序入口

my $rc = main();#函数的入口

 

exit($rc);#这里的exit内容可以被ETL接收到