egroupware/importexport/inc/class.importexport_schedule_ui.inc.php

648 lines
19 KiB
PHP

<?php
/**
* eGroupWare
*
* @license http://opensource.org/licenses/gpl-license.php GPL - GNU General Public License
* @package importexport
* @link http://www.egroupware.org
* @author Nathan Gray
* @copyright Nathan Gray
* @version $Id: class.importexport_import_ui.inc.php 27222 2009-06-08 16:21:14Z ralfbecker $
*/
/**
* userinterface for admins to schedule imports or exports using async services
*
*/
class importexport_schedule_ui {
public $public_functions = array(
'index' => true,
'edit' => true,
);
protected static $template;
public function __construct() {
$this->template = new etemplate_new();
}
public function index($content = array()) {
if(is_array($content['scheduled']))
{
foreach($content['scheduled'] as $row)
{
if($row['delete']) {
$key = urldecode(key($row['delete']));
ExecMethod('phpgwapi.asyncservice.cancel_timer', $key);
}
}
}
$async_list = ExecMethod('phpgwapi.asyncservice.read', 'importexport%');
$data = array();
if(is_array($async_list)) {
foreach($async_list as $id => $async) {
foreach(array('errors', 'warnings', 'result') as $messages)
{
if(is_array($async['data'][$messages]))
{
$list = array();
foreach($async['data'][$messages] as $target => $message)
{
$list[] = array(
'target' => (is_numeric($target) ? '' : $target),
'message' => implode("\n", (array)$message)
);
}
$async['data'][$messages] = $list;
}
}
if($results)
{
$async['data']['result'] = $results;
}
if(is_numeric($async['data']['record_count'])) {
$async['data']['record_count'] = lang('%1 records processed', $async['data']['record_count']);
}
$data['scheduled'][] = array_merge($async['data'], array(
'id' => urlencode($id),
'next' => egw_time::server2user($async['next']),
'times' => str_replace("\n", '', print_r($async['times'], true)),
'last_run' => $async['data']['last_run'] ? egw_time::server2user($async['data']['last_run']) : ''
));
}
array_unshift($data['scheduled'], false);
}
$sel_options = self::get_select_options($data);
$this->template->read('importexport.schedule_index');
$GLOBALS['egw_info']['flags']['app_header'] = lang('Schedule import / export');
$this->template->exec('importexport.importexport_schedule_ui.index', $data, $sel_options, $readonlys, $preserve);
}
public function edit($content = array()) {
$id = $_GET['id'] ? urldecode($_GET['id']) : $content['id'];
$definition_id = $_GET['definition'];
unset($content['id']);
$data = $content;
// Deal with incoming
if($content['save'] && self::check_target($content) === true) {
unset($content['save']);
ExecMethod('phpgwapi.asyncservice.cancel_timer', $id);
$id = self::generate_id($content);
$schedule = $content['schedule'];
// Async sometimes changes minutes to an array - keep what user typed
$content['min'] = $schedule['min'];
unset($content['schedule']);
// Remove any left blank
foreach($schedule as $key => &$value) {
if($value == '') unset($schedule[$key]);
}
$result = ExecMethod2('phpgwapi.asyncservice.set_timer',
$schedule,
$id,
'importexport.importexport_schedule_ui.exec',
$content
);
if($result) {
egw_framework::refresh_opener('', 'importexport');
egw_framework::window_close();
} else {
$data['message'] = lang('Unable to schedule');
unset($id);
}
}
if($id) {
$preserve['id'] = $id;
$async = ExecMethod('phpgwapi.asyncservice.read', $id);
if(is_array($async[$id]['data'])) {
$data += $async[$id]['data'];
$data['schedule'] = $async[$id]['times'];
unset($data['times']);
// Async sometimes changes minutes to an array - show user what they typed
if(is_array($data['schedule']['min'])) $data['schedule']['min'] = $data['min'];
} else {
$data['message'] = lang('Schedule not found');
}
} else {
$data['type'] = $content['type'] ? $content['type'] : 'import';
if((int)$definition_id) {
$bo = new importexport_definitions_bo();
$definition = $bo->read($definition_id);
if($definition['definition_id']) {
$data['type'] = $definition['type'];
$data['appname'] = $definition['application'];
$data['plugin'] = $definition['plugin'];
$data['definition'] = $definition['name'];
}
}
}
if($data['target'] && $data['type']) {
$file_check = self::check_target($data);
if($file_check !== true) $data['message'] .= ($data['message'] ? "\n" . $file_check : $file_check);
}
$data['no_delete_files'] = $data['type'] != 'import';
// Current server time for nice message
$data['current_time'] = time();
$sel_options = self::get_select_options($data);
$GLOBALS['egw']->js->validate_file('.','importexport','importexport');
$GLOBALS['egw_info']['flags']['app_header'] = lang('Schedule import / export');
$this->template->read('importexport.schedule_edit');
$this->template->exec('importexport.importexport_schedule_ui.edit', $data, $sel_options, $readonlys, $preserve, 2);
}
/**
* Get options for select boxes
*/
public static function get_select_options(Array $data) {
$options = array(
'type' => array(
'import' => lang('import'),
'export' => lang('export')
)
);
(array)$apps = importexport_helper_functions::get_apps($data['type'] ? $data['type'] : 'all');
if(count($apps)) {
$options['appname'] = array('' => lang('Select one')) + array_combine($apps,$apps);
}
$plugins = importexport_helper_functions::get_plugins($data['appname'] ? $data['appname'] : 'all', $data['type']);
if(is_array($plugins)) {
foreach($plugins as $appname => $types) {
if(!is_array($types[$data['type']])) continue;
foreach($types[$data['type']] as $key => $title) {
$options['plugin'][$key] = $title;
}
}
}
$options['definition'] = array();
if($data['file'] && !is_array($data['file'])) {
$extension = substr($data['file'], -3);
}
// If the query isn't started with something, bodefinitions won't load the definitions
$query = array();
$query['type'] = $data['type'];
$query['application'] = $data['application'];
$query['plugin'] = $data['plugin'];
$definitions = new importexport_definitions_bo($query);
foreach ((array)$definitions->get_definitions() as $identifier) {
try {
$definition = new importexport_definition($identifier);
} catch (Exception $e) {
// permission error
continue;
}
if ($title = $definition->get_title()) {
$options['definition'][$title] = $title;
}
unset($definition);
}
unset($definitions);
return $options;
}
/**
* Generate a async key
*/
public static function generate_id($data) {
$query = array(
'name' => $data['definition']
);
$definitions = new importexport_definitions_bo($query);
$definition_list = ((array)$definitions->get_definitions());
$id = 'importexport.'.$definition_list[0].'.'.$data['target'];
return $id;
}
/**
* Get plugins via ajax
*/
public function ajax_get_plugins($type, $appname, &$response = null) {
if($response) {
$return = false;
} else {
$response = new xajaxResponse();
}
$options = self::get_select_options(array('type' => $type, 'appname'=>$appname));
$response->addScript("clear_options('exec[plugin]');");
$response->addScript("selectbox_add_option('exec[plugin]','".lang('Select...')."', '',false);");
if(is_array($options['plugin'])) {
foreach ($options['plugin'] as $value => $title) {
$response->addScript("selectbox_add_option('exec[plugin]','$title', '$value',false);");
}
}
if(count($options['plugin']) == 1) {
$this->ajax_get_definitions($appname, $value, $response);
$response->assign('exec[plugin]','value',$value);
} else {
$response->addScript("xajax_doXMLHTTP('importexport.importexport_schedule_ui.ajax_get_definitions', '$appname', document.getElementById('exec[plugin]').value);");
}
// Trigger chosen
$response->addScript("\$j(document.getElementById('exec[plugin]')).trigger('liszt:updated');");
return $response->getXML();
}
/**
* Get definitions via ajax
*/
public function ajax_get_definitions($appname, $plugin, &$response = null) {
$options = self::get_select_options(array('appname'=>$appname, 'plugin'=>$plugin));
if($response) {
$return = false;
} else {
$response = new xajaxResponse();
}
$response->addScript("clear_options('exec[definition]');");
$response->addScript("selectbox_add_option('exec[definition]','".lang('Select...')."', '',false);");
if(is_array($options['definition'])) {
foreach ($options['definition'] as $value => $title) {
$response->addScript("selectbox_add_option('exec[definition]','$title', '$value',false);");
}
}
if(count($options['definition']) == 1) {
$response->assign('exec[definition]','value',$value);
} else {
$response->addScript("document.getElementById('exec[definition]').value = ''");
}
// Trigger chosen
$response->addScript("\$j(document.getElementById('exec[definition]')).trigger('liszt:updated');");
return $response->getXML();
}
/**
* Check that the target is valid for the type (readable or writable)
* and that they're not trying to write directly to the filesystem
*
* $data should contain target & type
*/
public static function check_target(Array $data) {
$scheme = parse_url($data['target'], PHP_URL_SCHEME);
if($scheme == '' || $scheme == 'file') {
return 'Direct file access not allowed';
}
if($scheme == EGroupware\Api\Vfs::SCHEME && !in_array(EGroupware\Api\Vfs::SCHEME, stream_get_wrappers())) {
stream_wrapper_register(EGroupware\Api\Vfs::SCHEME, 'vfs_stream_wrapper', STREAM_IS_URL);
}
if ($data['type'] == 'import' && ($scheme == EGroupware\Api\Vfs::SCHEME && !egw_vfs::is_readable($data['target'])))
{
return lang('%1 is not readable',$data['target']);
}
elseif ($data['type'] == 'import' && in_array($scheme, array('http','https')))
{
// Not supported by is_readable, try headers...
$options = array();
stream_context_set_default(array('http'=>array(
'method' => 'HEAD',
'ignore_errors' => 1
)));
$headers = get_headers($data['target'],1);
// Reset...
stream_context_set_default(array('http'=>array(
'method' => 'GET',
'ignore_errors' => 0
)));
// Response code has an integer key, but redirects may add more responses
for($i = 0; $i < count($headers); $i++)
{
if(!$headers[$i]) break;
if(strpos($headers[$i],'200') !== false) return true;
}
return lang('%1 is not readable',$data['target']);
}
elseif ($data['type'] == 'export' && !self::is__writable($data['target'])) {
return lang('%1 is not writable',$data['target']);
}
return true;
}
/**
* Writable that checks the folder too, in case the file does not exist yet
* http://ca3.php.net/manual/en/function.is-writable.php#73596
*
* @param path Path to check
*/
private static function is__writable($path) {
if ($path{strlen($path)-1}=='/') // recursively return a temporary file path
return self::is__writable($path.uniqid(mt_rand()).'.tmp');
else if (is_dir($path))
return self::is__writable($path.'/'.uniqid(mt_rand()).'.tmp');
// check tmp file for read/write capabilities
$rm = file_exists($path);
$f = @fopen($path, 'a');
if ($f===false)
return false;
fclose($f);
if (!$rm)
@unlink($path);
return true;
}
/**
* Execute a scheduled import or export
*/
public static function exec($data)
{
ob_start();
$data['record_count'] = 0;
unset($data['errors']);
unset($data['warnings']);
unset($data['result']);
if($data['lock'])
{
// Lock expires
if($data['lock'] < time())
{
unset($data['lock']);
$data['warnings'][][] = lang('Lock expired on previous run');
}
else
{
// Still running
ob_end_flush();
return;
}
}
$data['last_run'] = time();
// Lock job for an hour to prevent multiples overlapping
$data['lock'] = time() + 3600;
self::update_job($data, true);
// check file
$file_check = self::check_target($data);
if($file_check !== true) {
$data['errors'] = array($file_check=>'');
// Update job with results
self::update_job($data);
error_log('importexport_schedule: ' . date('c') . ": $file_check \n");
error_log(ob_get_flush());
return;
}
$definition = new importexport_definition($data['definition']);
if( $definition->get_identifier() < 1 ) {
$data['errors'] = array('Definition not found!');
// Update job with results
self::update_job($data);
error_log('importexport_schedule: ' . date('c') . ": Definition not found! \n");
return;
}
$GLOBALS['egw_info']['flags']['currentapp'] = $definition->application;
$po = new $definition->plugin;
$type = $data['type'];
if(is_dir($data['target']))
{
if($data['type'] == 'import')
{
$targets = array();
foreach(scandir($data['target']) as $target)
{
if ($target == '.' || $target == '..') continue;
$target = $data['target'].(substr($data['target'],-1) == '/' ? '' : '/').$target;
// Check modification time, make sure it's not currently being written
// Skip files modified in the last 10 seconds
$mod_time = filemtime($target);
if($mod_time >= time() - 10)
{
$data['result'][$target] = lang('Skipped');
continue;
}
$targets[$mod_time.$target] = $target;
}
if($targets)
{
ksort($targets);
}
}
else
{
// Create a unique file for export
$targets = array($data['target'].uniqid($definition->name).'.'.$po->get_filesuffix());
}
}
else
{
$targets = array($data['target']);
}
if($type == 'export')
{
// Set to export all or filter, if set
$selection = array('selection' => 'all');
if($definition->filter)
{
$fields = importexport_helper_functions::get_filter_fields($definition->application, $po);
$selection = array('selection' => 'filter');
$filters = array();
foreach($definition->filter as $field => $value)
{
// Handle multiple values
if(!is_array($value) && strpos($value,',') !== false) $value = explode(',',$value);
$filters[$field] = $value;
// Process relative dates into the current absolute date
if($filters[$field] && strpos($fields[$field]['type'],'date') === 0)
{
$filters[$field] = importexport_helper_functions::date_rel2abs($value);
}
}
// Update filter to use current absolute dates
$definition->filter = $filters;
}
if(!is_array($definition->plugin_options))
{
$definition->plugin_options = array();
}
$definition->plugin_options = array_merge($definition->plugin_options, $selection);
}
foreach($targets as $target)
{
// Update lock timeout
$data['lock'] = time() + 3600;
self::update_job($data, true);
$resource = null;
try {
if($resource = @fopen( $target, $data['type'] == 'import' ? 'rb' : 'wb' )) {
$result = $po->$type( $resource, $definition );
fclose($resource);
} else {
error_log('importexport_schedule: ' . date('c') . ": File $target not readable! \n");
$data['errors'][$target][] = lang('%1 is not readable',$target);
}
}
catch (Exception $i_ex)
{
fclose($resource);
$data['errors'][$target][] = $i_ex->getMessage();
}
if(method_exists($po, 'get_warnings') && $po->get_warnings()) {
$buffer = 'importexport_schedule: ' . date('c') . ": Import warnings:\n#\tWarning\n";
foreach($po->get_warnings() as $record => $msg) {
$data['warnings'][$target][] = "#$record: $msg";
$buffer += "$record\t$msg\n";
}
error_log($buffer);
}
if(method_exists($po, 'get_errors') && $po->get_errors()) {
$buffer = 'importexport_schedule: ' . date('c') . ": Import errors:\n#\tError\n";
foreach($po->get_errors() as $record => $error) {
$data['errors'][$target][] = "#$record: $error";
$buffer += "$record\t$error\n";
}
error_log($buffer);
}
if($po instanceof importexport_iface_import_plugin) {
if(is_numeric($result)) {
$data['record_count'] += $result;
$data['result'][$target][] = lang('%1 records processed', $result);
}
$data['result'][$target] = array();
foreach($po->get_results() as $action => $count) {
$data['result'][$target][] = lang($action) . ": $count";
}
} else {
if($result instanceof importexport_iface_export_record)
{
$data['record_count'] += $result->get_num_of_records();
$data['result'][$target][] = lang('%1 records processed', $result->get_num_of_records());
}
}
}
// Delete file?
if($data['delete_files'] && $type == 'import' && !$data['errors'])
{
foreach($targets as $target)
{
if(unlink($target))
{
$data['result'][$target][] .= "\n..." . lang('deleted');
}
else
{
$data['errors'][$target][] .= "\n..." . lang('Unable to delete');
}
}
}
// Run time in minutes
$data['run_time'] = round((time() - $data['last_run']) / 60,1);
// Clear lock
$data['lock'] = 0;
// Update job with results
self::update_job($data);
$contents = ob_get_contents();
// Log to error log
if($contents)
{
error_log('importexport_schedule: ' . date('c') . ": \n".$contents);
}
ob_end_clean();
}
/**
* Update the async job with current status, and send a notification
* to user if there were any errors.
*/
private static function update_job($data, $no_notification = false) {
$id = self::generate_id($data);
$async = ExecMethod('phpgwapi.asyncservice.read', $id);
$async = $async[$id];
if(is_array($async)) {
ExecMethod('phpgwapi.asyncservice.cancel_timer', $id);
$result = ExecMethod2('phpgwapi.asyncservice.set_timer',
$async['times'],
$id,
'importexport.importexport_schedule_ui.exec',
$data
);
}
if($no_notification) return $result;
// Send notification to user
if($data['warnings'] || $data['errors'])
{
$notify = new notifications();
$notify->set_sender($data['account_id']);
$notify->add_receiver($data['account_id']);
$notify->set_subject(lang('Schedule import | export'). ' ' . lang('errors'));
$contents = '';
if($data['warnings'])
{
$contents .= lang($data['type']) . ' ' . lang('Warnings') . ' ' . egw_time::to() . ':';
foreach($data['warnings'] as $target => $message)
{
$contents .= "\n". (is_numeric($target) ? '' : $target."\n");
$contents .= is_array($message) ? implode("\n",$message) : $message;
}
$contents .= "\n";
}
if($data['errors'])
{
$contents .= lang($data['type']) . ' ' . lang('Errors') . ' ' . egw_time::to() . ':';
foreach($data['errors'] as $target => $errors)
{
$contents .= "\n". (is_numeric($target) ? '' : $target."\n");
$contents .= is_array($errors) ? implode("\n",$errors) : $errors;
}
$contents .= "\n";
}
$notify->set_message($contents);
$notify->send();
}
return $result;
}
}
?>