话单文件采集测试：
1.基本测试：
基础流：设置本地时间，如：20071015 16:00:00 启动话单文件生成程序
2.规则测试：
备选流1：当日话单序号到达程序设定的最大值时，如：9999。话单文件序号将重置清零(20071015_0000)重新计数，测试是否可以正常采集
备选流2：跨天采集，当日话单文件到次日话单文件，如：20071015-20071016，修改测试机本地时间23:59:00
16日零时，话单文件序号将重置清零(20071016_0000)重新计数，测试是否可以正常采集
备选流3：当日(20071015)话单到达文件设定的最大值，还未到次日(20071016)，如：9999。
测试数据先将文件序号清零(20071015_0000)，查看是否可正常采集
其次，当到达次日时再次清零(20071016_0000)，查看是否可以正常采集
测试1=基础流+备选流1
测试2=基础流+备选流2
测试3=基础流+备选流3
3.容错性测试：
备选流1：跨月采集，设定话单文件为2007_10_31 23:58:00，查看是否可正常采集
备选流2：跨年采集，设定话单文件为2007_12_31 23:58:00，查看是否可正常采集
备选流3：当日(2007_12_31 23:58:00)话单到达文件设定的最大值，还未到次日(2008_01_01 00:00:00)，如：9999。
测试数据先将文件序号清零(20071231_0000)，查看是否可正常采集
其次，当到达次日时再次清零(20080101_0000)，查看是否可以正常采集
测试1=基础流+备选流1
测试2=基础流+备选流2
测试3=基础流+备选流3
4、最小话单序号采集，话单测试数据文件命名为：
第一组数据：6660 6661 0001 0002
第二组数据：6660 6661 0010 0011
第三组数据：6660 6661 0010 0011 0001
第四组数据：6660 6661 0010 0011 0035
5、话单文件序号有间隔
第一组数据：6660 6668 6669 6671
第二组数据：0000 0010 0005 0006
第三组数据：9999 0010 0014 0018
6、复合数据测试
跨天采集+当日序号清零
垮月采集+当日序号清零
跨年采集+当日序号清零
当日历史数据采集+当日序号清零
当月历史数据采集+当日序号清零
当年历史数据采集+当日序号清零
5、综合场景测试
复合所有采集规则
下面与各位分享一下，这个采集程序的部分Py脚本(有些地方写的很烂)
本次追加的功能：
1、更改的循环的方式，Python int类型不能支持百万级循环问题(改为多层嵌套，通过小时来控制循环时间)
2、修改了Sleep时间不支持毫秒级运算的问题
3、增加自动加载某些配置文件的方法
4、设置了可配置的话单最小序号控制阀值
5、修改了一些数据错误与Bug
6、增加了异常错误的日志分析
7、修改了更佳友好的操作界面，对一些输入值的判定条件进行了修改
config/config.ini
[config]
initialization_path=E://ftp//
data_prefixion=sdf
data_infix=rrse
data_postfix=.txt
initialization_number=0
run_count(hours)=1
min_circle_time=1
[workpath]
programme_workpath=C://Program Files//Huawei//infoX-InfoSpy//
ismg_probe_module=ismg_probe_module.ini
[switch]
the_second_day_min_count=0
the_current_day_min_count=1
# -*- coding: cp936 -*-#===============================================================================#程序需求：话单时间片替换程序#1.可自动识别任意话单的时间位置#2.可自动识别各种话单文件的长度，生成已当前本地时间为时间片的话单数据(替换原始话单中的时间列)#3.当日话单达到最大值9999时，话单文件序号自动清零重新计数，如：0000，0001#4.当话单文件跨天时，文件序号自动清零重新计数，如：0000，0001#5.提供参数配置功能(config.ini/run_log.txt/原始话单文件)#6.提供日志输出功能:可查看最后生成的话单文件关键参数与循环计数#7.min_circle_time(最小间隔时间支持毫秒级运算)，可以指定任意位数的float参量#8.加入隔天话单数据重新计数后，随机产生最小话单序号的为任意值(在配置文件中制作开关)#9.加入当天话单数据，随机产生跨号段序号(在配置文件中制作开关)#===============================================================================import os,time,sysclass DataBase(): def __init__(self): config_path = open(".//config//config.ini",'r') config_list = config_path.readlines() #通过配置文件获得话单数据的初始化文件路径 self.initialization_path = config_list[1][config_list[1].strip().find("=")+1:-1] #得到话单文件名前缀 self.data_prefixion = config_list[2][config_list[2].strip().find("=")+1:-1] #得到话单文件名日期 nowtime = time.localtime() #初始化参数 data1 = "" for i in range(3): #如果日期为单数需要补零 data1 += "%02d" % nowtime[i] self.data_data=data1 #得到话单文件中缀 self.data_infix = config_list[3][config_list[3].strip().find("=")+1:-1] #得到话单文件后缀 self.data_postfix = config_list[4][config_list[4].strip().find("=")+1:-1] #得到话单文件当前数据号 self.data_number = "%04d" % int(config_list[5][config_list[5].strip().find("=")+1:]) #print self.data_number self.data_path = self.initialization_path+"config/"+self.data_prefixion+self.data_data+self.data_infix+self.data_number+self.data_postfix #得到程序需要运行的次数 self.data_count = config_list[6][config_list[6].strip().find("=")+1:-1] #print self.data_count #得到话单文件生成的最小间隔时间 self.min_circle_time = config_list[7][config_list[7].strip().find("=")+1:-1] #得到程序工作路径 self.programme_workpath = config_list[10][config_list[10].strip().find("=")+1:-1] #得到短信网关配置文件名字 self.ismg_probe_module = config_list[11][config_list[11].strip().find("=")+1:-1] #得到隔天话单最小计数序号 self.the_second_day_min_count = config_list[14][config_list[14].strip().find("=")+1:-1] #得到当天话单数据间隔序号 self.the_current_day_min_count=config_list[15][config_list[14].strip().find("=")+2:] def collection_config(self): lastseq = raw_input("请输入上一次网关采集模块LastSeq的值:") #循环判断输入条件，只有满足条件才能执行之后的程序 while(True): try: lastseq = int(lastseq) break except: lastseq = raw_input("LastSeq不能为空或者非数字型字符，请重新输入:") continue count = [] path = self.programme_workpath+"config/"+self.ismg_probe_module #如果读取的文件失败，抛出异常供使用者进行检查 try: list = open(path,'r') data = list.readlines() list.close() for i in range(len(data)-1): if 'LastSeq' in data[i]: count.append(i) for i in range(len(count)): data.remove(data[count[i]]) data.insert(count[i],"LastSeq=%s " % (lastseq)) list_new = open(path,'w') for y in data: #print y list_new.writelines(y) list_new.close() print "----------------------------------------------------------------" print "ismg_probe_module.ini文件，LastSeq已被重置为%s!请重新启动网关采集模块！" % (lastseq) except: print "路径错误！请检查程序工作路径是否正确,再开始执行程序=",path def get_localtime(self): #获取本地时间，For Example：2008/10/02 00:44:11 nowtime = time.localtime() #print nowtime data1,time1 = "","" for i in range(3): #如果获得时间是单数，则需要补位0 data1 += ("%02d" % nowtime[i])+'/' #获得时间片的日期测试print data1[0:-1] for i in range(3,6): #如果获得日期是单数，则需要补位0 time1 += ("%02d" % nowtime[i])+':' #获得时间片的时间测试print time1[0:-1] #时间片的数据组合 time_area = data1[0:-1]+" "+time1[0:-1] return time_area def data_control(self): #得到需要重写话单文件的路径及命名self.data_count count = 0 data_time = self.data_data data_number = self.data_number dd = 1 #去掉了原来使用[i]循环，改为使用可配置LastCount来控制间隔循环次数 #由于需求修改为隔天采集序号可能是任意值，序号生成可能乱序，为了测试这种情况 #程序增加了对隔天最小话单序号的配置int(self.the_second_day_min_count) #以及当天话单序号的间隔值int(self.the_current_day_min_count) last_count = 0 for o in range(int(self.data_count)): for i in range(int(float(3600)/float(self.min_circle_time))): #循环开始话单生成的序号,加入了可配置的隔天序号，话单数据重置后初始化单生成的序号(可以指定任意的值开始) #i += 1 last_time = data_time last_data_number = data_number temp = dd last_count = last_count + int(self.the_current_day_min_count) if dd != 1: last_count1 = last_count - end_count #增加话单数据文件序号隔天重新计数 nowtime = time.localtime() data_time = "" for y in range(3): #如果日期为单数需要补零 data_time += "%02d" % nowtime[y] #如果当前取得时间是当天的 if data_time == last_time: last_data_number = data_number else: #重置循环次数,隔天生成的话单文件序号可配置last_count= #end_count = i end_count = last_count last_count = int(self.the_second_day_min_count) #制作标识为 dd = 2 #如果不相等，则重置时间序号 last_data_number = data_number = 0 #处理生成的文件路径数字超过4位 rebulid_number = "%04d" % int(str((int(last_data_number)+last_count))[-4:]) rebulid_path = self.initialization_path+self.data_prefixion+data_time+self.data_infix+rebulid_number+self.data_postfix #从初始化路径中得到数据 try: tel_node = open(self.data_path,"r") tel_node_new = open(rebulid_path,'w') tel_list = tel_node.readlines() for i in range(len(tel_list)): #数据的第一行 new_list = tel_list[i].split(',') for j in range(len(tel_list[i].split(','))): #赛选存在日期的序列，i等于时间片序列的纵行位置 if '/' in tel_list[i].split(',')[j]: #从原始列表中删除原始日期 new_list.remove(tel_list[i].split(',')[j]) #替换重新组合的日期时间 time_area = self.get_localtime() new_list.insert(j,time_area) #得到新的数据列表print new_list #合成数据从列表中读出数据添加分割符 new_string = "" for i in new_list: new_string += str(i)+',' #print new_string[0:-1] tel_node_new.writelines(new_string[0:-1]) count += 1 print "Execute!count=%s,current_time=%s,current_number=%s" % (count,data_time,rebulid_number) log_path = open(".//config//run_log.txt",'w') log_path.writelines("Execute!count=%s,current_time=%s,current_number=%s" % (count,data_time,rebulid_number)) time.sleep(int(self.min_circle_time)) except: errors = self.data_path if