feeds_flatstore_processor.module

<?php

/**
 * @file
 * Creates Feeds fetcher for storing Feeds results in a unique tables.
 */

/**
 * Implements hook_ctools_plugin_type().
 */
function feeds_flatstore_processor_ctools_plugin_type() {
  return array(
    'plugins' => array(
      'cache' => TRUE,
      'use hooks' => TRUE,
      'classes' => array('handler'),
    ),
  );
}

/**
 * Implements hook_form_alter().
 */
function feeds_flatstore_processor_form_alter(&$form, &$form_state, $form_id) {
  if ($form_id == 'feeds_ui_mapping_form') {
    $form['help'] = array(
      '#type' => 'item',
      '#title' => t('No Mapping Options'),
      '#markup' => t('NOTE: This processor does not have any mapping options. All items in results are saved into rows in a table.'),
    );
  }
}

/**
 * Implements hook_feeds_before_import().
 */
function feeds_flatstore_processor_feeds_before_import(FeedsSource $source) {
  if (isset($source->importer->config['processor']['config']['truncate']) && $source->importer->config['processor']['config']['truncate']) {
    // If the import it complete then we are re-importing.
    $progress_importing = $source->progressImporting();
    $count = $source->itemCount();
    if ($progress_importing == FEEDS_BATCH_COMPLETE && $count) {
      $table_name = feeds_flatstore_processor_table_name($source->id, $source->feed_nid);
      db_truncate($table_name)->execute();
    }
  }
}

/**
 * Implements hook_feeds_after_import().
 */
function feeds_flatstore_processor_feeds_after_import(FeedsSource $source) {
  // Unschedule for import if completed. TODO: only if periodic import on.
  $progress_importing = $source->progressImporting();
  $count = $source->itemCount();
  if ($progress_importing == FEEDS_BATCH_COMPLETE && $count) {
    $table_name = feeds_flatstore_processor_table_name($source->id, $source->feed_nid);
    $job = array(
      'type' => $source->id,
      'id' => $source->feed_nid,
    );
    // Remove: JobScheduler::get('feeds_source_import')->remove($job);
  }
}

/**
 * Implements hook_feeds_after_parse().
 */
function feeds_flatstore_processor_feeds_after_parse(FeedsSource $source, FeedsParserResult $result) {
  // Creates table for feed source if it doesn't exist.
  feeds_flatstore_processor_table($source, $result);
}

/**
 * Creates table name based on source id and entity id.
 */
function feeds_flatstore_processor_table_name($source_id, $entity_id, $prefix = 'feeds_datastore') {
  $name = $prefix . '_' . $source_id . '_' . $entity_id;
  drupal_alter('feeds_flatstore_processor_table_name', $name, $source_id, $entity_id);
  return $name;
}

/**
 * Retrieves table definition or creates one if table does not exist.
 */
function feeds_flatstore_processor_table(FeedsSource $source, $parser_result) {
  $table_name = feeds_flatstore_processor_table_name($source->id, $source->feed_nid);
  // If table exists, just return it.
  if ($table = data_get_table($table_name)) {
    return $table;
  }
  else {
    $title = '';
    $title_result = db_select('node', 'n')
      ->fields('n', array('title'))
      ->condition('nid', $source->feed_nid, '=')
      ->execute()
      ->fetchAssoc();
    if (isset($title_result['title'])) {
      $title = $title_result['title'];
    }
    // TODO: FIX.
    $title = $source->id;
    // If this is a CSV file, get more info.
    if ($source->importer->parser instanceof FeedsCSVParser) {
      $fetcher = get_class($source->importer->fetcher);
      $delimiter = $source->config['FeedsCSVParser']['delimiter'];
      $source_orig = $source->config[$fetcher]['source'];

      if ($source_orig) {
        // This gives us a better range of results for finding data types.
        $example_data = feeds_flatstore_processor_get_csv_results($source_orig, $delimiter);
      }
      else {
        $example_data = $parser_result;
      }
      $headers = $example_data[0];
      unset($example_data[0]);
      $types = feeds_flatstore_processor_find_columns($example_data, $limit = 500, $headers);
    }
    else {
      $types = feeds_flatstore_processor_find_columns($parser_result->items, $limit);
    }
    drupal_alter('feeds_flatstore_types', $types, $headers, $table_name);
    $schema = feeds_flatstore_processsor_base_schema();
    $field_schema = data_build_schema($types);
    feeds_flatstore_clean_reserved_words($field_schema);
    $schema['fields'] = array_merge($field_schema['fields'], $schema['fields']);
    $x = 0;
    $meta = array();
    foreach ($schema['fields'] as $name => $data) {
      if (isset($headers[$x])) {
        $meta['fields'][$name]['label'] = $headers[$x];
      }
      $x++;
    }
    $driver = db_driver();
    if ($driver == 'mysql') {
      $schema['mysql_engine'] = 'MYISAM';
    }
    // Add lat/lon fields if 'geolocation' is turned on.
    $processor = $source->importer->config['processor']['plugin_key'];
    if (module_exists('geocoder') && isset($source->config['FeedsFlatstoreProcessor']['geolocate']) && $source->config['FeedsFlatstoreProcessor']['geolocate']) {
      feeds_flatstore_processor_add_geo($schema);
      $meta['fields']['latitude']['label'] = t('lat');
      $meta['fields']['longitude']['label'] = t('lon');
    }

    $db_prefix = $GLOBALS['databases']['default']['default']['prefix'];
    $prefixed = $table_name;

    if ($db_prefix) {
      if (is_array($db_prefix) && isset($db_prefix['default'])) {
        $db_prefix = $db_prefix['default'];
      }
      $table_name = str_replace($db_prefix, "", $table_name);
      $prefixed = $db_prefix . $table_name;
    }

    $table = data_create_table($table_name, $schema, $title);
    if (!$table) {
      $table = DataTable::instance($table_name);
      $table->adopt();
    }
    $table->update(array('meta' => $meta));

    // Adding limited search with default MySQL installs.
    if ($driver == 'mysql' && $fulltext = feeds_flatstore_fulltext_fields($schema)) {
      $fulltext = explode(',', $fulltext);
      // The number of indexes depends on the MySQL version and installation.
      $fulltext = array_splice($fulltext, 0, variable_get('feeds_flatstore_mysql_index_length', 15));
      $fulltext = implode(',', $fulltext);
      Database::getConnection()->query("ALTER TABLE {$prefixed} ADD FULLTEXT INDEX $table_name" . "_index ($fulltext)");
    }
    return $table;
  }
}

/**
 * Adds geolocation fields to schema.
 */
function feeds_flatstore_processor_add_geo(&$schema) {
  $schema['fields']['latitude'] = array(
    'type' => 'float',
    'not null' => FALSE,
  );
  $schema['fields']['longitude'] = array(
    'type' => 'float',
    'not null' => FALSE,
  );
}

/**
 * Returns a string of fulltext fields from database schema.
 */
function feeds_flatstore_fulltext_fields($schema) {
  $fulltext = "";
  foreach ($schema['fields'] as $name => $elements) {
    if ($elements['type'] == 'char' || $elements['type'] == 'varchar' || $elements['type'] == 'text') {
      $fulltext = $fulltext ? $fulltext . ',' . $name : $name;
    }
  }
  return $fulltext;
}

/**
 * Cleans reserved words for mysql.
 *
 * See: http://dev.mysql.com/doc/refman/4.1/en/reserved-words.html
 */
function feeds_flatstore_clean_reserved_words(&$schema) {
  foreach ($schema['fields'] as $label => $value) {
    unset($schema['fields'][$label]);
    $schema['fields'][feeds_flatstore_processor_safe_name($label)] = $value;
  }
}

/**
 * Creates base schema for flatstore table.
 */
function feeds_flatstore_processsor_base_schema() {
  return array(
    'fields' => array(
      'feeds_flatstore_entry_id' => array(
        'type' => 'serial',
        'size' => 'normal',
        'unsigned' => TRUE,
        'not null' => TRUE,
      ),
      'timestamp' => array(
        'description' => 'The Unix timestamp for the data.',
        'type' => 'int',
        'unsigned' => TRUE,
        'not null' => FALSE,
      ),
      'feeds_entity_id' => array(
        'description' => 'The feeds entity id for the data.',
        'type' => 'int',
        'unsigned' => TRUE,
        'not null' => FALSE,
      ),
    ),
    'indexes' => array(
      'feeds_flatstore_entry_id' => array('feeds_flatstore_entry_id'),
      'timestamp' => array('timestamp'),
    ),
    'primary key' => array(
      '0' => 'feeds_flatstore_entry_id',
    ),
  );
}

/**
 * Opens csv file and grabs set amount of rows.
 *
 * This is used for finding database schema since the results of a parsed
 * feeds result will be only 50 rows by default. This is a relatively
 * inexpensive operation that should only be run once per feed source.
 */
function feeds_flatstore_processor_get_csv_results($filepath, $separator = ",", $limit = 1000) {
  static $size = 10000;
  $handle = fopen($filepath, 'r');
  ini_set('auto_detect_line_endings', TRUE);
  if ($handle === FALSE) {
    drupal_set_message(t('Could not open !filename', array('!filename' => $filepath)));
    return FALSE;
  }
  // If the field delimiter is a TAB, then use the tab character.
  if ($separator == 'TAB') {
    $separator = "\t";
  }

  $parsed_results = array();
  $rowcount = 0;
  while (($data = fgetcsv($handle, $size, $separator)) !== FALSE) {
    $result[] = $data;
    $rowcount++;
    if ($rowcount > $limit) {
      break;
    }
  }
  fclose($handle);
  return $result;
}

/**
 * Finds the schema type for values in an array.
 */
/**
 * Finds the schema type for values in an array.
 */
function feeds_flatstore_processor_find_columns($parsed_result, $limit, $headers = array()) {
  $rowcount = count($parsed_result);
  if (!$headers) {
    foreach ($parsed_result[0] as $key => $value) {
      $headers[$key] = $key;
    }
  }
  $colcount = count($headers);

  // List of possible types, from least to most specific.
  $types = array('bigtext' => 0, 'datetime' => 0, 'float' => 0, 'bigint' => 0, 'int' => 0);
  $type_counter = array();
  foreach ($headers as $key => $value) {
    $type_counter[$key]  = $types;
  }
  foreach ($parsed_result as $row) {
    foreach ($row as $key => $value) {
      if ($value == '') {
        continue;
      }

      // If a value is a number but with more than 19 chars then is treated as text.
      if (preg_match('/^-?\d+$/', $value) && strlen($value) > 9 && strlen($value) < 19) {
        $type_counter[$key]['bigint']++;
      }
      // Integer.
      elseif (preg_match('/^-?\d+$/', $value) && strlen($value) <= 9) {
        $type_counter[$key]['int']++;
      }

      // Float.
      elseif (preg_match('/^-?(?:\d+|\d*\.\d+)$/', trim($value))) {
        $type_counter[$key]['float']++;
      }

      // Datetime.
      elseif (preg_match("/\d{4}\-\d{2}\-\d{2} \d{2}:\d{2}:\d{2}/", trim($value)) ||
              preg_match("/\d{2}\-\d{2}\-\d{4} \d{2}:\d{2}:\d{2}/", trim($value)) ||
              preg_match("/\d{4}\-\d{2}\-\d{2} \d{2}:\d{2}/", trim($value)) ||
              preg_match("/\d{2}\-\d{2}\-\d{4} \d{2}:\d{2}/", trim($value)) ||
              preg_match("/\d{4}\/\d{2}\/\d{2}/", trim($value)) ||
              preg_match("/\d{2}\.\d{2}\.\d{4}/", trim($value))) {
        $type_counter[$key]['datetime']++;
      }
      else {
        $type_counter[$key]['bigtext']++;
      }
    }
  }
  $max = 0;
  foreach ($headers as $key => $header) {
    // Skip empty columns.
    if (!$header) {
      NEXT;
    }
    $max = isset($type_counter[$key]) ? max($type_counter[$key]) : '';
    if ($max != $limit) {
      // From the types that were encountered take the first one in the list.
      $type = array_keys(array_filter($type_counter[$key]));
      $result['type'][$header] = $type[0];
    }
    else {
      $result['type'][$header] = 'bigtext';
    }
  }

  $schema = $result['type'];
  return $schema;

}

/**
 * Implements hook_feeds_plugins().
 */
function feeds_flatstore_processor_feeds_plugins() {
  $path = drupal_get_path('module', 'feeds_flatstore_processor');

  return array(
    'FeedsFlatstoreProcessor' => array(
      'name' => 'Field Data Processor',
      'description' => 'Imports source data into unique database tables identified by the entity id they are attached to.',
      'help' => 'Imports source data into unique database tables identified by the entity id they are attached to.',
      'handler' => array(
        'parent' => 'FeedsProcessor',
        'class' => 'FeedsFlatstoreProcessor',
        'file' => 'FeedsFlatstoreProcessor.inc',
        'path' => $path,
      ),
    ),
  );
}

/**
 * Determines safe name using reserved words. Should move to data.
 */
function feeds_flatstore_processor_safe_name($name) {
  $map = array(
    '.' => '_',
    ':' => '',
    '/' => '',
    '-' => '_',
    ' ' => '_',
    ',' => '_',
  );
  $simple = trim(strtolower(strip_tags($name)));
  // Limit length to 64 http://dev.mysql.com/doc/refman/5.0/en/identifiers.html
  $simple = substr(strtr($simple, $map), 0, 64);

  if (is_numeric($simple)) {
    // We need to escape numerics because Drupal's drupal_write_record()
    // does not properly escape token MYSQL names.
    $simple = '__num_' . $simple;
  }

  $reserved = feeds_flatstore_reserved_words();
  if (in_array($simple, $reserved)) {
    $simple = $simple . '_';
  }

  return db_escape_table($simple);
}

/**
 * Creates list of reserved words for MySQL.
 */
function feeds_flatstore_reserved_words() {
  return array(
    // This is not a reserved word, but used by feeds flatstore as the
    // autoincrement id.
    'feeds_flatstore_entry_id',
    // This is not a reserved word, but used by feeds flatstore as the
    // itentifier for the feed entity id.
    'feeds_entity_id',
    // This is not a reserved word, but used by feeds flatstore as the
    // timestamp.
    'timestamp',
    'accessible',
    'add',
    'all',
    'alter',
    'analyze',
    'and',
    'as',
    'asc',
    'asensitive',
    'before',
    'between',
    'bigint',
    'binary',
    'blob',
    'both',
    'by',
    'call',
    'cascade',
    'case',
    'change',
    'char',
    'character',
    'check',
    'collate',
    'column',
    'condition',
    'constraint',
    'continue',
    'convert',
    'create',
    'cross',
    'current_date',
    'current_time',
    'current_timestamp',
    'current_user',
    'cursor',
    'database',
    'databases',
    'day_hour',
    'day_microsecond',
    'day_minute',
    'day_second',
    'dec',
    'decimal',
    'declare',
    'default',
    'delayed',
    'delete',
    'desc',
    'describe',
    'deterministic',
    'distinct',
    'distinctrow',
    'div',
    'double',
    'drop',
    'dual',
    'each',
    'else',
    'elseif',
    'enclosed',
    'escaped',
    'exists',
    'exit',
    'explain',
    'false',
    'fetch',
    'float',
    'float4',
    'float8',
    'for',
    'force',
    'foreign',
    'from',
    'fulltext',
    'get',
    'grant',
    'group',
    'having',
    'high_priority',
    'hour_microsecond',
    'hour_minute',
    'hour_second',
    'if',
    'ignore',
    'in',
    'index',
    'infile',
    'inner',
    'inout',
    'insensitive',
    'insert',
    'int',
    'int1',
    'int2',
    'int3',
    'int4',
    'int8',
    'integer',
    'interval',
    'into',
    'io_after_gtids',
    'io_before_gtids',
    'is',
    'iterate',
    'join',
    'key',
    'keys',
    'kill',
    'leading',
    'leave',
    'left',
    'like',
    'limit',
    'linear',
    'lines',
    'load',
    'localtime',
    'localtimestamp',
    'lock',
    'long',
    'longblob',
    'longtext',
    'loop',
    'low_priority',
    'master_bind',
    'master_ssl_verify_server_cert',
    'match',
    'maxvalue',
    'mediumblob',
    'mediumint',
    'mediumtext',
    'middleint',
    'minute_microsecond',
    'minute_second',
    'mod',
    'modifies',
    'natural',
    'not',
    'no_write_to_binlog',
    'null',
    'numeric',
    'on',
    'optimize',
    'option',
    'optionally',
    'or',
    'order',
    'out',
    'outer',
    'outfile',
    'partition',
    'precision',
    'primary',
    'procedure',
    'purge',
    'range',
    'read',
    'reads',
    'read_write',
    'real',
    'references',
    'regexp',
    'release',
    'rename',
    'repeat',
    'replace',
    'require',
    'resignal',
    'restrict',
    'return',
    'revoke',
    'right',
    'rlike',
    'schema',
    'schemas',
    'second_microsecond',
    'select',
    'sensitive',
    'separator',
    'set',
    'show',
    'signal',
    'smallint',
    'spatial',
    'specific',
    'sql',
    'sqlexception',
    'sqlstate',
    'sqlwarning',
    'sql_big_result',
    'sql_calc_found_rows',
    'sql_small_result',
    'ssl',
    'starting',
    'straight_join',
    'table',
    'terminated',
    'then',
    'tinyblob',
    'tinyint',
    'tinytext',
    'to',
    'trailing',
    'trigger',
    'true',
    'undo',
    'union',
    'unique',
    'unlock',
    'unsigned',
    'update',
    'usage',
    'use',
    'using',
    'utc_date',
    'utc_time',
    'utc_timestamp',
    'values',
    'varbinary',
    'varchar',
    'varcharacter',
    'varying',
    'when',
    'where',
    'while',
    'with',
    'write',
    'xor',
    'year_month',
    'zerofill',
  );
}