generate.py

来自「这是一个用与生成数据库测试文件的程序对于网站的测试有很大的帮助」· Python 代码 · 共 1,725 行 · 第 1/5 页

org_rec_used = {}  # Dictionary with record IDs of original records used to
                   # create duplicates

rec_cnt = 0  # Record counter

while (rec_cnt < num_dup_records):

  # Find an original record that has so far not been used to create - - - - - -
  # duplicates
  #
  rand_rec_num = random.randint(0, num_org_records)
  org_rec_id = 'rec-%i-org' % (rand_rec_num)

  while (org_rec_id in org_rec_used) or (org_rec_id not in org_rec):
    rand_rec_num = random.randint(0, num_org_records)  # Get new record number
    org_rec_id = 'rec-%i-org' % (rand_rec_num)

  # Randomly choose how many duplicates to create from this record
  #
  num_dups = random_select(prob_dist_list)

  if (VERBOSE_OUTPUT == True):
    print '  Use record %s to create %i duplicates' % (org_rec_id, num_dups)

  org_rec_dict = org_rec[org_rec_id]  # Get the original record

  d = 0  # Loop counter for duplicates for this record

  # Loop to create duplicate records - - - - - - - - - - - - - - - - - - - - -
  #
  while (d < num_dups) and (rec_cnt < num_dup_records):

    # Create a duplicate of the original record
    #
    dup_rec_dict = org_rec_dict.copy()  # Make a copy of the original record
    dup_rec_id =             'rec-%i-dup-%i' % (rand_rec_num, d)
    dup_rec_dict['rec_id'] = dup_rec_id

    num_modif_in_record = 0  # Count the number of modifications in this record

    # Set the field modification counters to zero for all fields
    #
    field_mod_count_dict = {}

    for field_dict in field_list:
      field_mod_count_dict[field_dict['name']] = 0

    # Do random swapping between fields if two or more modifications in record
    #
    if (max_num_record_modifi > 1):

      # - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
      # Random swapping of values between a pair of field values
      #
      field_swap_pair_list = field_swap_prob.keys()
      random.shuffle(field_swap_pair_list)

      for field_pair in field_swap_pair_list:

        if (random.random() <= field_swap_prob[field_pair]) and \
           (num_modif_in_record <= (max_num_record_modifi-2)):

          fname_a, fname_b = field_pair

          # Make sure both fields are in the record dictionary
          #
          if (fname_a in dup_rec_dict) and (fname_b in dup_rec_dict):
            fvalue_a = dup_rec_dict[fname_a]
            fvalue_b = dup_rec_dict[fname_b]

            dup_rec_dict[fname_a] = fvalue_b  # Swap field values
            dup_rec_dict[fname_b] = fvalue_a

            num_modif_in_record += 2

            field_mod_count_dict[fname_a] = field_mod_count_dict[fname_a] + 1
            field_mod_count_dict[fname_b] = field_mod_count_dict[fname_b] + 1

            if (VERBOSE_OUTPUT == True):
              print '    Swapped fields "%s" and "%s": "%s" <-> "%s"' % \
                    (fname_a, fname_b, fvalue_a, fvalue_b)

    # - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
    # Now introduce modifications up to the given maximal number

    while (num_modif_in_record < max_num_record_modifi):

      # Randomly choose a field
      #
      field_dict = random_select(select_prob_list)
      field_name = field_dict['name']

      # Make sure this field hasn't been modified already
      #
      while (field_mod_count_dict[field_name] == max_num_field_modifi):
        field_dict = random_select(select_prob_list)
        field_name = field_dict['name']

      if (field_dict['char_range'] == 'digit'):
        field_range = string.digits
      elif (field_dict['char_range'] == 'alpha'):
        field_range = string.lowercase
      elif (field_dict['char_range'] == 'alphanum'):
        field_range = string.digits+string.lowercase

      # - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
      # Randomly select the number of modifications to be done in this field
      # (and make sure we don't too many modifications in the record)
      #
      if (max_num_field_modifi == 1):
        num_field_mod_to_do = 1
      else:
        num_field_mod_to_do = random.randint(1, max_num_field_modifi)

      num_rec_mod_to_do = max_num_record_modifi - num_modif_in_record

      if (num_field_mod_to_do > num_rec_mod_to_do):
        num_field_mod_to_do = num_rec_mod_to_do

      if (VERBOSE_OUTPUT == True):
        print '    Choose field "%s" for %d modification' % \
              (field_name, num_field_mod_to_do)

      num_modif_in_field = 0  # Count the number of modifications in this field

      org_field_val = org_rec_dict.get(field_name, None) # Get original value

      # Loop over chosen number of modifications - - - - - - - - - - - - - - -
      #
      for m in range(num_field_mod_to_do):

        # Randomly choose a modification
        #
        mod_op = random_select(field_dict['prob_list'])

        old_field_val = dup_rec_dict.get(field_name, None)
        dup_field_val = old_field_val  # Modify this value

        # ---------------------------------------------------------------------
        # Do the selected modification

        # - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
        # Randomly choose a misspelling if the field value is found in the
        # misspellings dictionary
        #
        if (mod_op == 'misspell_prob') and ('misspell_dict' in field_dict) \
           and (old_field_val in field_dict['misspell_dict']):

          misspell_list = field_dict['misspell_dict'][old_field_val]

          if (len(misspell_list) == 1):
            dup_field_val = misspell_list[0]

          else:  # Randomly choose a value
            dup_field_val = random.choice(misspell_list)

          if (VERBOSE_OUTPUT == True):
            print '      Exchanged value "%s" in field "%s" with "%s"' % \
                  (old_field_val, field_name, dup_field_val) + \
                  ' from misspellings dictionary'

        # - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
        # Randomly exchange of a field value with another value
        #
        elif (mod_op == 'val_swap_prob') and (old_field_val != None):

          if (field_dict['type'] == 'freq'):  # A frequency file based field
            rand_num = random.randint(0, freq_files_length[field_name]-1)
            dup_field_val = freq_files[field_name][rand_num]

          elif (field_dict['type'] == 'date'):  # A date field
            rand_num = random.randint(field_dict['start_epoch'], \
                                      field_dict['end_epoch']-1)
            rand_date = epoch_to_date(rand_num)
            dup_field_val = rand_date[2]+rand_date[1]+rand_date[0]

          elif (field_dict['type'] == 'phone'):  # A phone number field
            area_code = random.choice(field_dict['area_codes'])
            max_digit = int('9'*field_dict['num_digits'])
            min_digit = int('1'*(int(1+round(field_dict['num_digits']/2.))))
            rand_num = random.randint(min_digit, max_digit)
            dup_field_val = area_code+' '+ \
                            str(rand_num).zfill(field_dict['num_digits'])

          elif (field_dict['type'] == 'ident'):  # A identification numb. field
            rand_num = random.randint(field_dict['start_id'], \
                                      field_dict['end_id']-1)
            dup_field_val = str(rand_num)

          if (dup_field_val != old_field_val):

            if (VERBOSE_OUTPUT == True):
              print '      Exchanged value in field "%s": "%s" -> "%s"' % \
                         (field_name, old_field_val, dup_field_val)

        # - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
        # Randomly set to missing value
        #
        elif (mod_op == 'miss_prob') and (old_field_val != None):

          dup_field_val = missing_value  # Set to a missing value

          if (VERBOSE_OUTPUT == True):
            print '      Set field "%s" to missing value: "%s" -> "%s"' % \
                      (field_name, old_field_val, dup_field_val)

        # - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
        # Randomly swap two words if the value contains at least two words
        #
        elif (mod_op == 'wrd_swap_prob') and (old_field_val != None) and \
             (' ' in old_field_val):

          # Count number of words
          #
          word_list = old_field_val.split(' ')
          num_words = len(word_list)

          if (num_words == 2):  # If only 2 words given
            swap_index = 0
          else:  # If more words given select position randomly
            swap_index = random.randint(0, num_words-2)

          tmp_word =                word_list[swap_index]
          word_list[swap_index] =   word_list[swap_index+1]
          word_list[swap_index+1] = tmp_word

          dup_field_val = ' '.join(word_list)

          if (dup_field_val != old_field_val):

            if (VERBOSE_OUTPUT == True):
              print '      Swapped words in field "%s": "%s" -> "%s"' % \
                    (field_name, old_field_val, dup_field_val)

        # - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
        # Randomly create a new value if the field value is empty (missing)
        #
        elif (mod_op == 'new_val_prob') and (old_field_val == None):

          if (field_dict['type'] == 'freq'):  # A frequency file based field
            rand_num = random.randint(0, freq_files_length[field_name]-1)
            dup_field_val = freq_files[field_name][rand_num]

          elif (field_dict['type'] == 'date'):  # A date field
            rand_num = random.randint(field_dict['start_epoch'], \
                                      field_dict['end_epoch']-1)
            rand_date = epoch_to_date(rand_num)
            dup_field_val = rand_date[2]+rand_date[1]+rand_date[0]

          elif (field_dict['type'] == 'phone'):  # A phone number field
            area_code = random.choice(field_dict['area_codes'])
            max_digit = int('9'*field_dict['num_digits'])
            min_digit = int('1'*(int(1+round(field_dict['num_digits']/2.))))
            rand_num = random.randint(min_digit, max_digit)
            dup_field_val = area_code+' '+ \
                            str(rand_num).zfill(field_dict['num_digits'])

          elif (field_dict['type'] == 'ident'):  # A identification number
            rand_num = random.randint(field_dict['start_id'], \
                                      field_dict['end_id']-1)
            dup_field_val = str(rand_num)

          if (VERBOSE_OUTPUT == True):
            print '      Exchanged missing value "%s" in field "%s" with "%s"'\
                  % (missing_value, field_name, dup_field_val)

        # - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
        # Random substitution of a character
        #
        elif (mod_op == 'sub_prob') and (old_field_val != None):

          # Get an substitution position randomly
          #
          rand_sub_pos = error_position(dup_field_val, 0)

          if (rand_sub_pos != None):  # If a valid position was returned

            old_char = dup_field_val[rand_sub_pos]
            new_char = error_character(old_char, field_dict['char_range'])

            new_field_val = dup_field_val[:rand_sub_pos] + new_char + \
                            dup_field_val[rand_sub_pos+1:]

            if (new_field_val != dup_field_val):
              dup_field_val = new_field_val

              if (VERBOSE_OUTPUT == True):
                print '      Substituted character "%s" with "%s" in field ' \
                      % (old_char, new_char) + '"%s": "%s" -> "%s"' % \
                      (field_name, old_field_val, dup_field_val)

        # - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
        # Random insertion of a character
        #
        elif (mod_op == 'ins_prob') and (old_field_val != None):

          # Get an insert position randomly
          #
          rand_ins_pos = error_position(dup_field_val, +1)
          rand_char =    random.choice(field_range)

          if (rand_ins_pos != None):  # If a valid position was returned
            dup_field_val = dup_field_val[:rand_ins_pos] + rand_char + \
                            dup_field_val[rand_ins_pos:]

            if (VERBOSE_OUTPUT == True):
              print '      Inserted char "%s" into field "%s": "%s" -> "%s"' \
                    % (rand_char, field_name, old_field_val, dup_field_val)

        # - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
        # Random deletion of a character
        #
        elif (mod_op == 'del_prob') and (old_field_val != None) and \
             (len(old_field_val) > 1):  # Field must have at least 2 characters

          # Get a delete position randomly
          #
          rand_del_pos = error_position(dup_field_val, 0)

          del_char = dup_field_val[rand_del_pos]

          dup_field_val = dup_field_val[:rand_del_pos] + \
                          dup_field_val[rand_del_pos+1:]

          if (VERBOSE_OUTPUT == True):
            print '      Deleted character "%s" in field "%s": "%s" -> "%s"' \
                  % (del_char, field_name, old_field_val, dup_field_val)

        # - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
        # Random transposition of two characters
        #
        elif (mod_op == 'trans_prob') and (old_field_val != None) and \
             (len(dup_field_val) > 1):  # Field must have at least 2 characters

          # Get a transposition position randomly
          #
          rand_trans_pos = error_position(dup_field_val, -1)

          trans_chars = dup_field_val[rand_trans_pos:rand_trans_pos+2]
          trans_chars2 = trans_chars[1] + trans_chars[0]  # Do transposition

          new_field_val = dup_field_val[:rand_trans_pos] + trans_chars2 + \
                          dup_field_val[rand_trans_pos+2:]

          if (new_