generate.py

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

  USAGE:
    [year, month, day] = epoch_to_date(daynum)

  ARGUMENTS:
    daynum  A integer giving the epoch day (0 = 1 January 1900)

  DESCRIPTION:
    Function for converting a number of days (integer value) since epoch time
    1 January 1900 (integer value) into a date tuple [day, month, year].

  EXAMPLES:
    [day, month, year] = epoch_to_date(0)      # returns ['01','01','1900']
    [day, month, year] = epoch_to_date(37734)  # returns ['25','04','2003']
  """

  if (not (isinstance(daynum, int) or isinstance(daynum, long))):
    print 'Error: Input value for "daynum" is not of integer type: %s' % \
          (str(daynum))
    raise Exception

  if (daynum >= -115860):
    year = 1600 + int(math.floor((daynum + 109573) / 365.2425))
  elif (daynum >= -693597):
    year = 4 + int(math.floor((daynum + 692502) / 365.2425))
  else:
    year = -4 + int(math.floor((daynum+695058) / 365.2425))

  days = daynum - first_day_of_year(year) + 1

  if (days <= 0):
    year -= 1
    days = daynum - first_day_of_year(year) + 1

  days_in_year = 365 + is_leap_year(year)  # Adjust for a leap year

  if (days > days_in_year):
    year += 1
    days = daynum - first_day_of_year(year) + 1

  # Add 10 days for dates between 15 October 1582 and 31 December 1582
  #
  if (daynum >= -115860) and (daynum <= -115783):
    days += 10

  day_count = 0
  month = 12
  leap_year_flag = is_leap_year(year)

  for m in range(12):
    day_count += days_in_month[leap_year_flag][m]
    if (day_count >= days):
      month = m + 1
      break

  # Add up the days in the prior months
  #
  prior_month_days = 0
  for m in range(month-1):
    prior_month_days += days_in_month[leap_year_flag][m]

  day = days - prior_month_days

  day_str =   string.zfill(str(day),2)  # Add '0' if necessary
  month_str = string.zfill(str(month),2)  # Add '0' if necessary
  year_str =  str(year)  # Is always four digits long

  return [day_str, month_str, year_str]

# -----------------------------------------------------------------------------

def date_to_epoch(day, month, year):
  """ Convert a date [day, month, year] into an epoch day number.
     (based on a function from the 'normalDate.py' module by Jeff Bauer, see:
     http://starship.python.net/crew/jbauer/normalDate/)

  USAGE:
    daynum = date_to_epoch(year, month, day)

  ARGUMENTS:
    day    Day value (string or integer number)
    month  Month value (string or integer number)
    year   Year value (string or integer number)

  DESCRIPTION:
    Function for converting a date into a epoch day number (integer value)
    since 1 january 1900.

  EXAMPLES:
    day = date_to_epoch('01', '01', '1900')  # returns 0
    day = date_to_epoch('25', '04', '2003')  # returns 37734
  """

  # Convert into integer values
  #
  try:
    day_int = int(day)
  except:
    print 'Error: "day" value is not an integer'
    raise Exception
  try:
    month_int = int(month)
  except:
    print 'Error: "month" value is not an integer'
    raise Exception
  try:
    year_int = int(year)
  except:
    print 'Error: "year" value is not an integer'
    raise Exception

  # Test if values are within range
  #
  if (year_int <= 1000):
    print 'Error: Input value for "year" is not a positive integer ' + \
          'number: %i' % (year)
    raise Exception

  leap_year_flag = is_leap_year(year_int)

  if (month_int <= 0) or (month_int > 12):
    print 'Error: Input value for "month" is not a possible day number: %i' % \
          (month)
    raise Exception

  if (day_int <= 0) or (day_int > days_in_month[leap_year_flag][month_int-1]):
    print 'Error: Input value for "day" is not a possible day number: %i' % \
          (day)
    raise Exception

  days = first_day_of_year(year_int) + day_int - 1

  for m in range(month_int-1):
    days += days_in_month[leap_year_flag][m]

  if (year_int == 1582):
    if (month_int > 10) or ((month_int == 10) and (day_int > 4)):
      days -= 10

  return days

# -----------------------------------------------------------------------------

def load_misspellings_dict(misspellings_file_name):
  """Load a look-up table containing misspellings for common words, which can
     be used to introduce realistic errors.

     Returns a dictionary where the keys are the correct spellings and the
     values are a list of one or more misspellings.
  """

  # Open file and read all lines into a list
  #
  try:
    f = open(misspellings_file_name, 'r')
  except:
    print 'Error: Can not read from misspellings file "%s"' % \
          (misspellings_file_name)
    raise IOError

  file_data = f.readlines()  # Read complete file
  f.close()

  misspell_dict = {}

  key = None  # Start with a non-existing eky word (correct word)

  # Now process all lines - - - - - - - - - - - - - - - - - - - - - - - - - - -
  #
  for line in file_data:
    l = line.strip()  # Remove line separators

    if (len(l) > 0) and (l[0] != '#'):  # Not empty line and not comment

      ll = l.split(':')  # Separate key from values

      if (ll[0] == '') and (len(ll) > 1):
        ll = ll[1:]

      if (len(ll) == 2):  # Line contains a key - - - - - - - - - - - - - - - -

        key = ll[0].strip().lower()  # Get key, make lower and strip spaces

        if (key == ''):
          print 'This should not happen: "%s"' % (l)
          raise Exception

        vals = ll[1].strip().lower() # Get values in a string

        if (vals == ''):
          print 'Error: No misspellings given for "%s" in line: "%s"' % \
                (key, l)
          raise Exception

        val_list = vals.split(',')
        val_set = sets.Set()
        for val in val_list:
          if (val != ''):
            val_set.add(val.strip())  # Remove all spaces

        # Check that all misspellings are different from the original
        #
        if (key in val_set):
          print 'Error: A misspelling is the same as the original value' + \
                ' "%s" in line: "%s"' % (key, l)
          raise Exception

        # Now insert into misspellings dictionary
        #
        key_val_set = misspell_dict.get(key, sets.Set())
        key_val_set = key_val_set.union(val_set)
        misspell_dict[key] = key_val_set

      elif (len(ll) == 1):  # Line contains only values - - - - - - - - - - - -

        if (key == None):
          print 'Error: No key (correct word) defined in line: "%s"' % (l)
          raise Exception

        vals = ll[0].lower() # Get values in a string
        val_list = vals.split(',')

        val_set = sets.Set()
        for val in val_list:
          if (val != ''):
            val_set.add(val.strip())  # Remove all spaces

        # Check that all misspellings are different from the original
        #
        if (key in val_set):
          print 'Error: A misspelling is the same as the original value' + \
                ' "%s" in line: "%s"' % (key, l)
          raise Exception

        # Now insert into misspellings dictionary
        #
        key_val_set = misspell_dict.get(key, sets.Set())
        key_val_set = key_val_set.union(val_set)
        misspell_dict[key] = key_val_set

      else:
        print 'error:Illegal line format in line: "%s"' % (l)
        raise Exception

  # Now convert all sets into lists - - - - - - - - - - - - - - - - - - - - -
  #
  for k in misspell_dict:
    misspell_dict[k] = list(misspell_dict[k])

  # print '  Length of misspellings dictionary: %d' % (len(misspell_dict))

  return misspell_dict

# -----------------------------------------------------------------------------

def random_select(prob_dist_list):
  """Randomly select one of the list entries (tuples of value and probability
     values).
  """

  rand_num = random.random()  # Random number between 0.0 and 1.0

  ind = -1
  while (prob_dist_list[ind][1] > rand_num):
    ind -= 1

  return prob_dist_list[ind][0]

# =============================================================================
# Start main program

if (len(sys.argv) != 8):
  print 'Seven arguments needed with %s:' % (sys.argv[0])
  print '  - Output file name'
  print '  - Number of original records'
  print '  - Number of duplicate records'
  print '  - Maximal number of duplicate records for one original record'
  print '  - Maximum number of modifications per field'
  print '  - Maximum number of modifications per record'
  print '  - Probability distribution for duplicates (uniform, poisson, zipf)'
  print 'All other parameters have to be set within the code'
  sys.exit()

output_file =           sys.argv[1]
num_org_records =       int(sys.argv[2])
num_dup_records =       int(sys.argv[3])
max_num_dups =          int(sys.argv[4])
max_num_field_modifi =  int(sys.argv[5])
max_num_record_modifi = int(sys.argv[6])
prob_distribution =     sys.argv[7][:3]

if (num_org_records <= 0):
  print 'Error: Number of original records must be positive'
  sys.exit()

if (num_dup_records < 0):
  print 'Error: Number of duplicate records must be zero or positive'
  sys.exit()

if (max_num_dups <= 0):
  print 'Error: Maximal number of duplicates per record must be positive'
  sys.exit()

if (max_num_field_modifi <= 0):
  print 'Error: Maximal number of modifications per field must be positive'
  sys.exit()

if (max_num_record_modifi <= 0):
  print 'Error: Maximal number of modifications per record must be positive'
  sys.exit()

if (max_num_record_modifi < max_num_field_modifi):
  print 'Error: Maximal number of modifications per record must be equal to'
  print '       or larger than maximal number of modifications per field'
  sys.exit()

if (prob_distribution not in ['uni', 'poi', 'zip']):
  print 'Error: Illegal probability distribution: %s' % (sys.argv[7])
  print '       Must be one of: "uniform", "poisson", or "zipf"'
  sys.exit()

# -----------------------------------------------------------------------------
# Check all user options within generate.py for validity
#
field_names = []  # Make a list of all field names

# A list of all probabilities to check ('select_prob' is checked separately)
#
prob_names = ['ins_prob','del_prob','sub_prob','trans_prob','val_swap_prob',
              'wrd_swap_prob','spc_ins_prob','spc_del_prob','miss_prob',
              'misspell_prob','new_val_prob']

select_prob_sum = 0.0  # Sum over all select probabilities

# Check if all defined field dictionaries have the necessary keys
#
i = 0  # Loop counter
for field_dict in field_list:

  if ('name' not in field_dict):
    print 'Error: No field name given for field dictionary'
    raise Exception
  elif (field_dict['name'] == 'rec_id'):
    print 'Error: Illegal field name "rec_id" (used for record identifier)'
    raise Exception
  else: