audcreindex.cpp

查重

/*
 *     文件名： checkDup.cpp
 *
 *    版权所有 (C) 2002 神州数码(中国)有限公司
 *
 *        描述：稽核查重主程序
 *
 *        作者：戴大刚
 *
 *    创建日期：2006-03-14
 *
 *        版本：ver1.0
 *
 *    参考文档：
 *
 *    修订记录：
 *    修订日期    修订人    修订内容
 *                                               
 *                           
 */

// 该模版适合于所有预处理，标准批价，查重，入库，下发处理等模块
// 该类模块特点是输出，输入文件一一对应
// 该类模版并不适合上发，分拣
// 在需要程序数据一致性的地方如查重需要编写redo,beginTran,commitTran函数

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <math.h>
#include <time.h>
#include <unistd.h>
#include <fcntl.h>
#include <malloc.h>
#include <dirent.h>
#include <sys/stat.h>
#include <stdarg.h>

/* 引用公共函数申明的头文件 */
#include "public.h"

/* 引用查重程序相关类型定义  */
#include "audCreIndex.h"

/* 储存已经初始化的索引文件名                         */
/* 在处理一个话单文件结束后需要重新刷新这些文件       */
/* 索引文件最大128天，应该已经足够                    */

typeIndexName  pIndexName[1280];
int  totalIndexFiles;

/* 定义程序运行必选参数 */
char inputPath[128+1];          // 输入路径
char filePattern[128+1];        // 输入文件匹配条件
char outputPath[128+1];         // 输出路径
char tmpPath[128+1];            // 临时路径
char logPath[128+1];            // 日志路径
char indexPath[128+1];          // 日志路径
char bakPath[128+1];            // 备份路径
char errorPath[128+1];          // 错误路径
char staticPath[128+1];         // 静态数据路径
char idxNamePrefix[12+1];       // 索引文件名前缀
char procID[2+1];               // 进程序号
int  runFlag;                   // 程序运行方式
int  sleepTime;                 // 没有文件处理时程序睡眠时间
char orgfileName[128+1];
/* runFlag 定义： 为 1 ，备份文件，守护进程运行       */
/* runFlag 定义： 为 2 ，不备份文件，非守护进程运行   */
/* runFlag 定义： 为 3 ，不备份文件，守护进程运行     */
/* runFlag 定义： 其他 ，备份文件，非守护进程运行     */
/* 下面定义应用程序需要的特殊运行参数                 */
/* 针对每一个应用程序都会不同                         */

char errMsg[128+1];             // 全局的出错信息，用于描述抛出的异常
typeLogBuf    pLogBuf;          // 日志输出相关参数

typedef struct{
char origenName[24+1];             //原始清单文件名
int  totalNum;                     //总话单数
int  curNum;                       //成功话单数 
}typeChkDupLogBuf;
/* 记录对应于原始清单日志                            */
typeChkDupLogBuf pLogCkDpBuf[1024];
int    totalLogBuf;

/* 根据应用程序需要从配置文件里读取相关参数 */ 
int readCfgFile(char * programName,typeCfgParam * pCfgParam)
{
char buf[128+1];
char tmpChar[56+1];

programName[strlen(programName)-4]='\0'; //去除.cfg后缀
if(strstr(programName,"audCreMobIdx")==NULL && strstr(programName,"audCreNetIdx")==NULL)
{
	printf("Cfg file is error! please check it!\n");
	return -1;
}

memset(tmpChar,0,sizeof(tmpChar));
if(strncmp(programName,"../cfg/",7)==0)
{
	memcpy(tmpChar,&programName[7],strlen(programName)-7);
}
else
{
	memcpy(tmpChar,programName,strlen(programName));
}

memset(buf,0,sizeof(buf));
sprintf(buf,"%s.%s",tmpChar,"inputPath");				
memset(inputPath,0,sizeof(inputPath));
if( readCfgParam(inputPath,buf,pCfgParam)<0)   return -1;	

memset(buf,0,sizeof(buf));
sprintf(buf,"%s.%s",tmpChar,"filePattern");					
memset(filePattern,0,sizeof(filePattern));	
if( readCfgParam(filePattern,buf,pCfgParam)<0)     return -1;

memset(buf,0,sizeof(buf));
sprintf(buf,"%s.%s",tmpChar,"outputPath");
memset(outputPath,0,sizeof(outputPath));	
if( readCfgParam(outputPath,buf,pCfgParam)<0) return -1;

memset(buf,0,sizeof(buf));
sprintf(buf,"%s.%s",tmpChar,"tmpPath");	
memset(tmpPath,0,sizeof(tmpPath));	
if( readCfgParam(tmpPath,buf,pCfgParam)<0)       return -1;

memset(buf,0,sizeof(buf));
sprintf(buf,"%s.%s",tmpChar,"logPath");	
memset(logPath,0,sizeof(logPath));	
if( readCfgParam(logPath,buf,pCfgParam)<0)       return -1;

memset(buf,0,sizeof(buf));
sprintf(buf,"%s.%s",tmpChar,"bakPath");
memset(bakPath,0,sizeof(bakPath));	
if( readCfgParam(bakPath,buf,pCfgParam)<0)       return -1;

memset(buf,0,sizeof(buf));
sprintf(buf,"%s.%s",tmpChar,"errorPath");
memset(errorPath,0,sizeof(errorPath));	
if( readCfgParam(errorPath,buf,pCfgParam)<0)   return -1;

memset(buf,0,sizeof(buf));
sprintf(buf,"%s.%s",tmpChar,"indexPath");
memset(indexPath,0,sizeof(indexPath));	
if( readCfgParam(indexPath,buf,pCfgParam)<0)   return -1;

memset(buf,0,sizeof(buf));
sprintf(buf,"%s.%s",tmpChar,"idxNamePrefix");
memset(idxNamePrefix,0,sizeof(idxNamePrefix));	
if( readCfgParam(idxNamePrefix,buf,pCfgParam)<0)   return -1;
rtrim(idxNamePrefix);
idxNamePrefix[3]='\0';

memset(buf,0,sizeof(buf));
sprintf(buf,"%s.%s",tmpChar,"procID");
memset(procID,0,sizeof(procID));	
if( readCfgParam(procID,buf,pCfgParam)<0) return -1;

memset(buf,0,sizeof(buf));
sprintf(buf,"%s.%s",tmpChar,"runFlag");
runFlag = 0;	
if( readCfgParam(&runFlag,buf,pCfgParam)<0)   return -1;

memset(buf,0,sizeof(buf));
sprintf(buf,"%s.%s",tmpChar,"sleepTime");
sleepTime = 30;
if( readCfgParam(&sleepTime,buf,pCfgParam)<0)   return -1;

/* 如果该应用程序需要其他参数配合，在下面添加 */

return 0;
}

/* 程序不正常关闭，在重启后需要作会滚操作*/
void  redoHandler(char * programName)
{
char fullPathName[128+1];
FILE * fp;
char buf[128+1];
char tmpFileName[128+1];
char objPath[128+1];

memset(fullPathName,0,sizeof(fullPathName));
sprintf(fullPathName,"%s/%s.break",tmpPath,programName);

/* 没有断点文件，返回 */
if( access(fullPathName,F_OK) != 0 ) return;
if( (fp=fopen(fullPathName,"r")) == NULL ){
	memset(buf,0,sizeof(buf));
	sprintf(buf,"Can't open break file %s for read!",fullPathName);
	throw (char *)buf;
	}

while(memset(buf,0,sizeof(buf)),fgets(buf,256,fp)!=NULL){
	memset(tmpFileName,0,sizeof(tmpFileName));
	memset(objPath,0,sizeof(objPath));
	getTextField(tmpFileName,buf,',',1);
	getTextField(objPath,buf,',',2);
	if( access(tmpFileName,F_OK) == 0 ) mvfile(tmpFileName,objPath);
	}
fclose(fp);
remove(fullPathName);		
}

/* 在程序需要维护数据一致性的地方开始事务 */
void beginTran(char * programName,char * fileName)
{
FILE * fp;
char fullPathName[128+1];
char newPathName[128+1];
char buf[128+1];
int  k;

memset(fullPathName,0,sizeof(fullPathName));
sprintf(fullPathName,"%s/.%s.break",tmpPath,programName);
if( (fp=fopen(fullPathName,"w")) == NULL ){
	memset(buf,0,sizeof(buf));
	sprintf(buf,"Can't open break file %s for write!",fullPathName);
	throw (char *)buf;
	}

/* 记录输入文件 */
memset(buf,0,sizeof(buf));
sprintf(buf,"%s/%s.lst,%s\n",inputPath,fileName,bakPath);
fputs(buf,fp);	
	
/* 记录输出文件 */
memset(buf,0,sizeof(buf));
sprintf(buf,"%s/%s.chk,%s\n",tmpPath,fileName,outputPath);
fputs(buf,fp);	

/* 记录错误文件 */
memset(buf,0,sizeof(buf));
sprintf(buf,"%s/%s.dup,%s\n",tmpPath,fileName,errorPath);
fputs(buf,fp);	

/* 记录索引文件 */
for(k=0;k<totalIndexFiles;k++){
	memset(buf,0,sizeof(buf));
	sprintf(buf,"%s/%s%s.idx,%s\n",tmpPath,idxNamePrefix,pIndexName[k].callDate,indexPath);
	fputs(buf,fp);
	}	
fclose(fp);

memset(newPathName,0,sizeof(newPathName));
sprintf(newPathName,"%s/%s.break",tmpPath,programName);
rename(fullPathName,newPathName);
}

/* 在程序需要维护数据一致性的地方提交事务 */
void commitTran(char * programName)
{
char fullPathName[128+1];
memset(fullPathName,0,sizeof(fullPathName));
sprintf(fullPathName,"%s/%s.break",tmpPath,programName);
remove(fullPathName);	
}

/* 获取程序调度标志 */
int  getProgramStatu(char * programName,char * tmpPath)
{
char fullPathName[128+1];

memset(fullPathName,0,sizeof(fullPathName));		
sprintf(fullPathName,"%s/%s.stop",tmpPath,programName);
/* 有停止进程标志文件存在，程序需要退出             */
if( access(fullPathName,F_OK) == 0 ){
	remove(fullPathName);
	return 0;  	
	}
return 1;
}



/* 返回值小于0为错误，错误代码范围为301-310   */
/* E301 计费号码msisdn非13开头，不足11位      */
/* E302 重单                                  */
/* E303 时间不合法                            */

/* 追加到日志记录 */
void addToLogBuf(char * origenName,int isDup)
{
/* 对无主清单特殊处理,文件名太多,忽略 */
if( totalLogBuf > 1020 ) return ;
if( strncmp(pLogCkDpBuf[totalLogBuf].origenName,origenName,24) == 0){
	pLogCkDpBuf[totalLogBuf].totalNum ++;
	pLogCkDpBuf[totalLogBuf].curNum   += ( 1 - isDup );	
	}
else {
	if( strlen(pLogCkDpBuf[totalLogBuf].origenName) != 0  ) totalLogBuf++;
	memcpy(pLogCkDpBuf[totalLogBuf].origenName,origenName,24);
	pLogCkDpBuf[totalLogBuf].origenName[24]='\0';
	pLogCkDpBuf[totalLogBuf].totalNum = 1;
	pLogCkDpBuf[totalLogBuf].curNum   = ( 1 - isDup );
	}
//printf("END [%4d][%s]\n",totalLogBuf,pLogBuf[totalLogBuf].origenName);getchar();
}

/* 向索引数据中追加索引              */
/*  1 追加成功，非重单               */
/* -1 追加失败，是重单               */
int addToIndexList(typeMoreKeyList * pMoreKeyList,int pos,char * sKey1,char * sKey2)
{
typeOneKeyList * * p;
int key1,key2;
int k;
int i;

key1 = atoi(sKey1);
key2 = atoi(sKey2);

/* 首先依次循环定长数组 */

//printf("pos =%d,key1=%d,key2=%d",pos,key1,key2);getchar();
for(i=0;i<4*ONEDAYONEUSERCDR;i++){
	if( (pMoreKeyList[pos].pKeyArray)[i].key1 == 0 && 
	    (pMoreKeyList[pos].pKeyArray)[i].key2 == 0 ){
	    	(pMoreKeyList[pos].pKeyArray)[i].key1 = key1; 
		(pMoreKeyList[pos].pKeyArray)[i].key2 = key2;
		return  1;  /* 非重单 */
		}

	if( (pMoreKeyList[pos].pKeyArray)[i].key1 == key1 && 
	    (pMoreKeyList[pos].pKeyArray)[i].key2 == key2 ){
		return -1;
	}
	
	}

/* 其次依次循环动态链表 */	
for(p=&(pMoreKeyList[pos].pKeyList);(*p)!=NULL;p=&((*p)->p)){
	if( (*p)->key1 == key1 && (*p)->key2 == key2 ){
		return -1;  
		}
	}
(*p) = new typeOneKeyList ;
if( (*p) == NULL){
	throw (char *)"new typeOneKeyList fail!" ;
	}

(*p)->key1 = key1;
(*p)->key2 = key2;
(*p)->p    = NULL;
return 1;                  /* 非重单 */
}

/*  根据通话日期初始化索引数据  */
void readIndexFile(char * callDate,char * indexPath,typeMoreKeyList * * pMoreKeyList,char * idxNamePrefix)
{
char fullPathName[128+1];
FILE * fpIdx;
char buf[256+1];
char tmpBuf[16+1];
char msisdn[11+1];
char callTime[6+1];
char sKey1[9+1];
char sKey2[9+1];
int  k;
int  i;
int  pos;

/* 为一天的索引申请内存空间  */
(*pMoreKeyList) = new typeMoreKeyList[10000];
if( pMoreKeyList == NULL ){
	memset(buf,0,sizeof(buf));
	sprintf(buf,"Can't new index pMoreKeyList!\n");
	throw (char *)buf;
	}	
//printf("ONEDAYONEUSERCDR=%d",ONEDAYONEUSERCDR);getchar();
/* 索引关键字全部初始化为0   */
for(k=0;k<10000;k++){
	for(i=0;i<4*ONEDAYONEUSERCDR;i++) {
		((*pMoreKeyList)[k].pKeyArray)[i].key1 = 0;
		((*pMoreKeyList)[k].pKeyArray)[i].key2 = 0;
		}
	(*pMoreKeyList)[k].pKeyList = NULL;
	}

/* 打开索引文件用于读写索引数据  */
memset(fullPathName,0,sizeof(fullPathName));

if ((strncmp(orgfileName,"p",1) == 0) || (strncmp(orgfileName,"m",1) == 0))
	sprintf(fullPathName,"%s/%s%10.10s.idx01",indexPath,idxNamePrefix,callDate);
if ((strncmp(orgfileName,"q",1) == 0) || (strncmp(orgfileName,"w",1) == 0))
	sprintf(fullPathName,"%s/%s%10.10s.idx02",indexPath,idxNamePrefix,callDate);
	
if(access(fullPathName,F_OK) != 0) return;
if( (fpIdx=fopen(fullPathName,"r")) == NULL){
	memset(buf,0,sizeof(buf));
	sprintf(buf,"Can't init index file %s!\n",fullPathName);
	throw (char *)buf;
	}
//printf("fullPathName=%s\n",fullPathName);getchar();
while(memset(buf,0,sizeof(buf)),fgets(buf,256,fpIdx)!=NULL){
	subStrCpy(msisdn,buf,0,9);
	subStrCpy(callTime,buf,10,6);
	subStrCpy(sKey2,buf,17,9);
	
	subStrCpy(tmpBuf,msisdn,5,4);
	pos = atoi(tmpBuf);
	
	memset(sKey1,0,sizeof(sKey1));
	sprintf(sKey1,"%5.5s%4.4s",msisdn,callTime+2);
	
	//printf("%s\n",buf);getchar();
	//printf("Index: %9d,%s,%s\n",pos,sKey1,sKey2);getchar();
	addToIndexList(*pMoreKeyList,pos,sKey1,sKey2);
	//getchar();
	}
fclose(fpIdx);
}