#include "csta.h"

/******************************************************/
/******            Heartbeat Parameter           ******/
/******************************************************/
DWORD PROC_HEARTBEAT_OID[12]={1,3,6,1,4,1,1373,2,4,10,12};
int PROC_HEARTBEAT_OIDLEN=11;
unsigned long heartbeat_timeout=0;
int sendHeartbeat();

int getOmcSysNo(int* sysNo);
int getOmcStatus(int sysNo);
int omcSysNo=-1;

static char cstaproc_sqlstr[8192];

/*
return:
	TIMETO_NONE
	TIMETO_GET_CSTA
	TIMETO_OUTPUT_FILE
	TIMETO_CHANGE_LOG
*/
int checkTime();
/*
return:
	0-95
*/
int getInstance();

/*
return:
	-1:fail
	0-95:success
*/
int getLastInstance(csta_src *pcsta,csta_sys *psys);

/*
return:
	0:fail
	1:success
*/

void csta_log(const char *fmt, ...);

int cstasrc_init();
void csta_receive();
void csta_request(csta_src *pcsta,csta_sys *psys);
void csta_analyze(snmp_pdu *pdu);
int csta_send(csta_src *pcsta,csta_sys *psys);
int csta_record(csta_src *pcsta,csta_sys *psys,snmp_pdu *pdu);
int csta_record_multi(csta_src *pcsta,csta_sys *psys);
int csta_create_csv();

int parse_map_table(csta_obj *pobj);
void getSystemIP();
void update_system();

BYTE str2oid(char *str, DWORD * array, BYTE max);
int parseLine(char *str, char (*row)[32], int maxrow);

/* Global variable */
int change_log_flag=1;
int output_file_flag=1;
int csta_flag[96];
FILE *logfp=NULL;
MYSQL *pCstaConn;
MYSQL *pubConn;
csta_src CstaSet[MAX_CSTA_SRC_NUM];
int csta_src_num=0;
int global_instance=0;
int csv_src_index=0;
char csv_sent_dir[128];
char csv_sending_dir[128];
int csv_send_flag=0;

// csta keep days setting, by simon begin ---
int max_keep_days = RECORD_KEEP_DAY;
// --- end

int csta_init()
{
	csta_log("cstaCollector starting ...");
	
	pCstaConn=mysql_conn("localhost",CSTA_DB);
	if(pCstaConn == NULL)
	{
		csta_log("Connect to %s fail",CSTA_DB);
		return 0;
	}
	
	pubConn=mysql_conn("localhost","OMC_PUB");
	if(pubConn == NULL)
	{
		csta_log("Connect to OMC_PUB fail");
		return 0;
	}
	
	getOmcSysNo(&omcSysNo);
	if(omcSysNo<0 || omcSysNo >1)
		printf("Fail to get the omcSysNo\n");
	
	cstasrc_init();
	return 1;
}

void csta_timer()
{
	static int update_system_timeout=0;
	csta_src *pcsta;
	csta_sys *psys;
	csta_obj *pobj;
	int i,j;
	int omcStatus;
	
	if(time(NULL) - heartbeat_timeout > 5)
	{
		heartbeat_timeout=time(NULL);
		sendHeartbeat();
	}
	
	omcStatus=getOmcStatus(omcSysNo);
	
	//Master or single
	if(omcStatus == 5 || omcStatus == 7)
	{
		if(update_system_timeout-- < 0)
		{
			update_system_timeout=UPDATE_SYSTEM_INTERVAL;
			update_system();
			
			if(mysql_ping(pCstaConn))
			{
				csta_log("[ERR 2006][csta_timer]:mysql CSTA_DB has gone away,reconnect");
				mysql_close(pCstaConn);
				pCstaConn=mysql_conn("localhost","CSTA_DB");
				if(pCstaConn == NULL)
				{
				  csta_log("[ERR][csta_timer]:reconnect CSTA_DB");
				}
			}
		}
	
		switch(checkTime())
		{
		case TIMETO_GET_CSTA:
			for(i=0;i<csta_src_num;i++)
			{
				pcsta=&CstaSet[i];
				for(j=0;j<pcsta->systemCount;j++)
				{
					if(pcsta->systems[j].system_state == STATE_NORMAL && pcsta->systems[j].msg_state != STATE_MSG_WAIT_RESPONSE)
					{
						pcsta->systems[j].msg_state=STATE_MSG_GET_REQUEST;
						pcsta->systems[j].instance=getInstance();
						pcsta->systems[j].objindex=0;
						pcsta->systems[j].id1=pcsta->objects[0].id1_start;
						pcsta->systems[j].id2=pcsta->objects[0].id2_start;
						csta_log("sysTypeNo=%d,start to get instance=%d csta data!",pcsta->sysTypeNo,pcsta->systems[j].instance);
					}
				}
			}
			break;
		case TIMETO_OUTPUT_FILE:
			csv_src_index=0;
			csta_log("It is time to create csv file");
			break;
		case TIMETO_CHANGE_LOG:
			//cstalog_init();
			break;
		case TIMETO_NONE:
		default:
			break;
		}
		csta_create_csv();
		csta_receive();
		for(i=0;i<csta_src_num;i++)
		{
			pcsta=&CstaSet[i];
			for(j=0;j<pcsta->systemCount;j++)
			{
				psys=&pcsta->systems[j];
				switch(psys->msg_state)
				{
				case STATE_MSG_GET_REQUEST:
					csta_request(pcsta,psys);
					break;
				case STATE_MSG_WAIT_RESPONSE:
					//The csta_analyze function will change the state to STATE_MSG_GET_REQUEST
					psys->fail++;
					if(psys->fail > 50*10)
					{
						psys->fail=0;
						psys->retry++;
						
						psys->instance=psys->pre_instance;
						psys->objindex=psys->pre_objindex;
						psys->id1=psys->pre_id1;
						psys->id2=psys->pre_id2;
						psys->msg_state=STATE_MSG_GET_REQUEST;
	#if CSTA_DEBUG
						csta_log("Get response timeout,sysTypeNo=%d,instance=%d,objindex=%d,id1=%d,id2=%d,retry=%d",pcsta->sysTypeNo,psys->instance,psys->objindex,psys->id1,psys->id2,psys->retry);
	#endif
					}
					
					if(psys->retry > 2)
					{
	#if CSTA_DEBUG
						csta_log("Get response timeout and retry %d time, ignore it",psys->retry);
	#endif
						psys->retry=0;
						
						if(psys->system_state == STATE_CHASE)
						{
							psys->msg_state=STATE_MSG_GET_REQUEST;
							pobj=&pcsta->objects[psys->objindex];
							if(pobj->csta_type == 0)
							{
								psys->objindex++;
								if(psys->objindex == pcsta->objectCount)
								{
									psys->objindex=0;
									psys->instance++;
									psys->instance%=96;
								}
							}
							else if(pobj->csta_type == 1)
							{
								psys->id1++;
								if(psys->id1 == pobj->id1_end)
								{
									psys->id1=0;
									psys->objindex++;
									if(psys->objindex == pcsta->objectCount)
									{
										psys->objindex=0;
										psys->instance++;
										psys->instance%=96;
									}
								}
							}
							else
							{
								psys->id2++;
								if(psys->id2 == pobj->id2_end)
								{
									psys->id2=0;
									psys->id1++;
									if(psys->id1 == pobj->id1_end)
									{
										psys->id1=0;
										psys->objindex++;
										if(psys->objindex == pcsta->objectCount)
										{
											psys->objindex=0;
											psys->instance++;
											psys->instance%=96;
										}
									}
								}
							}
							
							if(psys->instance == global_instance)
							{
								psys->system_state=STATE_NORMAL;
								psys->msg_state=STATE_MSG_NONE;
							}
						}
						else
						{
							psys->msg_state=STATE_MSG_NONE;
						}
					}
					break;
				case STATE_MSG_NONE:
				default:
					break;
				}
			}
		}
	}
}

int getOmcSysNo(int* sysNo)
{
	char fileName[512];
	unsigned char buf[4096];
	unsigned char param[32][512];
	char paramName[128];
	char paramValue[384];
	char* temp;
	FILE* fp;
	int fi;
	int i;
	//int j;
	int paramCount;
	int byteCount;
	
	paramCount=0;
	byteCount=0;
	memset(buf,0,4096);
	memset(param,0,32*512);
	sprintf(fileName,"./conf/omcd.conf");
	fp=fopen(fileName,"r");
	if(fp)
	{
		fi=fileno(fp);
		if(read(fi,buf,4096))
		{
			for(i=0;i<4096;i++)
			{	
				if(buf[i] == 0x0A)
				{
					//printf("\n");
					if(byteCount<512 && param[paramCount][0] != 0x00 && param[paramCount][0] != 0x23)
						paramCount++;
					else
						memset(param[paramCount],0,512);	
					byteCount=0;
				}	
				else
				{	
					param[paramCount][byteCount]=buf[i];
					byteCount++;
				}
				
				//printf("%02x ",buf[i]);
			}
			
			for(i=0;i<paramCount;i++)
			{
				temp=strtok((char*)param[i],"=");
				sprintf(paramName,"%s",temp);
				//printf("Name=%s,",temp);
				temp=strtok(NULL,"=");
				sprintf(paramValue,"%s",temp);
				//printf("Value=%s\n",temp);
				
				
				if(strcasecmp(paramName,"omcSysNo") == 0)
				{
					*sysNo=atoi(paramValue);
				}	
			}
		}	
		close(fi);
		fclose(fp);
	}	
	else
	{
		exitLog("cdrCollector:getOmcSysNo, Can not open the omcd.conf");
		exit(1);
	}	
	return 1;
}

int getOmcStatus(int sysNo)
{
	MYSQL_RES *res;
	MYSQL_ROW row;
	char sqlstr[256];
	int status;
	char binCode[256];
	
	sprintf(sqlstr,
         "SELECT detailStat FROM OMC_PUB.sysStat WHERE sysTypeNo=0 and sysNo=%d ",sysNo);

	res=mysql_getres(pubConn,sqlstr);
	if(res == NULL)
	{
		csta_log("[ERR %d][getOmcStatus]:%s\n",mysql_errno(pubConn),sqlstr);
		return -1;
	}
	
	if((row = mysql_fetch_row(res)) != NULL){
		sprintf(binCode,"%s",row[0]);
		//printf("52-%c,53-%c\n",binCode[52],binCode[53]);
		status=(binCode[52]-0x30)*16+(binCode[53]-0x30);
	}
	mysql_free_result(res);
	return status;
}

int sendHeartbeat()
{
	snmp_pdu pdu;
	snmp_addr addr;
	char proc_name[16];
	
	memset(proc_name,0,16);
	sprintf(proc_name,"cstaCollector");
	
	//pdu
	memset(&pdu,0,sizeof(snmp_pdu));
	pdu.pdu_type=7;
	pdu.request_id=0;
	sprintf(pdu.community,"public");
	pdu.var_num=1;
	pdu.error_status=0;
	pdu.error_index=0;
	
	//Addr
	addr.local_port = 4957;
	//addr.remote_ip = inet_addr("127.0.0.1");
	addr.remote_ip = GetLocalIP();
	addr.broadcast = 0;
	addr.remote_port = 4957;
	
	//Varlist
	memcpy(pdu.var[0].oid,PROC_HEARTBEAT_OID,PROC_HEARTBEAT_OIDLEN*sizeof(DWORD));

	pdu.var[0].oidlen=PROC_HEARTBEAT_OIDLEN;
	pdu.var[0].vartype=0x04;
	pdu.var[0].msglen=16;
	memcpy(pdu.var[0].msg,proc_name,16);
	
	if(snmp_send(&pdu, &addr) <= 0)
	{
		csta_log("Send heartbeat fail");
		return 0;
	}
	return 1;
}

int checkTime()
{
	time_t l_time;
	struct tm *t;
	int hour,minute;
	int instance;
	
	l_time=time(NULL);
	t=localtime(&l_time);
	hour=t->tm_hour;
	minute=t->tm_min;
	
	//for changing log file
	if(change_log_flag == 0 && hour == 23 && minute == 0)
		change_log_flag=1;
	if(change_log_flag && hour == 0 && minute == 0)
	{
		change_log_flag=0;
		return TIMETO_CHANGE_LOG;
	}

	//for output file
	if(output_file_flag == 0 && hour == OUTPUT_FILE_TIME_HOUR && minute == (OUTPUT_FILE_TIME_MINUTE-15))
		output_file_flag=1;
	if(output_file_flag == 1 && hour == OUTPUT_FILE_TIME_HOUR && minute == OUTPUT_FILE_TIME_MINUTE)
	{
		output_file_flag=0;
		return TIMETO_OUTPUT_FILE;
	}

	// fj for debug
	//for get csta
	#if 0
	// 一分钟取一次数据
	if(((minute%1) == 1) || minute == 5 || minute == 20 || minute == 35 || minute == 50)
	#else
	// 15分钟取一次数据
	if( minute == 5 || minute == 20 || minute == 35 || minute == 50)
	#endif
	{
		instance=(hour*4 + minute/15 - 1 + 96)%96;
		if(csta_flag[instance] == 0)
		{
			csta_flag[instance]=1;
			csta_flag[(instance - 1 + 96)%96]=0;
			global_instance++;
			global_instance=global_instance%96;
			return TIMETO_GET_CSTA;
		}
	}
	
	return TIMETO_NONE;
}

int getInstance()
{
	long l_time;
	struct tm *t;
	int hour,minute;
	
	l_time=time(NULL);
	t=localtime(&l_time);
	hour=t->tm_hour;
	minute=t->tm_min;
	
	return (hour*4 + minute/15 - 1 + 96)%96;
}

int getLastInstance(csta_src *pcsta,csta_sys *psys)
{
	char sqlstr[1024];
	MYSQL_RES *res_set;
	MYSQL_ROW row;
	csta_obj *pobj;
	time_t l_time;
	struct tm *t;
	time_t now_time;
	
	pobj=&pcsta->objects[0];
	sprintf(sqlstr,"select max(UNIX_TIMESTAMP(csta_datetime)) from %s.%sData ",CSTA_DB,pobj->detail_table);
	
	res_set=mysql_getres(pCstaConn,sqlstr);
	if(res_set == NULL)
	{
		csta_log("Get last instance from %sData fail,res_set is null",pobj->detail_table);
		return -1;
	}
	
	if(mysql_num_rows(res_set) != 1)
	{
		csta_log("There is no last instance in the table of system %d",pcsta->sysTypeNo);
		mysql_free_result(res_set);
		return -1;
	}
	
	row=mysql_fetch_row(res_set);
	if(row == NULL)
	{
		csta_log("Get last instance from %sData fail,row is null",pobj->detail_table);
		mysql_free_result(res_set);
		return -1;
	}
	mysql_free_result(res_set);
	
	if(row[0] == NULL || strlen(row[0]) == 0)
		l_time=0;
	else
		l_time=(time_t)(strtoul(row[0],NULL,10));
	t=localtime(&l_time);
	
	now_time=time(NULL);
	if(now_time < l_time)
	{
		csta_log("Get last instance from %sData fail,time is invalid",pobj->detail_table);
		return -1;
	}
	if((now_time - l_time) > 24 * 60 * 60)
	{
		csta_log("Get last instance from %sData fail,ems may be stopped for one more day",pobj->detail_table);
		return -1;
	}
	
	return t->tm_hour*4 + t->tm_min/15;
}

int cstasrc_init()
{
	char sqlstr[8192];
	char tmpstr[256];
	MYSQL_RES *res_set;
	MYSQL_ROW row;
	int index;
	csta_src *pcsta;
	csta_obj *pobj;
	int i;
	
	long l_time;
	struct tm *t;
	int hour,minute;
	
	csta_log("Load csta source");
	memset(csta_flag,0,sizeof(csta_flag));
	
	l_time=time(NULL);
	t=localtime(&l_time);
	hour=t->tm_hour;
	minute=t->tm_min;
	
	if(minute == 5 || minute == 20 || minute == 35 || minute == 50)
		global_instance=((hour*60+minute-5+24*60)%(24*60))/15-2+96;
	else
		global_instance=((hour*60+minute-5+24*60)%(24*60))/15-1+96;
	global_instance%=96;
		
	sprintf(sqlstr,
			"SELECT sysTypeNo,recordTable,sumTable,sumUp,hasSubSys,width,isFullRow,checkTime,useAgent,csvPath,detailToSum FROM %s.cstaDefine WHERE isEnable > '0' GROUP BY sysTypeNo,isFullRow",
			CSTA_DB);
	
	res_set=mysql_getres(pCstaConn,sqlstr);
	if(res_set == NULL)
	{
		csta_log("Load csta source fail");
		return 0;
	}
	
	while ((row = mysql_fetch_row(res_set)) != NULL) 
	{
		CstaSet[csta_src_num].sysTypeNo=atoi(row[0]);
		CstaSet[csta_src_num].hassubsys=atoi(row[4]);
		CstaSet[csta_src_num].sum_flag=atoi(row[3]);
		CstaSet[csta_src_num].check_time_flag=atoi(row[7]);
		CstaSet[csta_src_num].single_message_flag=atoi(row[6]);
		CstaSet[csta_src_num].systemCount=0;
		CstaSet[csta_src_num].objectCount=0;
		sprintf(CstaSet[csta_src_num].csv_path,"%s",row[9]);
		
		csta_src_num++;
		if(csta_src_num >= MAX_CSTA_SRC_NUM)
		{
			csta_log("The csta source is greater than %d\n",MAX_CSTA_SRC_NUM);
			break;
		}
	}
	mysql_free_result(res_set);
	
	for(index=0;index<csta_src_num;index++)
	{
		pcsta=&CstaSet[index];
		
		sprintf(sqlstr,"select objIndex,oid,recordTable,sumTable,cstaType,width,id1Start,id1End,id1Offset,id1Field,id2Start,id2End,id2Offset,id2Field,fieldList,detailToSum,csvPath from %s.cstaDefine where sysTypeNo='%d' and isEnable > 0 order by objIndex",CSTA_DB,pcsta->sysTypeNo);
		
		res_set=mysql_getres(pCstaConn,sqlstr);
		while ((row = mysql_fetch_row(res_set)) != NULL)
		{
			pobj=&pcsta->objects[pcsta->objectCount];
			pobj->csta_type=atoi(row[4]);
			pobj->field_width=atoi(row[5]);
			pobj->oidlen=str2oid(row[1],pobj->oid,OID_MAXLEN);
			pobj->id1_start=atoi(row[6]);
			pobj->id1_end=atoi(row[7]);
			//csta_log("id1_end:%d",pobj->id1_end);
			pobj->id1_offset=atoi(row[8]);
			sprintf(pobj->id1_field,"%s",row[9]);
			pobj->id2_start=atoi(row[10]);
			pobj->id2_end=atoi(row[11]);
			pobj->id2_offset=atoi(row[12]);
			sprintf(pobj->id2_field,"%s",row[13]);
			sprintf(pobj->sql_detail,"%s",row[14]);
			pobj->detail_field_count=parseLine(row[14],pobj->detail_field_list,MAX_FIELD_NUM);
			sprintf(pobj->detail_table,"%s",row[2]);
			sprintf(pobj->sum_table,"%s",row[3]);
			//map table
			if(row[15] != NULL && strlen(row[15])>0 && row[15][0] != ' ')
			{
				sprintf(pobj->relate_table,"%s",row[15]);
				if(!parse_map_table(pobj))
					csta_log("Parse map table fail,sysTypeNo=%d",pcsta->sysTypeNo);
			}
			else
			{
				if(pcsta->sum_flag)
				{
					pobj->sum_field_count=pobj->detail_field_count;
					memset(pobj->sum_field_list,0,sizeof(pobj->sum_field_list));
					memcpy(pobj->sum_field_list,pobj->detail_field_list,sizeof(pobj->detail_field_list));
					memset(pobj->sql_sum,0,sizeof(pobj->sql_sum));
					for(i=0;i<pobj->sum_field_count;i++)
					{
						sprintf(tmpstr,"sum(%s) as %s",pobj->sum_field_list[i],pobj->sum_field_list[i]);
						if(i != (pobj->sum_field_count-1))
							strcat(tmpstr,",");
						strcat(pobj->sql_sum,tmpstr);
					}
					sprintf(pobj->relate_table,"NoRelation");
				}
			}
			pcsta->objectCount++;
		}
		mysql_free_result(res_set);
	}
	
	//Initial the csv src index
	csv_src_index=csta_src_num;
	
	//Get csv_sending_dir
	sprintf(sqlstr,"SELECT pubVarValue FROM %s.cstaPubVarConf WHERE pubVarName='sending_directory' ",CSTA_DB);
	res_set=mysql_getres(pCstaConn,sqlstr);
	if(res_set == NULL)
	{
		csta_log("Load sending directory fail, set it to /tmp/");
		sprintf(csv_sending_dir,"/tmp/");
	}
	else
	{
		row=mysql_fetch_row(res_set);
		if(row == NULL || strlen(row[0]) == 0)
			sprintf(csv_sending_dir,"/tmp/");
		else
			sprintf(csv_sending_dir,"%s",row[0]);
		mysql_free_result(res_set);
	}
	csta_log("CSV sending direcory=%s",csv_sending_dir);
	
	//Get csv_sent_dir
	sprintf(sqlstr,"SELECT pubVarValue FROM %s.cstaPubVarConf WHERE pubVarName='sent_directory' ",CSTA_DB);
	res_set=mysql_getres(pCstaConn,sqlstr);
	if(res_set == NULL)
	{
		csta_log("Load sent directory fail, set it to /tmp/");
		sprintf(csv_sent_dir,"/tmp/");
	}
	else
	{
		row=mysql_fetch_row(res_set);
		if(row == NULL || strlen(row[0]) == 0)
			sprintf(csv_sent_dir,"/tmp/");
		else
			sprintf(csv_sent_dir,"%s",row[0]);
		mysql_free_result(res_set);
	}
	csta_log("CSV sent direcory=%s",csv_sent_dir);

	// Get max_keep_days
	sprintf(sqlstr, "SELECT pubVarValue FROM %s.cstaPubVarConf WHERE pubVarName='max_keep_days' ", CSTA_DB);
	res_set = mysql_getres(pCstaConn, sqlstr);
	if (res_set == NULL)
	{
		csta_log("max_keep_days, set default: %d", max_keep_days);
	}
	else
	{
		row = mysql_fetch_row(res_set);
		if (row != NULL && strlen(row[0]) != 0)
		{
			max_keep_days = atoi(row[0]);
		}
		mysql_free_result(res_set);
	}
	csta_log("max_keep_days=%d", max_keep_days);

	csv_send_flag=0;
	sprintf(sqlstr,"SELECT pubVarValue FROM OMC_PUB.omcPubVarConf WHERE pubVarName='csta_store_server'");
	res_set=mysql_getres(pCstaConn,sqlstr);
	if(res_set == NULL)
	{
		csv_send_flag=0;
	}
	else
	{
		if((row=mysql_fetch_row(res_set)) != NULL) 
		{
			if(strlen(row[0]) == 0)
			{	
				csv_send_flag=0;
			}	
			else
			{	
				csv_send_flag=1;
			}
		}
		mysql_free_result(res_set);
	}
	csta_log("CSV send flag=%d",csv_send_flag);
	
	return 1;
}

int parse_map_table(csta_obj *pobj)
{
	char sqlstr[8192];
	char tmpstr2[512];
	char tmplist[8192];
	MYSQL_RES *res_set;
	MYSQL_ROW row;
	int i;
	
	if(strcmp(pobj->relate_table,"NoRelation") == 0)
		return 0;
	
	memset(sqlstr,0,sizeof(sqlstr));
	memset(tmpstr2,0,sizeof(tmpstr2));
	memset(tmplist,0,sizeof(tmplist));
	
	sprintf(sqlstr,"select distinct sumFieldList from %s.%s ",CSTA_DB,pobj->relate_table);
	res_set=mysql_getres(pCstaConn,sqlstr);
	if(res_set == NULL)
		return 0;
		
	pobj->sum_field_count=0;
	while ((row = mysql_fetch_row(res_set)) != NULL)
	{
		sprintf(pobj->sum_field_list[pobj->sum_field_count],"%s",row[0]);
		pobj->sum_field_count++;
	}
	mysql_free_result(res_set);
	memset(pobj->sql_sum,0,sizeof(pobj->sql_sum));
	for(i=0;i<pobj->sum_field_count;i++)
	{
		sprintf(sqlstr,"select detailFieldList from %s.%s where sumFieldList='%s' ",CSTA_DB,pobj->relate_table,pobj->sum_field_list[i]);
		res_set=mysql_getres(pCstaConn,sqlstr);
		if(res_set == NULL)
			return 0;

		memset(tmplist,0,sizeof(tmplist));
		while ((row = mysql_fetch_row(res_set)) != NULL)
		{
			sprintf(tmpstr2,"sum(%s)+",row[0]);
			strcat(tmplist,tmpstr2);
		}
		if(tmplist[strlen(tmplist)-1] == '+')
			tmplist[strlen(tmplist)-1]='\0';
		sprintf(tmpstr2," as %s",pobj->sum_field_list[i]);
		strcat(tmplist,tmpstr2);
		if(i != (pobj->sum_field_count-1))
			strcat(tmplist,",");
		strcat(pobj->sql_sum,tmplist);
		mysql_free_result(res_set);
	}
	return 1;
}

void update_system()
{
	char sqlstr[512];
	MYSQL_RES *res_set;
	MYSQL_ROW row;
	csta_src *pcsta;
	csta_sys *psys;
	csta_obj *pobj;
	int i;
	int last_instance;
	
	for(i=0;i<csta_src_num;i++)
	{
		pcsta=&CstaSet[i];
		pcsta->systemCount=0;
		sprintf(sqlstr,
				"SELECT sysNo,subSysNo,ip,isRemote,netId,DPC FROM OMC_PUB.sysInfo WHERE sysTypeNo=%d ORDER BY sysNo,subSysNo",
				pcsta->sysTypeNo);
		res_set=mysql_getres(pCstaConn,sqlstr);
		if(res_set == NULL)
			return;
		while ((row = mysql_fetch_row(res_set)) != NULL)
		{
			psys=&pcsta->systems[pcsta->systemCount++];
			if(psys->addr.remote_ip == inet_addr(row[2]))
			{	
				continue;
			}
			
			psys->sysid=atoi(row[0]);
			psys->subsysid=atoi(row[1]);
			psys->addr.remote_ip=inet_addr(row[2]);
							
			psys->addr.local_port=CSTA_PORT;
			psys->addr.remote_port=SNMP_PORT;
			psys->objindex=0;
			psys->msglen=0;
			pobj=&pcsta->objects[0];
			if(pobj->csta_type > 0)
				psys->id1=pobj->id1_start;
			if(pobj->csta_type == 2)
			{
				psys->id2=pobj->id2_start;
			}
			
			if(pcsta->check_time_flag)
			{
				last_instance=getLastInstance(pcsta,psys);
				if(last_instance == -1)
					psys->instance=(global_instance+2)%96;
				else
					psys->instance=last_instance;
					
				if(last_instance == global_instance)
				{
					psys->msg_state=STATE_MSG_NONE;
					psys->system_state=STATE_NORMAL;
				}
				else
				{
					psys->timeout=CHASE_INTERVAL;
					psys->msg_state=STATE_MSG_GET_REQUEST;
					psys->system_state=STATE_CHASE;
				}
			}
			else
			{
				psys->instance=global_instance;
				psys->timeout=0;
				psys->msg_state=STATE_MSG_NONE;
				psys->system_state=STATE_NORMAL;
			}
			csta_log("Connect to system,sysTypeNo=%d,sysid=%d,subsysid=%d,ip=%s,instance=%d",pcsta->sysTypeNo,psys->sysid,psys->subsysid,row[2],psys->instance);
		}
		mysql_free_result(res_set);
	}
}

void csta_request(csta_src *pcsta,csta_sys *psys)
{
	csta_obj *pobj,*pobj_n;
	
	pobj=&pcsta->objects[psys->objindex];
	psys->pre_objindex=psys->objindex;
	psys->pre_instance=psys->instance;
	psys->pre_id1=psys->id1;
	psys->pre_id2=psys->id2;
	if(psys->objindex < pcsta->objectCount)
	{
		switch(pobj->csta_type)
		{
		case 0:
			if(csta_send(pcsta,psys))
			{
				psys->objindex++;
			}
			break;
		case 1:
			//csta_log("psys->id1:%d",psys->id1);
			if(psys->id1 <= pobj->id1_end)
			{
				if(csta_send(pcsta,psys))
				{
					psys->id1++;
				}
			}
			else
			{
				psys->objindex++;
				pobj_n=&pcsta->objects[psys->objindex];
				psys->id1=pobj_n->id1_start;
			}
			break;
		case 2:
			if(psys->id1 <= pobj->id1_end)
			{
				if(psys->id2 <= pobj->id2_end)
				{
					if(csta_send(pcsta,psys))
						psys->id2++;
				}
				else
				{
					psys->id1++;
					psys->id2=pobj->id2_start;
				}
			}
			else
			{
				psys->objindex++;
				pobj_n=&pcsta->objects[psys->objindex];
				psys->id1=pobj_n->id1_start;
				psys->id2=pobj_n->id2_start;
			}
			break;
		default:
			break;
		}
	}
	else
	{
		psys->objindex=0;
		pobj_n=&pcsta->objects[psys->objindex];
		psys->id1=pobj_n->id1_start;
		psys->id2=pobj_n->id2_start;
		
		if(psys->instance == global_instance)
		{
			psys->system_state=STATE_NORMAL;
			psys->msg_state=STATE_MSG_NONE;
			csta_log("sysTypeNo=%d,finish to get instance=%d csta data.",pcsta->sysTypeNo,psys->instance);
			if(!pcsta->single_message_flag)
				csta_record_multi(pcsta,psys);
		}
		else
		{
			if(psys->system_state == STATE_CHASE)
			{
#if CSTA_DEBUG
				csta_log("sysTypeNo=%d,finish to chase instance=%d csta data.",pcsta->sysTypeNo,psys->instance);
#endif
				if(!pcsta->single_message_flag)
					csta_record_multi(pcsta,psys);
			}
		}
		
		psys->instance++;
		psys->instance%=96;
	}
}

int csta_send(csta_src *pcsta,csta_sys *psys)
{
	snmp_pdu csta_pdu;
	csta_obj *pobj;
	int b_oidlen;
	
	pobj=&pcsta->objects[psys->objindex];
	
	csta_pdu.pdu_type=PDU_GET;
	csta_pdu.var_num=1;
	sprintf(csta_pdu.community,"public");
	b_oidlen=pobj->oidlen;
	memcpy(csta_pdu.var[0].oid,pobj->oid,b_oidlen * 4);
	
	switch(pobj->csta_type)
	{
	case 0:
		if(pcsta->sysTypeNo==226 || pcsta->sysTypeNo==227){
			//Entry OID
			csta_pdu.var[0].oid[b_oidlen++]=1;
			//Data OID
			csta_pdu.var[0].oid[b_oidlen++]=2; 
		}
		break;
	case 1:
		csta_pdu.var[0].oid[b_oidlen++]=psys->id1;
		if(pcsta->sysTypeNo==320 || pcsta->sysTypeNo==379 || pcsta->sysTypeNo==385 || pcsta->sysTypeNo==387)
		{
			//Entry OID
			csta_pdu.var[0].oid[b_oidlen++]=1;
			//Data OID
			csta_pdu.var[0].oid[b_oidlen++]=2; 
			//printf("sysTypeNo=%d,csta_type=1,objindex=%d,id1=%d\n",pcsta->sysTypeNo,psys->objindex,psys->id1);
		}
		
		break;
	case 2:
		if(pcsta->sysTypeNo != 350){
		csta_pdu.var[0].oid[b_oidlen++]=psys->id1;
		}
		if(pcsta->sysTypeNo==320 || pcsta->sysTypeNo==378|| pcsta->sysTypeNo==387)
		{
			//Entry OID
			csta_pdu.var[0].oid[b_oidlen++]=1;
			//Data OID
			csta_pdu.var[0].oid[b_oidlen++]=3;
			//printf("sysTypeNo=%d,csta_type=2,objindex=%d,id1=%d,id2=%d\n",pcsta->sysTypeNo,psys->objindex,psys->id1,psys->id2);
		}
		csta_pdu.var[0].oid[b_oidlen++]=psys->id2;
		break;
	}
	if(psys->system_state == STATE_NORMAL)
		psys->instance=getInstance();
	csta_pdu.var[0].oid[b_oidlen]=psys->instance;
	csta_pdu.var[0].oidlen=b_oidlen+1;
	csta_pdu.request_id=0;
	
	psys->request_id=snmp_send(&csta_pdu,&psys->addr);
	psys->msg_state=STATE_MSG_WAIT_RESPONSE;
	
	//printf("send,sysTypeNo=%d,request_id=%ld,instance=%d,obj=%d,id1=%d,id2=%d\n",pcsta->sysTypeNo,psys->request_id,psys->instance,psys->objindex,psys->id1,psys->id2);
	return 1;
}

void csta_receive()
{
	snmp_pdu csta_pdu;
	snmp_addr addr; 
	int loop=20;
	addr.local_port=CSTA_PORT;
	while(loop-->0)
	{
		if(snmp_receive(&csta_pdu,&addr) == 0)
		{
			continue;
		}
		if(csta_pdu.pdu_type != PDU_RSP)
		{	
			continue;
		}
		csta_analyze(&csta_pdu);
		break;
	}
}

void csta_analyze(snmp_pdu *pdu)
{
	csta_src *pcsta;
	csta_sys *psys;
	csta_obj *pobj;
	var_list *pvar;
	int i,j;
	
	pvar=&pdu->var[0];
	for(i=0;i<csta_src_num;i++)
	{
		pcsta=&CstaSet[i];
		for(j=0;j<pcsta->systemCount;j++)
		{
			psys=&pcsta->systems[j];
			if(psys->msg_state == STATE_MSG_WAIT_RESPONSE && psys->request_id == pdu->request_id)
			{
				//printf("receive,sysTypeNo=%d,pre_objindex=%d,objindex=%d\n",pcsta->sysTypeNo,psys->pre_objindex,psys->objindex);
				psys->fail=0;
				psys->retry=0;
				psys->request_id=0;
				csta_record(pcsta,psys,pdu);

				if(psys->system_state == STATE_NORMAL)
				{
					if(psys->pre_objindex == (pcsta->objectCount-1))
					{
						pobj=&pcsta->objects[psys->pre_objindex];
						switch(pobj->csta_type)
						{
						case 0:
							psys->msg_state=STATE_MSG_NONE;
							csta_log("sysTypeNo=%d,finish to get instance=%d csta data!",pcsta->sysTypeNo,psys->instance);
							if(!pcsta->single_message_flag)
							{
								csta_record_multi(pcsta,psys);
							}
							break;
						case 1:
							if(psys->id1 == (pobj->id1_end+1))
							{
								psys->msg_state=STATE_MSG_NONE;
								csta_log("sysTypeNo=%d,finish to get instance=%d csta data!!",pcsta->sysTypeNo,psys->instance);
							}
							else
							{
								psys->msg_state=STATE_MSG_GET_REQUEST;
							}
							break;
						case 2:
							if(psys->id1 == (pobj->id1_end+1) && psys->id2 == (pobj->id2_end+1))
							{
								psys->msg_state=STATE_MSG_NONE;
								csta_log("sysTypeNo=%d,finish to get instance=%d csta data!!!",pcsta->sysTypeNo,psys->instance);
							}
							else
							{
								psys->msg_state=STATE_MSG_GET_REQUEST;
							}
						}
					}
					//normal get
					else
					{
						psys->msg_state=STATE_MSG_GET_REQUEST;
					}
				}
				//chase
				else
				{
					psys->msg_state=STATE_MSG_GET_REQUEST;
				}
				return;
			}
		}
	}
	
	if(i == csta_src_num)
	{
		csta_log("The package is not match any system, discard it");
	}
	return;
}

int csta_record(csta_src *pcsta,csta_sys *psys,snmp_pdu *pdu)
{
	csta_obj *pobj;
	var_list *pvar;
	DWORD d_timestamp,*p_long; 
	DWORD *pval4;
	WORD *pval2;
	char sqlstr[8192],sumsqlstr[8192],sumCondition[512],sumFieldList[8192];
	char sys_field[128],sys_value[128];
	char key_field[128],key_value[128];
	char tmpstr3[128];
	int i;
	//for sum up
	int needSumFlag=0;
	int isRecord=0;
	MYSQL_RES *res_set;
	MYSQL_ROW row;
	
	memset(sqlstr,0,sizeof(sqlstr));
	memset(sumsqlstr,0,sizeof(sumsqlstr));
	memset(sumCondition,0,sizeof(sumCondition));
	memset(sumFieldList,0,sizeof(sumFieldList));
	memset(sys_field,0,sizeof(sys_field));
	memset(sys_value,0,sizeof(sys_field));
	memset(key_field,0,sizeof(sys_field));
	memset(key_value,0,sizeof(sys_field));
	memset(tmpstr3,0,sizeof(tmpstr3));
	
	pvar=&pdu->var[0];

	if(pvar->vartype == 0x04 && pvar->msglen > 0)
	{
		pobj=&pcsta->objects[psys->pre_objindex];
		p_long = (DWORD *) pvar->msg;
		d_timestamp = htonl(*p_long);
		
		//no timestamp
		if(d_timestamp == 0)
		{
#if CSTA_DEBUG
			csta_log("sysTypeNo=%d,objindex=%d,timestamp=0",pcsta->sysTypeNo,psys->objindex);
#endif
			return 0;
		}
			
		//message length invalid
		if(pvar->msglen < 4 + pobj->field_width * pobj->detail_field_count)
		{
			csta_log("sysTypeNo=%d,objindex=%d,message length(%d) is invalid,must be(%d)",pcsta->sysTypeNo,psys->objindex,pvar->msglen,4 + pobj->field_width * pobj->detail_field_count);
			return 0;
		}
		
		if(pcsta->sum_flag && strcmp(pobj->relate_table,"NoRelation") != 0)
			needSumFlag=1;
		
		if(pcsta->hassubsys)
		{
			sprintf(sys_field,"key_sys_id,key_subsys_id,");
			sprintf(sys_value,"%d,%d,",psys->sysid,psys->subsysid);
			if(needSumFlag)
			{
				sprintf(tmpstr3,"key_sys_id=%d and key_subsys_id=%d",psys->sysid,psys->subsysid);
				strcat(sumCondition,tmpstr3);
			}
		}
		else
		{
			sprintf(sys_field,"key_sys_id,");
			sprintf(sys_value,"%d,",psys->sysid);
			if(needSumFlag)
			{
				sprintf(tmpstr3,"key_sys_id=%d",psys->sysid);
				strcat(sumCondition,tmpstr3);
			}
		}

		switch(pobj->csta_type) 
		{
		case 1:
			sprintf(key_field,"%s,",pobj->id1_field);
			sprintf(key_value,"%d,",psys->pre_id1 + pobj->id1_offset);
			if(needSumFlag)
			{
				sprintf(tmpstr3," and %s=%d",pobj->id1_field,psys->pre_id1 + pobj->id1_offset);
				strcat(sumCondition,tmpstr3);
			}

			break;
		case 2:
			sprintf(key_field,"%s,%s,",pobj->id1_field, pobj->id2_field);
			sprintf(key_value,"%d,%d,",psys->pre_id1 + pobj->id1_offset,psys->pre_id2 + pobj->id2_offset);
			if(needSumFlag)
			{
				sprintf(tmpstr3," and %s=%d and %s=%d",pobj->id1_field,psys->pre_id1 + pobj->id1_offset,pobj->id2_field,psys->pre_id2 + pobj->id2_offset);
				strcat(sumCondition,tmpstr3);
			}

			break;
		}
		//csta_log("single_message_flag:%d",pcsta->single_message_flag);
		if(pcsta->single_message_flag)
		{
			sprintf(sqlstr, "replace into %s.%sData (%s%s%s,csta_datetime)VALUES(%s%s",
				CSTA_DB,pobj->detail_table,sys_field,key_field,pobj->sql_detail,sys_value,key_value);

			//sum up accord to the map table
			if(needSumFlag)
			{
				sprintf(tmpstr3," and UNIX_TIMESTAMP(csta_datetime) = %ld",d_timestamp);
				strcat(sumCondition,tmpstr3);
				sprintf(sumsqlstr,"select %s from %s.%sData where %s",pobj->sql_sum,CSTA_DB,pobj->detail_table,sumCondition);

			}
			
			if(pobj->field_width == 2)
			{
				for(i=0;i<pobj->detail_field_count;i++)
				{
					pval2=(WORD*)(pvar->msg + 4 + i * 2);
					sprintf(tmpstr3,"%d,",htons(*pval2));
					strcat(sqlstr,tmpstr3);
				}
			}
			else
			{
			if (pcsta->sysTypeNo != 362)
			{
				for(i=0;i<pobj->detail_field_count;i++)
				{
					pval4=(DWORD*)(pvar->msg + 4 + i * 4);
					sprintf(tmpstr3,"%u,",htonl(*pval4));
					strcat(sqlstr,tmpstr3);
				}
			}else{ //SPGW
					for(i=0;i<pobj->detail_field_count-2;i++)
					{
						pval4=(DWORD*)(pvar->msg + 4 + i * 4);
						sprintf(tmpstr3,"%u,",htonl(*pval4));
						strcat(sqlstr,tmpstr3);

						   // printf("Field %d: %u\n", i + 1, htonl(*pval4));

					} //前十六个

				DWORD *uplink_tmp1, *uplink_tmp2;
				DWORD *downlink_tmp1, *downlink_tmp2;
				uint64_t uplink_total_volume; 
				uint64_t downlink_total_volume; 
				// 处理 uplink_total_volume
				uplink_tmp1 = (DWORD*)(pvar->msg + 4 + (pobj->detail_field_count - 2) * 4); //68
				uplink_tmp2 = (DWORD*)(pvar->msg + 4 + (pobj->detail_field_count -1) * 4); // 72
				uplink_total_volume = ((uint64_t)ntohl(*uplink_tmp1) << 32) +ntohl(*uplink_tmp2);

				sprintf(tmpstr3, "%llu,", uplink_total_volume);
				strcat(sqlstr, tmpstr3);

				// 处理 downlink_total_volume
				downlink_tmp1 = (DWORD*)(pvar->msg + 4 + (pobj->detail_field_count ) * 4); //76开始
				downlink_tmp2 = (DWORD*)(pvar->msg + 4 + (pobj->detail_field_count +1) * 4); //80开始
				downlink_total_volume = ((uint64_t)ntohl(*downlink_tmp1) << 32) +ntohl(*downlink_tmp2);

				sprintf(tmpstr3, "%llu,", downlink_total_volume);
				strcat(sqlstr, tmpstr3);
			}
			}
			sprintf(tmpstr3,"FROM_UNIXTIME(%ld))",d_timestamp);
			strcat(sqlstr,tmpstr3);
			
			if(mysql_getnores(pCstaConn,sqlstr) != 0)
			{
				
				if(mysql_errno(pCstaConn) == 1146)
				{
					char dstTable[64];
					sprintf(dstTable,"%sData",pobj->detail_table);
					if(mysql_tblCpy(pCstaConn,dstTable,pobj->detail_table) != 0)
					{
						csta_log("create table %s fail",dstTable);
					}
					else
					{
						if(mysql_getnores(pCstaConn,sqlstr) != 0)
						{
							csta_log("Store csta fail,sysTypeNo=%d,instance=%d,objindex=%d,id1=%d,id2=%d",pcsta->sysTypeNo,psys->pre_instance,psys->pre_objindex,psys->pre_id1,psys->pre_id2);
							csta_log("SQL-1:%s\n",sqlstr);
						}
						else
						{
							//csta_log("Record single csta data,sysTypeNo=%d,instance=%d,objindex=%d,id1=%d,id2=%d,time=%s",pcsta->sysTypeNo,psys->pre_instance,psys->pre_objindex,psys->pre_id1,psys->pre_id2,ctime((time_t*)&d_timestamp));
							isRecord=1;
						}
					}
				}
				else
				{
					csta_log("Store csta fail,sysTypeNo=%d,instance=%d,objindex=%d,id1=%d,id2=%d",pcsta->sysTypeNo,psys->pre_instance,psys->pre_objindex,psys->pre_id1,psys->pre_id2);
					csta_log("SQL-2:%s\n",sqlstr);
				}
			}
			else
			{
				//csta_log("Record single csta data,sysTypeNo=%d,instance=%d,objindex=%d,id1=%d,id2=%d,time=%s",pcsta->sysTypeNo,psys->pre_instance,psys->pre_objindex,psys->pre_id1,psys->pre_id2,ctime((time_t*)&d_timestamp));
				isRecord=1;
			}
			
			//printf("%s\n",sqlstr);
		}
		else
		{
			//multi-message consist one record
			//csta_log("Store the objindex=%d csta data for sysTypeNo=%d,instance=%d,pvar->msglen=%d",psys->pre_objindex,pcsta->sysTypeNo,psys->pre_instance,pvar->msglen);
			memcpy(psys->cstamsg[psys->pre_objindex],pvar->msg,pvar->msglen);
			psys->msglen+=pvar->msglen;
		}

	}
	else
	{
		//csta_log("Record csta fail,msglen=%d or vartype=%02x,sysTypeNo=%d,instance=%d,objindex=%d,id1=%d,id2=%d",pvar->msglen,pvar->vartype,pcsta->sysTypeNo,psys->pre_instance,psys->pre_objindex,psys->pre_id1,psys->pre_id2);
		return 0;
	}
	
	if(isRecord&&needSumFlag)
	{
		res_set=mysql_getres(pCstaConn,sumsqlstr);
		if(res_set == NULL)
		{
			csta_log("Sumup csta fail,res_set is null,sysTypeNo=%d,instance=%d,objindex=%d,id1=%d,id2=%d",pcsta->sysTypeNo,psys->pre_instance,psys->pre_objindex,psys->pre_id1,psys->pre_id2);
			return 0;
		}
		if(mysql_num_rows(res_set) != 1)
		{
			csta_log("Sumup csta fail,rownum=%d,sysTypeNo=%d,instance=%d,objindex=%d,id1=%d,id2=%d",mysql_num_rows(res_set),pcsta->sysTypeNo,psys->pre_instance,psys->pre_objindex,psys->pre_id1,psys->pre_id2);
			mysql_free_result(res_set);
			return 0;
		}
		
		row=mysql_fetch_row(res_set);
		if(row == NULL)
		{
			csta_log("Sumup csta fail,rowis null,sysTypeNo=%d,instance=%d,objindex=%d,id1=%d,id2=%d",pcsta->sysTypeNo,psys->pre_instance,psys->pre_objindex,psys->pre_id1,psys->pre_id2);
			mysql_free_result(res_set);
			return 0;
		}
		
		mysql_free_result(res_set);
		
		memset(sumFieldList,0,sizeof(sumFieldList));
		for(i=0;i<pobj->sum_field_count;i++)
		{
			sprintf(tmpstr3,"%s,",pobj->sum_field_list[i]);
			strcat(sumFieldList,tmpstr3);
		}
		
		sprintf(sqlstr, "replace into %s.%sData (%s%s%scsta_datetime)VALUES(%s%s",
				CSTA_DB,pobj->sum_table,sys_field,key_field,sumFieldList,sys_value,key_value);



		for(i=0;i<pobj->sum_field_count;i++)
		{
			if(row[i] == NULL)
			{
				sprintf(tmpstr3,"0");
			}
			else
			{
				if(strlen(row[i]) <= 0)
					sprintf(tmpstr3,"0");
				else
					sprintf(tmpstr3,"%s",row[i]);
			}
			strcat(sqlstr,tmpstr3);
			strcat(sqlstr,",");
		}
		sprintf(tmpstr3,"FROM_UNIXTIME(%ld))",d_timestamp);
		strcat(sqlstr,tmpstr3);

		//only 320 mscSumData.check Signal Error
 		if (pcsta->sysTypeNo == 320 && strcmp(pobj->sum_table, "mscSumData") == 0) {

			int sysNo = psys->sysid;       // 从 psys 获取 sysNo
			int subSysNo = psys->subsysid;

			unsigned int SignalErr = 0, Attempt = 0 ,KeyType =0;
			sscanf(sqlstr, "%*[^,],%*[^,],%u,%*[^,],%u,%*[^,],%u", &KeyType,&Attempt, &SignalErr);

		// MO-12 MT-13
		if (Attempt > 0 && (KeyType == 12 || KeyType == 13)) {
			double ratio = (double)SignalErr / Attempt;

			// 如果 SignalErr / Attempt > 5%，则触发告警
			if (ratio > 5.0) {
				time_t current_time = time(NULL);
				char alarmIdStr[32];
				sprintf(alarmIdStr, "%ld", (long)current_time); // 将时间戳转换为字符串

				// 构造插入 alarmInfo 表的 SQL 语句
				sprintf(cstaproc_sqlstr, "insert into OMC_PUB.sysAlarmInfo(sysTypeNo,sysNo,subSysNo,compCode,alarmCode,alarmTime,clearTime,clearMode,alarmInfo,alarmId) "
										"values(%d,%d,%d,%d,%d,CURRENT_TIMESTAMP,'0000-00-00 00:00:00',1,'Call failure rate exceeds 5%%');",
						pcsta->sysTypeNo,sysNo,subSysNo, 0, 10,  alarmIdStr);

				// 执行 SQL 语句
				if (mysql_getnores(pubConn, cstaproc_sqlstr) != 0) {
					csta_log("[SQL ERR]: %s", cstaproc_sqlstr);
				} else {
					csta_log("Inserted into alarmInfo: %s", cstaproc_sqlstr);
				}
			}else{
				//clear 5% alarm
						sprintf(cstaproc_sqlstr, "update OMC_PUB.sysAlarmInfo set clearTime=CURRENT_TIMESTAMP,clearMode=0,clearBy='System' "
							"where sysTypeNo=%d and sysNo=%d and subSysNo=%d and compCode=%d and alarmCode=%d and clearMode=1;",
				pcsta->sysTypeNo, sysNo, subSysNo, 0, 10);

				// 执行 SQL 语句
				if (mysql_getnores(pubConn, cstaproc_sqlstr) != 0) {
				csta_log("[SQL ERR]: %s", cstaproc_sqlstr);
				} else {
				csta_log("Cleared alarmInfo: %s", cstaproc_sqlstr);
				}				
			}
		} else {
			csta_log("Attempt is zero, skipping SignalErr/Attempt calculation.");
		}			

		}

		if(mysql_getnores(pCstaConn,sqlstr) != 0)
		{
			if(mysql_errno(pCstaConn) == 1146)
			{
				char dstTable[64];
				sprintf(dstTable,"%sData",pobj->sum_table);
				if(mysql_tblCpy(pCstaConn,dstTable,pobj->sum_table) != 0)
				{
					csta_log("create table %s fail",dstTable);
				}
				else
				{
					if(mysql_getnores(pCstaConn,sqlstr) != 0)
					{
						csta_log("Sumup csta fail,sysTypeNo=%d,instance=%d,objindex=%d,id1=%d,id2=%d",pcsta->sysTypeNo,psys->pre_instance,psys->pre_objindex,psys->pre_id1,psys->pre_id2);
						return 0;
					}
					else
					{
						;//csta_log("Sumup single csta data,sysTypeNo=%d,instance=%d,objindex=%d,id1=%d,id2=%d,time=%s",pcsta->sysTypeNo,psys->pre_instance,psys->pre_objindex,psys->pre_id1,psys->pre_id2,ctime((time_t*)&d_timestamp));
					}
				}
			}
			else
			{
				csta_log("Sumup csta fail,sysTypeNo=%d,instance=%d,objindex=%d,id1=%d,id2=%d",pcsta->sysTypeNo,psys->pre_instance,psys->pre_objindex,psys->pre_id1,psys->pre_id2);
				csta_log("%s\n",sqlstr);
				return 0;
			}
		}
		else
		{
			;//csta_log("Sumup single csta data,sysTypeNo=%d,instance=%d,objindex=%d,id1=%d,id2=%d,time=%s",pcsta->sysTypeNo,psys->pre_instance,psys->pre_objindex,psys->pre_id1,psys->pre_id2,ctime((time_t*)&d_timestamp));
		}

	}
	return 1;
}

int csta_record_multi(csta_src *pcsta,csta_sys *psys)
{
	DWORD d_timestamp,*p_long;
	DWORD *pval4;
	WORD *pval2;
	char sqlstr[8192],tol_field_sql[8192];
	char sys_field[128],sys_value[128];
	char tmpstr3[8192];
	
	unsigned long ref_data_len;
	csta_obj *pobj;
	int i,j;
	
	memset(sqlstr,0,sizeof(sqlstr));
	memset(tol_field_sql,0,sizeof(tol_field_sql));
	memset(sys_field,0,sizeof(sys_field));
	memset(sys_value,0,sizeof(sys_value));
	memset(tmpstr3,0,sizeof(tmpstr3));

	ref_data_len=0;
	//csta_log("objectCount:%d",pcsta->objectCount);
	for(i=0;i<pcsta->objectCount;i++)
	{
		pobj=&pcsta->objects[i];
		if(pobj->csta_type != 0)
		{
			csta_log("The %d system csta source configure error, it must be 0 when isFullRow=0",pcsta->sysTypeNo);
			psys->msglen=0;
			return 0;
		}
		//csta_log("detail_field_count:%d,field_width:%d",pobj->detail_field_count,pobj->field_width);
		ref_data_len += pobj->detail_field_count * pobj->field_width;
		ref_data_len += 4;
		//csta_log("ref_data_len[%d]:%d",i,ref_data_len);
	}
	
	if(ref_data_len != psys->msglen)
	{
		csta_log("Record multi csta data fail,sysTypeNo=%d,instance=%d,message length(%ld) != %ld",pcsta->sysTypeNo,psys->pre_instance,psys->msglen,ref_data_len);
		psys->msglen=0;
		return 0;
	}
	
	psys->msglen=0;
	psys->objindex=0;
	p_long=(DWORD *)psys->cstamsg[0];
	d_timestamp=htonl(*p_long);
	
	if(pcsta->hassubsys)
	{
		sprintf(sys_field,"key_sys_id,key_subsys_id,");
		sprintf(sys_value,"%d,%d,",psys->sysid,psys->subsysid);
	}
	else
	{
		sprintf(sys_field,"key_sys_id,");
		sprintf(sys_value,"%d,",psys->sysid);
	}
	
	memset(tol_field_sql,0,sizeof(tol_field_sql));
	for(i=0;i<pcsta->objectCount;i++)
	{
		sprintf(tmpstr3,"%s",pcsta->objects[i].sql_detail);
		if(i != (pcsta->objectCount-1))
			strcat(tmpstr3,",");
		strcat(tol_field_sql,tmpstr3);
	}
	
	sprintf(sqlstr, "replace into %s.%sData (%s%s,csta_datetime)VALUES(%s",
				CSTA_DB,pobj->detail_table,sys_field,tol_field_sql,sys_value);
				
	for(i=0;i<pcsta->objectCount;i++)
	{
		pobj=&pcsta->objects[i];
		for(j=0;j<pobj->detail_field_count;j++)
		{
			if(pobj->field_width == 2)
			{
				pval2=(WORD*)(psys->cstamsg[i] + 4 + j * 2);
				sprintf(tmpstr3,"%d,",htons(*pval2));
				strcat(sqlstr,tmpstr3);
			}
			else
			{
				pval4=(DWORD*)(psys->cstamsg[i] + 4 + j * 4);
				sprintf(tmpstr3,"%u,",htonl(*pval4));
				strcat(sqlstr,tmpstr3);
			}
		}
	}
	
	sprintf(tmpstr3,"FROM_UNIXTIME(%ld))",d_timestamp);
	strcat(sqlstr,tmpstr3);
	
	if(mysql_getnores(pCstaConn,sqlstr) != 0)
	{	
		if(mysql_errno(pCstaConn) == 1146)
		{
			char dstTable[64];
			sprintf(dstTable,"%sData",pcsta->objects[0].detail_table);
			if(mysql_tblCpy(pCstaConn,dstTable,pcsta->objects[0].detail_table) != 0)
			{
				csta_log("create table %s fail",dstTable);
				return 0;
			}
			else
			{
				if(mysql_getnores(pCstaConn,sqlstr) != 0)
				{
					csta_log("Store csta fail,sysTypeNo=%d,instance=%d",pcsta->sysTypeNo,psys->pre_instance);
					csta_log("SQL-3:%s\n",sqlstr);
					return 0;
				}
				else
				{
					;//csta_log("Record multi csta data,sysTypeNo=%d,instance=%d,time=%s",pcsta->sysTypeNo,psys->pre_instance,ctime((time_t*)&d_timestamp));
				}
			}
		}
		else
		{
			csta_log("Store csta fail,sysTypeNo=%d,instance=%d",pcsta->sysTypeNo,psys->pre_instance);
			csta_log("SQL4:%s\n",sqlstr);
			return 0;
		}
	}
	else
	{
		;//csta_log("Record multi csta data,sysTypeNo=%d,instance=%d,time=%s",pcsta->sysTypeNo,psys->pre_instance,ctime((time_t*)&d_timestamp));
	}
	
	return 1;
}

int csta_create_csv()
{
	csta_src *pcsta;
	char sqlstr[512];
	char command[512];
	time_t l_time;
	struct tm *t;
	char csvfile[64];
	char tablename[64];
	
	if(csv_src_index == csta_src_num)
		return 1;
		
	pcsta=&CstaSet[csv_src_index];
	//Create the csv dir if not exist
	if(chdir(pcsta->csv_path) == -1) 
	{
		sprintf(command,"mkdir -p -m755 %s",pcsta->csv_path);
		system(command);
		csta_log("The %s is not exist, create it",pcsta->csv_path);
		if(chdir(pcsta->csv_path) == -1)
		{	
			csta_log("Create %s fail",pcsta->csv_path);
			csv_src_index++;
			return 0;
		}
	}
	
	if(chdir(csv_sent_dir) == -1) 
	{
		sprintf(command,"mkdir -p -m755 %s",csv_sent_dir);
		system(command);
		csta_log("The %s is not exist, create it",csv_sent_dir);
		if(chdir(csv_sent_dir) == -1)
		{	
			csta_log("Create %s fail",csv_sent_dir);
			csv_src_index++;
			return 0;
		}
	}
	
	if(chdir(csv_sending_dir) == -1) 
	{
		sprintf(command,"mkdir -p -m755 %s",csv_sending_dir);
		system(command);
		csta_log("The %s is not exist, create it",csv_sending_dir);
		if(chdir(csv_sending_dir) == -1)
		{	
			csta_log("Create %s fail",csv_sending_dir);
			csv_src_index++;
			return 0;
		}
	}
	
	l_time=time(NULL) - 24 * 60 * 60;
	t=localtime(&l_time);

	sprintf(csvfile,"%s_%04d%02d%02d.csv",pcsta->objects[0].detail_table,t->tm_year+1900,t->tm_mon+1,t->tm_mday);
	sprintf(tablename,"%sData",pcsta->objects[0].detail_table);
	sprintf(sqlstr,"SELECT * INTO OUTFILE '%s%s' FIELDS TERMINATED BY ',' ENCLOSED BY '' LINES TERMINATED BY '\n' FROM %s.%s WHERE csta_datetime >= '%04d-%02d-%02d 00:00:00' AND csta_datetime < '%04d-%02d-%02d 23:59:59'",
			pcsta->csv_path,csvfile,CSTA_DB,tablename,
			t->tm_year + 1900,t->tm_mon + 1,t->tm_mday,t->tm_year + 1900,t->tm_mon + 1,t->tm_mday);
	//printf("%s\n",sqlstr);
	
	if(mysql_getnores(pCstaConn,sqlstr) != 0)
	{
		csta_log("Create %s fail",csvfile);
	}
	else
	{
		csta_log("Create %s successfully",csvfile);
		if(csv_send_flag)
		{
			sprintf(command,"cp %s%s %s/%s",pcsta->csv_path,csvfile,csv_sending_dir,csvfile);
			system(command);
			
			sprintf(command,"cp %s%s %s/%s",pcsta->csv_path,csvfile,csv_sent_dir,csvfile);
			system(command);
			
			sprintf(command,"chmod 644 %s%s",pcsta->csv_path,csvfile);
			system(command);
			sprintf(command,"chown mysql %s%s",pcsta->csv_path,csvfile);
			system(command);
			sprintf(command,"chgrp mysql %s%s",pcsta->csv_path,csvfile);
			system(command);
			
			csta_log("Move %s to %s prepare to send",csvfile,csv_sending_dir);
		}
		else
		{
			sprintf(command,"chmod 644 %s%s",pcsta->csv_path,csvfile);
			system(command);
		}
	}
	
	l_time=time(NULL) - 24 * 60 * 60;
	t=localtime(&l_time);
	//Create the mscSum csv table
	if(pcsta->sum_flag)
	{
		sprintf(csvfile,"%s_%04d%02d%02d.csv",pcsta->objects[0].sum_table,t->tm_year+1900,t->tm_mon+1,t->tm_mday);
		sprintf(tablename,"%sData",pcsta->objects[0].sum_table);
		sprintf(sqlstr,"SELECT * INTO OUTFILE '%s%s' FIELDS TERMINATED BY ',' ENCLOSED BY '' LINES TERMINATED BY '\n' FROM %s.%s WHERE csta_datetime >= '%04d-%02d-%02d 00:00:00' AND csta_datetime < '%04d-%02d-%02d 23:59:59'",
				pcsta->csv_path,csvfile,CSTA_DB,tablename,
				t->tm_year + 1900,t->tm_mon + 1,t->tm_mday,t->tm_year + 1900,t->tm_mon + 1,t->tm_mday);
		//printf("%s\n",sqlstr);
		
		if(mysql_getnores(pCstaConn,sqlstr) != 0)
		{
			csta_log("Create %s fail",csvfile);
		}
		else
		{
			csta_log("Create %s successfully",csvfile);
			if(csv_send_flag)
			{
				sprintf(command,"cp %s%s %s/%s",pcsta->csv_path,csvfile,csv_sending_dir,csvfile);
				system(command);
				
				sprintf(command,"cp %s%s %s/%s",pcsta->csv_path,csvfile,csv_sent_dir,csvfile);
				system(command);
				
				sprintf(command,"chmod 644 %s%s",pcsta->csv_path,csvfile);
				system(command);
				sprintf(command,"chown mysql %s%s",pcsta->csv_path,csvfile);
				system(command);
				sprintf(command,"chgrp mysql %s%s",pcsta->csv_path,csvfile);
				system(command);
				
				csta_log("Move %s to %s prepare to send",csvfile,csv_sending_dir);
			}
			else
			{
				sprintf(command,"chmod 644 %s%s",pcsta->csv_path,csvfile);
				system(command);
			}
		}
	}

	// by simon begin --- 
	// Delete the over-time record
	l_time = time(NULL) - max_keep_days * 24 * 60 * 60;
	// --- end
	t=localtime(&l_time);
	sprintf(tablename,"%sData",pcsta->objects[0].detail_table);
	sprintf(sqlstr,"DELETE FROM %s.%s WHERE csta_datetime < '%04d-%02d-%02d 00:00:00' ",
		CSTA_DB,tablename,t->tm_year + 1900,t->tm_mon + 1,t->tm_mday);
	csta_log("%s",sqlstr);	
	if(mysql_getnores(pCstaConn,sqlstr) != 0)
	{
		csta_log("Delete over-time record of %s fail",tablename);
	}
	else
	{
		csta_log("Delete over-time1 record of %s successfully",tablename);
	}
	
	if(pcsta->sum_flag)
	{
		// by simon begin ---
		l_time = time(NULL) - max_keep_days * 24 * 60 * 60;
		// --- end
		t=localtime(&l_time);
		sprintf(tablename,"%sData",pcsta->objects[0].sum_table);
		sprintf(sqlstr,"DELETE FROM %s.%s WHERE csta_datetime < '%04d-%02d-%02d 00:00:00' ",
			CSTA_DB,tablename,t->tm_year + 1900,t->tm_mon + 1,t->tm_mday);
		csta_log("%s",sqlstr);
		if(mysql_getnores(pCstaConn,sqlstr) != 0)
		{
			csta_log("Delete over-time record of %s fail",tablename);
		}
		else
		{
			csta_log("Delete over-time2 record of %s successfully",tablename);
		}
	}
	
	
	csv_src_index++;
	return 1;
}


void csta_log(const char *fmt, ...)
{
	char logFile[64];
	FILE *fp;
	time_t l_time;
	va_list ap;
	char buf[8192];
	struct tm *t;
	char timestr[64];
	
	l_time=time(NULL);
	t=localtime(&l_time);
	sprintf(logFile,"/usr/local/omc/log/cstalog_%04d%02d%02d.log",t->tm_year+1900,t->tm_mon+1,t->tm_mday);
	fp=fopen(logFile,"a");
	if(fp!=NULL)
	{
		va_start(ap,fmt);
		vsprintf(buf,fmt,ap);
		sprintf(timestr,"%02d:%02d:%02d: ",t->tm_hour,t->tm_min,t->tm_sec);
		fputs(timestr,fp);
		fputs(buf,fp);
		fputs("\n",fp);
		fflush(fp);
		va_end(ap);
		fclose(fp);
	}
}

BYTE str2oid(char *str, DWORD * array, BYTE max)
{
	BYTE sub = 0;
	short len, i;
	char *pvar;

	len = strlen(str);
	pvar = str;
	for (i = 0; i < len && sub < max; i++) {
		if (str[i] == '.') {
			str[i] = '\0';
			if (strlen(pvar) == 0)
				continue;
			array[sub++] = atoi(pvar);
			pvar = str + i + 1;
		}
	}
	if (strlen(pvar) == 0)
		return sub;
	array[sub++] = atoi(pvar);
	return sub;
}

int parseLine(char *str, char (*row)[32], int maxrow)
{
	char *pstr = str;
	int len, index = 0, i;
	BOOL black = 0;
	len = strlen(str);
	for (i = 0; i < len && index < maxrow; i++) 
	{
		if (isspace(str[i]) || str[i] == ',' || str[i] == ';' ||
			str[i] == '/' || str[i] == '|' || str[i] == '.') 
		{
			if (black == 0)
			{	
				pstr = str + i + 1;
			}
			else 
			{
				str[i] = '\0';
				if (strlen(pstr) > 64)
				{
					pstr[63] = 0;
				}
				sprintf(row[index++], "%s", pstr);
				pstr = str + i + 1;
			}
			black = 0;
		} 
		else
		{	
			black = 1;
		}
	}
	if (black == 1) 
	{
		if (strlen(pstr) > 64)
		{	
			pstr[63] = 0;
		}
		sprintf(row[index++], "%s", pstr);
	}
	return index;
}