datasource.pm

普通的ETL工具

# vim:ts=4 sw=4
# ----------------------------------------------------------------------------------------------------
#  Name		: ETL::Pequel3::Type::DataSource.pm
#  Created	: 7 September 2006
#  Author	: Mario Gaffiero (gaffie)
#
# Copyright 1999-2007 Mario Gaffiero.
# 
# This file is part of Pequel(TM).
# 
# Pequel is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; version 2 of the License.
# 
# Pequel is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
# 
# You should have received a copy of the GNU General Public License
# along with Pequel; if not, write to the Free Software
# Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
# ----------------------------------------------------------------------------------------------------
# Modification History
# When          Version     Who     What
# ----------------------------------------------------------------------------------------------------
package ETL::Pequel3::Type::DataSource;
require 5.005_62;
use strict;
use warnings;
use ETL::Pequel3::Type::DataMember;
# ----------------------------------------------------------------------------------------------------
{
	package ETL::Pequel3::Type::DataSource::Abstract;
	use base qw(Class::STL::Element);
	use Class::STL::ClassMembers 
		qw( 
			err 
			properties 
			datasource_name 
			fdname 
			io_name
			vname
			sorter 
			shared 
			ds_vname 
			target_name
			pequel_ref
			dataset
			description
		), 
		ETL::Pequel3::Type::DataMember::User->new(name => 'record_delimiter', default => '\n');
	use Class::STL::ClassMembers::Constructor;
	sub new_extra
	{
		my $self = shift;
		use ETL::Pequel3::Error;
		$self->err(ETL::Pequel3::Error->new());
		use ETL::Pequel3::Type::Properties;
		$self->properties(ETL::Pequel3::Type::Properties->new());
		$self->target_name($self->fdname()) if (defined($self->fdname()));
		$self->vname($self->fdname()) if (defined($self->fdname()));
		$self->vname
		(
			'_' 
			. uc(substr($self->io_name(),0,1)) 
			. '_' 
			. do{ my $f=uc($self->target_name());$f=~s/\W+/_/g;$f}
		) unless (defined($self->vname()));
		$self->fdname(($self->shared() ? '::' : '') . $self->vname()) unless (defined($self->fdname()));
		$self->ds_vname('_DS' . $self->vname());
		return $self;
	}
	sub _map_output
	{
		my $self = shift;
		my %p = @_;
	}
	sub _code_init
	{
		my $self = shift;
		my $c = shift || ETL::Pequel3::CodeStyler::Program::Perl->new();
		return $c;
	}
	sub _code_open
	{
		my $self = shift;
		my $c = shift || ETL::Pequel3::CodeStyler::Program::Perl->new();
		return $c;
	}
	sub _code_close
	{
		my $self = shift;
		my $c = shift || ETL::Pequel3::CodeStyler::Program::Perl->new();
		return $c;
	}
	sub _code_prepare
	{
		my $self = shift;
		my $c = shift || ETL::Pequel3::CodeStyler::Program::Perl->new();
		return $c;
	}
	sub _code_lock
	{
		my $self = shift;
		my $c = shift || ETL::Pequel3::CodeStyler::Program::Perl->new();
		$c->code("use Fcntl ':flock';");
		$c->code("flock(@{[ $self->fdname() ]}, LOCK_EX);");
		return $c;
	}
	sub _code_unlock
	{
		my $self = shift;
		my $c = shift || ETL::Pequel3::CodeStyler::Program::Perl->new();
		$c->code("flock(@{[ $self->fdname() ]}, LOCK_UN);");
		return $c;
	}
}
# ----------------------------------------------------------------------------------------------------
{
	package ETL::Pequel3::Type::DataSource::Input::Abstract; 
	use base qw(ETL::Pequel3::Type::DataSource::Abstract);
	use Class::STL::ClassMembers
		Class::STL::ClassMembers::DataMember->new(name => 'io_name', default => 'input');
	use Class::STL::ClassMembers::Constructor;
	sub _code_prepare
	{
		my $self = shift;
		my $c = shift || ETL::Pequel3::CodeStyler::Program::Perl->new();
		defined($self->record_delimiter()) && $self->record_delimiter() ne ''
			? $c->code("local(\$/)=\"@{[ $self->record_delimiter() ]}\";")
			: $c->code("local(\$/)=undef;");
		return $c;
	}
	sub _map_output
	{
		my $self = shift;
		my %p = @_;
#<		$p{dataset}->input_fields()->map_fields($p{field_map}, $p{dataset}) #TODO: key_field ???
		$self->dataset()->input_fields()->map_fields($p{field_map}, $self->dataset()) #TODO: key_field ???
			if (exists($p{field_map}) && defined($p{field_map}));
	}
}
# ----------------------------------------------------------------------------------------------------
{
	package ETL::Pequel3::Type::DataSource::Output::Abstract;
	use base qw(ETL::Pequel3::Type::DataSource::Abstract);
	use Class::STL::ClassMembers
		Class::STL::ClassMembers::DataMember->new(name => 'io_name', default => 'output');
	use Class::STL::ClassMembers::Constructor;
	sub _code_prepare
	{
		my $self = shift;
		my $c = shift || ETL::Pequel3::CodeStyler::Program::Perl->new();
		defined($self->record_delimiter()) && $self->record_delimiter() ne ''
			? $c->code("local(\$\\)=\"@{[ $self->record_delimiter() ]}\";")
			: $c->code("local(\$\\)=undef;");
		return $c;
	}
}
# ----------------------------------------------------------------------------------------------------
{
	package ETL::Pequel3::Type::DataSource::Input::Pipe;
	use base qw(ETL::Pequel3::Type::DataSource::Input::Abstract);
	use Class::STL::ClassMembers
		Class::STL::ClassMembers::DataMember->new(name => 'description', default => 'The input data stream for this datasource type is I<piped> via F<STDIN>.'),
		Class::STL::ClassMembers::DataMember->new(name => 'datasource_name', default => 'stdin'),
		ETL::Pequel3::Type::DataMember::User->new(name => 'filter_cmd'),
		ETL::Pequel3::Type::DataMember::User->new(name => 'target_name', default => 'STDIN'),
		ETL::Pequel3::Type::DataMember::User->new(name => 'fdname', default => 'STDIN');
	use Class::STL::ClassMembers::Constructor;
	sub _code_open
	{
		my $self = shift;
		my $c = shift || ETL::Pequel3::CodeStyler::Program::Perl->new();
		$c->newline_off();
		$c->code("open(@{[ $self->fdname() ]}, ");
		$c->code("'-|', ") if (defined($self->filter_cmd()) || defined($self->sorter()));
		$c->code('"');
		if (defined($self->filter_cmd())) 
		{
			$c->code("@{[ $self->filter_cmd() ]} 2>/dev/null"); # gzcat, etc
			$c->code(" | @{[ $self->sorter()->code_open()->raw() ]} 2>/dev/null")
				if (defined($self->sorter()));
		}
		elsif (defined($self->sorter()))
		{
			$c->code("@{[ $self->sorter()->code_open()->raw() ]} 2>/dev/null")
		}
		else
		{
			$c->code("-");
		}
		$c->newline_on();
		$c->code('")');
		$c->code(" || die(\"Unable to open '@{[ $self->fdname() ]}'\");");
		return $c;
	}
	sub _code_close
	{
		my $self = shift;
		my $c = shift || ETL::Pequel3::CodeStyler::Program::Perl->new();
		$c->code("close(@{[ $self->fdname() ]});");
		return $c;
	}
}
# ----------------------------------------------------------------------------------------------------
{
	package ETL::Pequel3::Type::DataSource::Output::Pipe;
	use base qw(ETL::Pequel3::Type::DataSource::Output::Abstract);
	use Class::STL::ClassMembers
		Class::STL::ClassMembers::DataMember->new(name => 'description', default => 'The output data stream for this datasource type is I<piped> via F<STDOUT>.'),
		Class::STL::ClassMembers::DataMember->new(name => 'datasource_name', default => 'stdout'),
		ETL::Pequel3::Type::DataMember::User->new(name => 'filter_cmd'),
		ETL::Pequel3::Type::DataMember::User->new(name => 'target_name', default => 'STDOUT'),
		ETL::Pequel3::Type::DataMember::User->new(name => 'fdname', default => 'STDOUT');
	use Class::STL::ClassMembers::Constructor;
	sub _code_open
	{
		my $self = shift;
		my $c = shift || ETL::Pequel3::CodeStyler::Program::Perl->new();
		return $c unless (defined($self->sorter()) || defined($self->filter_cmd()));
		$c->newline_off();
		$c->code("open(@{[ $self->fdname() ]}, ");
		$c->code("'|-', ");
		$c->code('"');
		$c->code("@{[ $self->sorter()->code_open()->raw() ]} 2>/dev/null") if (defined($self->sorter()));
		$c->code(" | ") if (defined($self->sorter()) && defined($self->filter_cmd()));
		$c->code("@{[ $self->filter_cmd() ]}") if (defined($self->filter_cmd())); # gzcat, etc
		$c->newline_on();
		$c->code('")');
		$c->over();
		$c->code(" || die(\"Unable to open '@{[ $self->fdname() ]}'\");");
		$c->back();
		return $c;
	}
	sub _code_close
	{
		my $self = shift;
		my $c = shift || ETL::Pequel3::CodeStyler::Program::Perl->new();
		$c->code("close(@{[ $self->fdname() ]});");
		return $c;
	}
}
# ----------------------------------------------------------------------------------------------------
{
	package ETL::Pequel3::Type::DataSource::Input::File;
	use base qw(ETL::Pequel3::Type::DataSource::Input::Abstract);
	use Class::STL::ClassMembers qw( path filepath ),
		Class::STL::ClassMembers::DataMember->new(name => 'description', default => 'Specify an external disk file for the datasource.'),
		ETL::Pequel3::Type::DataMember::User->new(name => 'filter_cmd'),
#?		ETL::Pequel3::Type::DataMember::User->new(name => 'target_name'),
		Class::STL::ClassMembers::DataMember->new(name => 'datasource_name', default => 'file');
	use Class::STL::ClassMembers::Constructor;
	sub new_extra
	{
		my $self = shift;
		my $path_delim = '/';
#>		if (index($self->target_name(), $self->os_util()->path_delim()) == -1) {
		if (defined($self->target_name())) {
		   if (index($self->target_name(), $path_delim) != -1) {
				$self->path(substr($self->target_name(), 0, rindex($self->target_name(), $path_delim)));
				$self->target_name(substr($self->target_name(), rindex($self->target_name(), $path_delim)+1));
			}
			$self->filepath((defined($self->path()) ? $self->path() . '/' : '') . $self->target_name());
		}
#<		$self->ds_vname("_DS@{[ $self->fdname() ]}") if (defined($self->fdname()));
		return $self;
	}
	sub _code_init 
	{
		my $self = shift;
		my $c = shift || ETL::Pequel3::CodeStyler::Program::Perl->new();

		$self->filepath((defined($self->path()) ? $self->path() . '/' : '') . $self->target_name()); # path may have changed;
		$c->newline_off();
		$c->code("my \$@{[ $self->ds_vname() ]} = ");
		$c->newline_on();
		($self->filepath() =~ /^\.?\//)
			? $c->code("'@{[ $self->filepath() ]}';")
			: $c->code("\${\$g_USER_OPTION\{prefix}}\[0] . '/@{[ $self->filepath() ]}';");
#<		($self->filepath() =~ /^\.?\//)
#<			? $c->code("'@{[ $self->filepath() ]}';")
#<			: $self->properties()->prefix() ne ''
#<				? $c->code("'@{[ $self->properties()->prefix() ]}/@{[ $self->filepath() ]}';")
#<				: $c->code("(\"\@{[ defined(\$ENV{PEQUEL_PREFIX}) ? \$ENV{PEQUEL_PREFIX} . '/' : '' ]}\") . '@{[ $self->filepath() ]}';");
		return $c;
	}
	sub _code_open
	{
		my $self = shift;
		my $c = shift || ETL::Pequel3::CodeStyler::Program::Perl->new();
		$c->code("die(\"Data source file '\$@{[ $self->ds_vname() ]}'\"");
		$c->over();
			$c->code(". \" for '@{[ $self->fdname() ]}' does not exist or is unreadable!\")");
			$c->code("unless (-e \$@{[ $self->ds_vname() ]} && -r \$@{[ $self->ds_vname() ]});");
		$c->back();
		$c->newline_off();
		$c->code("open(@{[ $self->fdname() ]}, ");
		$c->code(defined($self->sorter()) || defined($self->filter_cmd()) ? "'-|', " : "'<', ");
		$c->code('"');
		if (defined($self->filter_cmd())) 
		{
			$c->code("@{[ $self->filter_cmd() ]} \$@{[ $self->ds_vname() ]} 2>/dev/null");
			$c->code(" | @{[ $self->sorter()->code_open()->raw() ]} 2>/dev/null")
				if (defined($self->sorter()));
		}
		elsif (defined($self->sorter()))
		{
			$c->code("@{[ $self->sorter()->code_open()->raw() ]} \$@{[ $self->ds_vname() ]} 2>/dev/null")
		}
		else
		{
			$c->code("\$@{[ $self->ds_vname() ]}");
		}
		$c->newline_on();
		$c->code('")');
		$c->over();
		$c->code(" || die(\"Unable to open data source file '\$@{[ $self->ds_vname() ]}'\"");
		$c->over();
		$c->code(". \" for '@{[ $self->fdname() ]}'\");");
		$c->back();
		$c->back();
		return $c;
	}
	sub _code_close
	{
		my $self = shift;
		my $c = shift || ETL::Pequel3::CodeStyler::Program::Perl->new();
		$c->code("close(@{[ $self->fdname() ]});");
		return $c;
	}
}
# ----------------------------------------------------------------------------------------------------
{
	package ETL::Pequel3::Type::DataSource::Output::File;
	use base qw(ETL::Pequel3::Type::DataSource::Input::File);
	use Class::STL::ClassMembers
		Class::STL::ClassMembers::DataMember->new(name => 'description', default => 'Specify an external disk file for the datasource.'),
		Class::STL::ClassMembers::DataMember->new(name => 'datasource_name', default => 'file'),
		ETL::Pequel3::Type::DataMember::User->new(name => 'append', default => 0), #TODO: test
#?		ETL::Pequel3::Type::DataMember::User->new(name => 'target_name'),
		Class::STL::ClassMembers::DataMember->new(name => 'io_name', default => 'output');
	use Class::STL::ClassMembers::Constructor;
	sub _code_open
	{
		my $self = shift;
		my $c = shift || ETL::Pequel3::CodeStyler::Program::Perl->new();
		$c->newline_off();
		$c->code("open(@{[ $self->fdname() ]}, ");
		$c->code(defined($self->sorter()) || defined($self->filter_cmd()) 
			? "'|-', " 
			: "'>@{[ $self->append() ? q/>/ : q// ]}', ");
		$c->code('"');
		if (defined($self->sorter()))
		{
			$c->code("@{[ $self->sorter()->code_open()->raw() ]} 2>/dev/null");
		   	$c->code(defined($self->filter_cmd()) ? " | @{[ $self->filter_cmd() ]} 2>/dev/null >@{[ $self->append() ? q/>/ : q// ]} " : ' -o ');
		}
		elsif (defined($self->filter_cmd()))
		{
			$c->code("@{[ $self->filter_cmd() ]} 2>/dev/null >@{[ $self->append() ? q/>/ : q// ]} ")
		}
		$c->code("\$@{[ $self->ds_vname() ]}");
		$c->newline_on();
		$c->code('")');
		$c->over();
		$c->code(" || die(\"Unable to open data source file '\$@{[ $self->ds_vname() ]}'\"");
		$c->over();
		$c->code(". \" for '@{[ $self->fdname() ]}'\");");
		$c->back();
		$c->back();
		return $c;
	}
	sub _code_prepare
	{
		my $self = shift;