Amazon-MWS
view release on metacpan or search on metacpan
lib/Amazon/MWS/Uploader.pm view on Meta::CPAN
package Amazon::MWS::Uploader;
use utf8;
use strict;
use warnings;
use DBI;
use Amazon::MWS::XML::Feed;
use Amazon::MWS::XML::Order;
use Amazon::MWS::Client;
use Amazon::MWS::XML::Response::FeedSubmissionResult;
use Amazon::MWS::XML::Response::OrderReport;
use Data::Dumper;
use File::Spec;
use DateTime;
use SQL::Abstract;
use Try::Tiny;
use Path::Tiny;
use Scalar::Util qw/blessed/;
use XML::Compile::Schema;
use Moo;
use MooX::Types::MooseLike::Base qw(:all);
use namespace::clean;
our $VERSION = '0.18';
use constant {
AMW_ORDER_WILDCARD_ERROR => 999999,
DEBUG => $ENV{AMZ_UPLOADER_DEBUG},
};
=head1 NAME
Amazon::MWS::Uploader -- high level agent to upload products to AMWS
=head1 DESCRIPTION
This module provide an high level interface to the upload process. It
has to keep track of the state to resume the uploading, which could
get stuck on the Amazon's side processing, so database credentials
have to be provided (or the database handle itself).
The table structure needed is defined and commented in sql/amazon.sql
=head1 SYNOPSIS
my $agent = Amazon::MWS::Uploader->new(
db_dsn => 'DBI:mysql:database=XXX',
db_username => 'xxx',
db_password => 'xxx',
db_options => \%options
# or dbh => $dbh,
schema_dir => '/path/to/xml_schema',
feed_dir => '/path/to/directory/for/xml',
merchant_id => 'xxx',
access_key_id => 'xxx',
secret_key => 'xxx',
marketplace_id => 'xxx',
endpoint => 'xxx',
products => \@products,
);
# say once a day, retrieve the full batch and send it up
$agent->upload;
# every 10 minutes or so, continue the work started with ->upload, if any
$agent->resume;
=head1 UPGRADE NOTES
When migrating from 0.05 to 0.06 please execute this SQL statement
ALTER TABLE amazon_mws_products ADD COLUMN listed BOOLEAN;
UPDATE amazon_mws_products SET listed = 1 WHERE status = 'ok';
When upgrading to 0.16, please execute this SQL statement:
ALTER TABLE amazon_mws_products ADD COLUMN warnings TEXT;
=head1 ACCESSORS
The following keys must be passed at the constructor and can be
accessed read-only:
=over 4
=item dbh
The DBI handle. If not provided will be built using the following
self-describing accessor:
=item db_dsn
=item db_username
=item db_password
=item db_options
E.g.
{
mysql_enable_utf8 => 1,
}
AutoCommit and RaiseError are set by us.
=cut
has db_dsn => (is => 'ro');
has db_password => (is => 'ro');
has db_username => (is => 'ro');
has db_options => (is => 'ro',
isa => AnyOf[Undef,HashRef],
);
has dbh => (is => 'lazy');
=item skus_warnings_modes
Determines how to treat warnings. This is a hash reference with the
code of the warning as key and one of the following modes as value:
=over 4
=item warn
lib/Amazon/MWS/Uploader.pm view on Meta::CPAN
You can pass an arbitrary identifier to the constructor which will be
used to keep the database records separated if you have multiple
amazon accounts. If not provided, the merchant id will be used, which
will work, but it's harder (for the humans) to spot and debug.
=cut
has shop_id => (is => 'ro');
has _unique_shop_id => (is => 'lazy');
sub _build__unique_shop_id {
my $self = shift;
if (my $id = $self->shop_id) {
return $id;
}
else {
return $self->merchant_id;
}
}
=item debug
Print out additional information.
=item logfile
Passed to L<Amazon::MWS::Client> constructor.
=cut
has debug => (is => 'ro');
has logfile => (is => 'ro');
=item quiet
Boolean. Do not warn on timeouts and aborts (just print) if set to
true.
=cut
has quiet => (is => 'ro');
sub _build_dbh {
my $self = shift;
my $dsn = $self->db_dsn;
die "Missing dns" unless $dsn;
my $options = $self->db_options || {};
# forse raise error and auto-commit
$options->{RaiseError} = 1;
$options->{AutoCommit} = 1;
my $dbh = DBI->connect($dsn, $self->db_username, $self->db_password,
$options) or die "Couldn't connect to $dsn!";
return $dbh;
}
=item purge_missing_products
If true, the first time C<products_to_upload> is called, products not
passed to the C<products> constructor will be purged from the
C<amazon_mws_products> table. Default to false.
This setting is DEPRECATED because can have some unwanted
side-effects. You are recommended to delete the obsoleted products
yourself.
=cut
has purge_missing_products => (is => 'rw');
=item reset_all_errors
If set to a true value, don't skip previously failed items and
effectively reset all of them.
Also, when the accessor is set for send_shipping_confirmation, try to
upload again previously failed orders.
=cut
has reset_all_errors => (is => 'ro');
=item reset_errors
A string containing a comma separated list of error codes, optionally
prefixed with a "!" (to reverse its meaning).
Example:
"!6024,6023"
Meaning: reupload all the products whose error code is B<not> 6024 or
6023.
"6024,6023"
Meaning: reupload the products whose error code was 6024 or 6023
=cut
has reset_errors => (is => 'ro',
isa => sub {
my $string = $_[0];
# undef/0/'' is fine
if ($string) {
die "reset_errors must be a comma separated list of error code, optionally prefixed by a '!' to negate its meaning"
if $string !~ m/^\s*!?\s*(([0-9]+)(\s*,\s*)?)+/;
}
});
has _reset_error_structure => (is => 'lazy');
sub _build__reset_error_structure {
my $self = shift;
my $reset_string = $self->reset_errors || '';
$reset_string =~ s/^\s*//;
$reset_string =~ s/\s*$//;
return unless $reset_string;
my $negate = 0;
if ($reset_string =~ m/^\s*!\s*(.+)/) {
$reset_string = $1;
$negate = 1;
}
my %codes = map { $_ => 1 } grep { $_ } split(/\s*,\s*/, $reset_string);
return unless %codes;
return {
negate => $negate,
codes => \%codes,
};
}
=item force
Same as above, but only for the selected items. An arrayref is
expected here with the B<skus>.
=cut
has force => (is => 'ro',
isa => ArrayRef,
);
has _force_hashref => (is => 'lazy');
sub _build__force_hashref {
my $self = shift;
my %forced;
if (my $arrayref = $self->force) {
%forced = map { $_ => 1 } @$arrayref;
}
return \%forced;
}
=item limit_inventory
If set to an integer, limit the inventory to this value. Setting this
to 0 will disable it.
=item job_hours_timeout
If set to an integer, abort the job after X hours are elapsed since
the job was started. Default to 3 hours. Set to 0 to disable (not
recommended).
This doesn't affect jobs for order acknowledgements (C<order_ack>), see below.
=item order_ack_days_timeout
Order acknowlegments times out at different rate, because it's somehow
sensitive.
=cut
has job_hours_timeout => (is => 'ro',
isa => Int,
default => sub { 3 });
has order_ack_days_timeout => (is => 'ro',
isa => Int,
default => sub { 30 });
has limit_inventory => (is => 'ro',
isa => Int);
=item schema_dir
The directory where the xsd files for the feed building can be found.
=item feeder
A L<Amazon::MWS::XML::Feed> object. Lazy attribute, you shouldn't pass
this to the constructor, it is lazily built using C<products>,
C<merchant_id> and C<schema_dir>.
=item feed_dir
A working directory where to stash the uploaded feeds for inspection
if problems are detected.
=item schema
The L<XML::Compile::Schema> object, built lazily from C<feed_dir>
=item xml_writer
The xml writer, built lazily.
=item xml_reader
The xml reader, built lazily.
=cut
has schema_dir => (is => 'ro',
required => 1,
isa => sub {
die "$_[0] is not a directory" unless -d $_[0];
});
has feed_dir => (is => 'ro',
required => 1,
isa => sub {
die "$_[0] is not a directory" unless -d $_[0];
});
has schema => (is => 'lazy');
sub _build_schema {
my $self = shift;
my $files = File::Spec->catfile($self->schema_dir, '*.xsd');
my $schema = XML::Compile::Schema->new([glob $files]);
return $schema;
}
has xml_writer => (is => 'lazy');
sub _build_xml_writer {
my $self = shift;
return $self->schema->compile(WRITER => 'AmazonEnvelope');
}
has xml_reader => (is => 'lazy');
sub _build_xml_reader {
my $self = shift;
return $self->schema->compile(READER => 'AmazonEnvelope');
}
=item generic_feeder
Return a L<Amazon::MWS::XML::GenericFeed> object to build a feed using
the XML writer.
=cut
sub generic_feeder {
my $self = shift;
return Amazon::MWS::XML::GenericFeed->new(
xml_writer => $self->xml_writer,
merchant_id => $self->merchant_id,
);
}
=item merchant_id
The merchant ID provided by Amazon.
=item access_key_id
Provided by Amazon.
=item secret_key
Provided by Amazon.
=item marketplace_id
L<http://docs.developer.amazonservices.com/en_US/dev_guide/DG_Endpoints.html>
=item endpoint
Ditto.
=cut
has merchant_id => (is => 'ro', required => 1);
has access_key_id => (is => 'ro', required => 1);
has secret_key => (is => 'ro', required => 1);
has marketplace_id => (is => 'ro', required => 1);
has endpoint => (is => 'ro', required => 1);
=item products
An arrayref of L<Amazon::MWS::XML::Product> objects, or anything that
(properly) responds to C<as_product_hash>, C<as_inventory_hash>,
C<as_price_hash>. See L<Amazon::MWS::XML::Product> for details.
B<This is set as read-write, so you can set the product after the
object construction, but if you change it afterward, you will get
unexpected results>.
This routine also check if the product needs upload and delete
disappeared products. If you are doing the check yourself, use
C<checked_products>.
=item checked_products
As C<products>, but no check is performed. This takes precedence.
=item sqla
Lazy attribute to hold the C<SQL::Abstract> object.
=cut
has products => (is => 'rw',
isa => ArrayRef);
has sqla => (
is => 'ro',
default => sub {
return SQL::Abstract->new;
}
);
has existing_products => (is => 'lazy');
sub _build_existing_products {
my $self = shift;
my $sth = $self->_exe_query($self->sqla->select(amazon_mws_products => [qw/sku
timestamp_string
status
listed
error_code
/],
{
status => { -not_in => [qw/deleted/] },
shop_id => $self->_unique_shop_id,
}));
my %uploaded;
while (my $row = $sth->fetchrow_hashref) {
$row->{timestamp_string} ||= 0;
$uploaded{$row->{sku}} = $row;
}
return \%uploaded;
}
has products_to_upload => (is => 'lazy');
has checked_products => (is => 'rw', isa => ArrayRef);
sub _build_products_to_upload {
my $self = shift;
if (my $checked = $self->checked_products) {
return $checked;
}
my $product_arrayref = $self->products;
return [] unless $product_arrayref && @$product_arrayref;
my @products = @$product_arrayref;
my $existing = $self->existing_products;
my @todo;
foreach my $product (@products) {
my $sku = $product->sku;
if (my $exists = $existing->{$sku}) {
# mark the item as visited
$exists->{_examined} = 1;
}
print "Checking $sku\n" if $self->debug;
next unless $self->product_needs_upload($product->sku, $product->timestamp_string);
print "Scheduling product " . $product->sku . " for upload\n";
if (my $limit = $self->limit_inventory) {
my $real = $product->inventory;
if ($real > $limit) {
print "Limiting the $sku inventory from $real to $limit\n" if $self->debug;
$product->inventory($limit);
}
}
if (my $children = $product->children) {
my @good_children;
foreach my $child (@$children) {
# skip failed children, but if the current status of
# parent is failed, and we reached this point, retry.
if (! exists $self->_force_hashref->{$child} and
$existing->{$child} and
$existing->{$sku} and
$existing->{$sku}->{status} ne 'failed' and
$existing->{$child}->{status} eq 'failed') {
print "Ignoring failed variant $child\n";
}
else {
push @good_children, $child;
}
}
$product->children(\@good_children);
}
push @todo, $product;
}
if ($self->purge_missing_products) {
# nuke the products not passed
# print Dumper($existing);
my @deletions = map { $_->{sku} }
grep { !$_->{_examined} }
values %$existing;
if (@deletions) {
$self->delete_skus(@deletions);
}
}
return \@todo;
}
=item client
An L<Amazon::MWS::Client> object, built lazily, so you don't have to
pass it.
=back
=cut
has client => (is => 'lazy');
sub _build_client {
my $self = shift;
my %mws_args = map { $_ => $self->$_ } (qw/merchant_id
marketplace_id
access_key_id
secret_key
debug
logfile
endpoint/);
return Amazon::MWS::Client->new(%mws_args);
}
has _mismatch_patterns => (is => 'lazy', isa => HashRef);
sub _build__mismatch_patterns {
my $self = shift;
my $merchant_re = qr{\s+\((?:Merchant|Verkäufer):\s+'(.*?)'\s+/};
my $amazon_re = qr{\s+.*?/\s*Amazon:\s+'(.*?)'\)};
my %patterns = (
# informative only
asin => qr{ASIN(?:\s+überein)?\s+([0-9A-Za-z]+)},
shop_part_number => qr{part_number$merchant_re},
amazon_part_number => qr{part_number$amazon_re},
shop_title => qr{item_name$merchant_re},
amazon_title => qr{item_name$amazon_re},
shop_manufacturer => qr{manufacturer$merchant_re},
amazon_manufacturer => qr{manufacturer$amazon_re},
shop_brand => qr{brand$merchant_re},
amazon_brand => qr{brand$amazon_re},
shop_color => qr{color$merchant_re},
amazon_color => qr{color$amazon_re},
shop_size => qr{size$merchant_re},
amazon_size => qr{size$amazon_re},
);
return \%patterns;
}
=head1 MAIN METHODS
=head2 upload
If the products is set, begin the routine to upload them. Because of
the asynchronous way AMWS works, at some point it will bail out,
saving the state in the database. You should reinstantiate the object
and call C<resume> on it every 10 minutes or so.
The workflow is described here:
L<http://docs.developer.amazonservices.com/en_US/feeds/Feeds_Overview.html>
This has to be done for each feed: Product, Inventory, Price, Image,
Relationship (for variants).
This method first generate the feeds in the feed directory, and then
calls C<resume>, which is in charge for the actual uploading.
=head2 resume
Restore the state and resume where it was left.
This method accepts an optional list of parameters. Each parameter may be:
=over 4
=item a scalar
This is considered a job id.
=item a hashref
This will be merged in the query to retrieve the pending jobs. A
sample usage could be:
$upload->resume({ task => [qw/upload product_deletion/] });
to resume only those specific tasks.
=back
=head2 get_pending_jobs
Return the list of hashref with the pending jobs out of the database.
Accepts the same parameters as C<resume> (which actually calls this
method).
=cut
=head1 INTERNAL METHODS
=head2 prepare_feeds($type, { name => $feed_name, content => "<xml>..."}, { name => $feed_name2, content => "<xml>..."}, ....)
Prepare the feed of type $type with the feeds provided as additional
arguments.
Return the job id
=cut
sub _feed_job_dir {
my ($self, $job_id, $create) = @_;
die unless $job_id;
my $shop_id = $self->_unique_shop_id;
$shop_id =~ s/[^0-9A-Za-z_-]//g;
die "The shop id without word characters results in an empty string"
unless $shop_id;
my $feed_root = File::Spec->catdir($self->feed_dir,
$shop_id);
if ($create) {
mkdir $feed_root unless -d $feed_root;
}
my $feed_subdir = File::Spec->catdir($feed_root,
$job_id);
if ($create) {
mkdir $feed_subdir unless -d $feed_subdir;
}
return $feed_subdir;
}
sub _feed_file_for_method {
my ($self, $job_id, $feed_type) = @_;
die unless $job_id && $feed_type;
my $feed_subdir = $self->_feed_job_dir($job_id, "create");
my $file = File::Spec->catfile($feed_subdir, $feed_type . '.xml');
return File::Spec->rel2abs($file);
}
sub _slurp_file {
my ($self, $file) = @_;
open (my $fh, '<', $file) or die "Couldn't open $file $!";
local $/ = undef;
my $content = <$fh>;
close $fh;
return $content;
}
sub upload {
my $self = shift;
# create the feeds to be uploaded using the products
my @products = @{ $self->products_to_upload };
unless (@products) {
print "No products, can't upload anything\n";
return;
}
my $feeder = Amazon::MWS::XML::Feed->new(
products => \@products,
xml_writer => $self->xml_writer,
merchant_id => $self->merchant_id,
);
my @feeds;
foreach my $feed_name (qw/product
inventory
price
image
variants
/) {
my $method = $feed_name . "_feed";
if (my $content = $feeder->$method) {
push @feeds, {
name => $feed_name,
content => $content,
};
}
}
if (my $job_id = $self->prepare_feeds(upload => \@feeds)) {
$self->_mark_products_as_pending($job_id, @products);
return $job_id;
}
return;
}
sub _mark_products_as_pending {
my ($self, $job_id, @products) = @_;
die "Bad usage" unless $job_id;
# these skus were cleared up when asking for the products to upload
foreach my $p (@products) {
my %identifier = (
sku => $p->sku,
shop_id => $self->_unique_shop_id,
);
my %data = (
amws_job_id => $job_id,
status => 'pending',
warnings => '', # clear out
timestamp_string => $p->timestamp_string,
);
my $check = $self
->_exe_query($self->sqla->select(amazon_mws_products => [qw/sku/], { %identifier }));
my $existing = $check->fetchrow_hashref;
$check->finish;
if ($existing) {
$self->_exe_query($self->sqla->update(amazon_mws_products => \%data, \%identifier));
}
else {
$self->_exe_query($self->sqla->insert(amazon_mws_products => { %identifier, %data }));
}
}
}
sub prepare_feeds {
my ($self, $task, $feeds) = @_;
die "Missing task ($task) and feeds ($feeds)" unless $task && $feeds;
return unless @$feeds; # nothing to do
my $job_id = $task . "-" . DateTime->now->strftime('%F-%H-%M-%S');
my $job_started_epoch = time();
$self->_exe_query($self->sqla
->insert(amazon_mws_jobs => {
amws_job_id => $job_id,
shop_id => $self->_unique_shop_id,
task => $task,
job_started_epoch => $job_started_epoch,
}));
# to complete the process, we need to fill out these five
# feeds. every feed has the same procedure, as per
# http://docs.developer.amazonservices.com/en_US/feeds/Feeds_Overview.html
# so we put a flag on the feed when it is done. The processing
# of the feed itself is tracked in the amazon_mws_feeds
# TODO: we could pass to the object some flags to filter out results.
foreach my $feed (@$feeds) {
# write out the feed if we got something to do, and add a row
# to the feeds.
# when there is no content, no need to create a job for it.
if (my $content = $feed->{content}) {
my $name = $feed->{name} or die "Missing feed_name";
my $file = $self->_feed_file_for_method($job_id, $name);
open (my $fh, '>', $file) or die "Couldn't open $file $!";
print $fh $content;
close $fh;
# and prepare a row for it
my $insertion = {
feed_name => $name,
feed_file => $file,
amws_job_id => $job_id,
shop_id => $self->_unique_shop_id,
};
$self->_exe_query($self->sqla
->insert(amazon_mws_feeds => $insertion));
}
}
return $job_id;
}
sub get_pending_jobs {
my ($self, @args) = @_;
my %additional;
my @named_jobs;
foreach my $arg (@args) {
if (!ref($arg)) {
push @named_jobs, $arg;
}
elsif (ref($arg) eq 'HASH') {
# add the filters
foreach my $key (keys %$arg) {
if ($additional{$key}) {
die "Attempt to overwrite $key in the additional parameters!\n";
}
else {
$additional{$key} = $arg->{$key};
}
}
}
else {
die "Argument must be either a scalar with a job name and/or "
. "an hashref with additional filters!";
}
}
if (@named_jobs) {
$additional{amws_job_id} = { -in => \@named_jobs };
}
my ($stmt, @bind) = $self->sqla->select(amazon_mws_jobs => '*',
{
%additional,
aborted => 0,
success => 0,
shop_id => $self->_unique_shop_id,
},
{ -asc => 'job_started_epoch'});
my $pending = $self->_exe_query($stmt, @bind);
my %jobs;
while (my $row = $pending->fetchrow_hashref) {
$jobs{$row->{task}} ||= [];
push @{$jobs{$row->{task}}}, $row;
}
my @out;
foreach my $task (qw/product_deletion upload shipping_confirmation order_ack/) {
if (my $list = delete $jobs{$task}) {
if ($task eq 'order_ack') {
for (1..2) {
push @out, pop @$list if @$list;
}
}
elsif ($task eq 'shipping_confirmation') {
while (@$list) {
push @out, pop @$list;
}
}
else {
push @out, @$list if @$list;
}
}
}
return @out;
}
sub resume {
my ($self, @args) = @_;
foreach my $row ($self->get_pending_jobs(@args)) {
print "Working on $row->{amws_job_id}\n";
# check if the job dir exists
if (-d $self->_feed_job_dir($row->{amws_job_id})) {
if (my $seconds_elapsed = $self->job_timed_out($row)) {
$self->_print_or_warn_error("Timeout reached for $row->{amws_job_id}, aborting: "
. Dumper($row));
$self->cancel_job($row->{task}, $row->{amws_job_id},
"Job timed out after $seconds_elapsed seconds");
next;
}
$self->process_feeds($row);
}
else {
warn "No directory " . $self->_feed_job_dir($row->{amws_job_id}) .
" found, removing job id $row->{amws_job_id}\n";
$self->cancel_job($row->{task}, $row->{amws_job_id},
"Job canceled due to missing feed directory");
}
}
}
=head2 cancel_job($task, $job_id, $reason)
Abort the job setting the aborted flag in C<amazon_mws_jobs> table.
=cut
sub cancel_job {
my ($self, $task, $job_id, $reason) = @_;
$self->_exe_query($self->sqla->update('amazon_mws_jobs',
{
aborted => 1,
status => $reason,
},
{
amws_job_id => $job_id,
shop_id => $self->_unique_shop_id,
}));
# and revert the products' status
my $status;
if ($task eq 'product_deletion') {
# let's pretend we were deleting good products
$status = 'ok';
}
elsif ($task eq 'upload') {
$status = 'redo';
}
if ($status) {
print "Updating product to $status for products with job id $job_id\n";
$self->_exe_query($self->sqla->update('amazon_mws_products',
{ status => $status },
{
amws_job_id => $job_id,
shop_id => $self->_unique_shop_id,
}));
}
}
=head2 process_feeds(\%job_row)
Given the hashref with the db row of the job, check at which point it
is and resume.
=cut
sub process_feeds {
my ($self, $row) = @_;
# print Dumper($row);
# upload the feeds one by one and stop if something is blocking
my $job_id = $row->{amws_job_id};
print "Processing job $job_id\n";
# query the feeds table for this job
my ($stmt, @bind) = $self->sqla->select(amazon_mws_feeds => '*',
{
amws_job_id => $job_id,
aborted => 0,
success => 0,
shop_id => $self->_unique_shop_id,
},
['amws_feed_pk']);
my $sth = $self->_exe_query($stmt, @bind);
my $unfinished;
while (my $feed = $sth->fetchrow_hashref) {
last unless $self->upload_feed($feed);
}
$sth->finish;
($stmt, @bind) = $self->sqla->select(amazon_mws_feeds => '*',
{
shop_id => $self->_unique_shop_id,
amws_job_id => $job_id,
});
$sth = $self->_exe_query($stmt, @bind);
my ($total, $success, $aborted) = (0, 0, 0);
# query again and check if we have aborted jobs;
while (my $feed = $sth->fetchrow_hashref) {
$total++;
$success++ if $feed->{success};
$aborted++ if $feed->{aborted};
}
# a job was aborted
my $update;
if ($aborted) {
$update = {
aborted => 1,
status => 'Feed error',
};
$self->_print_or_warn_error("Job $job_id aborted!\n");
}
elsif ($success == $total) {
$update = { success => 1 };
print "Job successful!\n";
# if we're here, all the products are fine, so mark them as
# such if it's an upload job
if ($row->{task} eq 'upload') {
$self->_exe_query($self->sqla->update('amazon_mws_products',
{ status => 'ok',
listed_date => DateTime->now,
listed => 1,
},
{
amws_job_id => $job_id,
shop_id => $self->_unique_shop_id,
}));
}
}
else {
print "Job still to be processed\n";
}
if ($update) {
$self->_exe_query($self->sqla->update(amazon_mws_jobs => $update,
{
amws_job_id => $job_id,
shop_id => $self->_unique_shop_id,
}));
}
}
=head2 upload_feed($type, $feed_id);
Routine to upload the feed. Return true if it's complete, false
otherwise.
=cut
sub upload_feed {
my ($self, $record) = @_;
my $job_id = $record->{amws_job_id};
my $type = $record->{feed_name};
my $feed_id = $record->{feed_id};
print "Checking $type feed for $job_id\n";
# http://docs.developer.amazonservices.com/en_US/feeds/Feeds_FeedType.html
my %names = (
product => '_POST_PRODUCT_DATA_',
inventory => '_POST_INVENTORY_AVAILABILITY_DATA_',
price => '_POST_PRODUCT_PRICING_DATA_',
image => '_POST_PRODUCT_IMAGE_DATA_',
variants => '_POST_PRODUCT_RELATIONSHIP_DATA_',
order_ack => '_POST_ORDER_ACKNOWLEDGEMENT_DATA_',
shipping_confirmation => '_POST_ORDER_FULFILLMENT_DATA_',
);
die "Unrecognized type $type" unless $names{$type};
# no feed id, it's a new batch
if (!$feed_id) {
print "No feed id found, doing a request for $job_id $type\n";
my $feed_content = $self->_slurp_file($record->{feed_file});
my $res;
try {
$res = $self->client
->SubmitFeed(content_type => 'text/xml; charset=utf-8',
FeedType => $names{$type},
FeedContent => $feed_content,
MarketplaceIdList => [$self->marketplace_id],
);
}
catch {
warn "Failure to submit $type feed: \n";
if (ref($_)) {
if ($_->can('xml')) {
warn $_->xml;
}
else {
warn Dumper($_);
}
}
else {
warn $_;
}
};
# do not register the failure on die, because in this case (no
# response) there could be throttling, or network failure
die unless $res;
# update the feed_id row storing it and updating.
if ($feed_id = $record->{feed_id} = $res->{FeedSubmissionId}) {
$self->_exe_query($self->sqla
->update(amazon_mws_feeds => $record,
{
amws_feed_pk => $record->{amws_feed_pk},
shop_id => $self->_unique_shop_id,
}));
}
lib/Amazon/MWS/Uploader.pm view on Meta::CPAN
retrieve_orderline_sub => $get_orderline);
}
return @orders;
}
=head2 order_already_registered($order)
Check in the amazon_mws_orders table if we already registered this
order.
Return the row for this table (as an hashref) if present, nothing
underwise.
=cut
sub order_already_registered {
my ($self, $order) = @_;
die "Bad usage, missing order" unless $order;
my $sth = $self->_exe_query($self->sqla->select(amazon_mws_orders => '*',
{
amazon_order_id => $order->amazon_order_number,
shop_id => $self->_unique_shop_id,
}));
if (my $exists = $sth->fetchrow_hashref) {
$sth->finish;
return $exists;
}
else {
return;
}
}
=head2 acknowledge_successful_order(@orders)
Accept a list of L<Amazon::MWS::XML::Order> objects, prepare a
acknowledge feed with the C<Success> status, and insert the orders in
the database.
=cut
sub acknowledge_successful_order {
my ($self, @orders) = @_;
my @orders_to_register;
foreach my $ord (@orders) {
if (my $existing = $self->order_already_registered($ord)) {
if ($existing->{confirmed}) {
print "Skipping already confirmed order $existing->{amazon_order_id} => $existing->{shop_order_id}\n";
}
else {
# it's not complete, so print out diagnostics
warn "Order $existing->{amazon_order_id} uncompletely registered with id $existing->{shop_order_id}, please indagate why (skipping)\n" . Dumper($existing);
}
}
else {
push @orders_to_register, $ord;
}
}
return unless @orders_to_register;
my $feed_content = $self->acknowledge_feed(Success => @orders_to_register);
# here we have only one feed to upload and check
my $job_id = $self->prepare_feeds(order_ack => [{
name => 'order_ack',
content => $feed_content,
}]);
# store the pairing amazon order id / shop order id in our table
foreach my $order (@orders_to_register) {
my %order_pairs = (
shop_id => $self->_unique_shop_id,
amazon_order_id => $order->amazon_order_number,
# this will die if we try to insert an undef order_number
shop_order_id => $order->order_number,
amws_job_id => $job_id,
);
$self->_exe_query($self->sqla->insert(amazon_mws_orders => \%order_pairs));
}
}
=head2 acknowledge_feed($status, @orders)
The first argument is usually C<Success>. The other arguments is a
list of L<Amazon::MWS::XML::Order> objects.
=cut
sub acknowledge_feed {
my ($self, $status, @orders) = @_;
die "Missing status" unless $status;
die "Missing orders" unless @orders;
my $feeder = $self->generic_feeder;
my $counter = 1;
my @messages;
foreach my $order (@orders) {
my $data = $order->as_ack_order_hashref;
$data->{StatusCode} = $status;
push @messages, {
MessageID => $counter++,
OrderAcknowledgement => $data,
};
}
return $feeder->create_feed(OrderAcknowledgement => \@messages);
}
=head2 delete_skus(@skus)
Accept a list of skus. Prepare a C<product_deletion> feed and update
the database.
=cut
sub delete_skus {
my ($self, @skus) = @_;
return unless @skus;
print "Trying to purge missing items " . join(" ", @skus) . "\n";
# delete only products which are not in pending status
my $check = $self
lib/Amazon/MWS/Uploader.pm view on Meta::CPAN
=head2 register_errors($job_id, $result)
The first argument is the job ID. The second is a
L<Amazon::MWS::XML::Response::FeedSubmissionResult> object.
This method will update the status of the products (either C<failed>
or C<redo>) in C<amazon_mws_products>.
=head2 register_order_ack_errors($job_id, $result);
Same arguments as above, but for order acknowledgements.
=head2 register_ship_order_errors($job_id, $result);
Same arguments as above, but for shipping notifications.
=cut
sub register_errors {
my ($self, $job_id, $result) = @_;
# first, get the list of all the skus which were scheduled for this job
# we don't have a products hashref anymore.
# probably we could parse back the produced xml, but looks like an overkill.
# just mark them as redo and wait for the next cron call.
my @products = $self->skus_in_job($job_id);
my $errors = $result->skus_errors;
my @errors_with_sku = grep { $_->{sku} } @$errors;
# turn it into an hash
my %errs = map { $_->{sku} => {job_id => $job_id, code => $_->{code}, error => $_->{error}} } @errors_with_sku;
foreach my $sku (@products) {
if ($errs{$sku}) {
$self->_exe_query($self->sqla->update('amazon_mws_products',
{
status => 'failed',
error_code => $errs{$sku}->{code},
error_msg => "$errs{$sku}->{job_id} $errs{$sku}->{code} $errs{$sku}->{error}",
},
{
sku => $sku,
shop_id => $self->_unique_shop_id,
}));
}
else {
# this is good, mark it to be redone
$self->_exe_query($self->sqla->update('amazon_mws_products',
{
status => 'redo',
},
{
sku => $sku,
shop_id => $self->_unique_shop_id,
}));
print "Scheduling $sku for redoing\n";
}
}
}
=head2 skus_in_job($job_id)
Check the amazon_mws_product for the SKU which were uploaded by the
given job ID.
=cut
sub skus_in_job {
my ($self, $job_id) = @_;
my $sth = $self->_exe_query($self->sqla->select('amazon_mws_products',
[qw/sku/],
{
amws_job_id => $job_id,
shop_id => $self->_unique_shop_id,
}));
my @skus;
while (my $row = $sth->fetchrow_hashref) {
push @skus, $row->{sku};
}
return @skus;
}
=head2 get_asin_for_eans(@eans)
Accept a list of EANs and return an hashref where the keys are the
eans passed as arguments, and the values are the ASIN for the current
marketplace. Max EANs: 5.x
http://docs.developer.amazonservices.com/en_US/products/Products_GetMatchingProductForId.html
=head2 get_asin_for_skus(@skus)
Same as above (with the same limit of 5 items), but for SKUs.
=head2 get_asin_for_sku($sku)
Same as above, but for a single sku. Return the ASIN or undef if not
found.
=head2 get_asin_for_ean($ean)
Same as above, but for a single ean. Return the ASIN or undef if not
found.
=cut
sub get_asin_for_skus {
my ($self, @skus) = @_;
return $self->_get_asin_for_type(SellerSKU => @skus);
}
sub get_asin_for_eans {
my ($self, @eans) = @_;
return $self->_get_asin_for_type(EAN => @eans);
}
sub _get_asin_for_type {
my ($self, $type, @list) = @_;
die "Max 5 products to get the asin for $type!" if @list > 5;
my $client = $self->client;
my $res;
try {
$res = $client->GetMatchingProductForId(IdType => $type,
lib/Amazon/MWS/Uploader.pm view on Meta::CPAN
sub get_lowest_price_for_ean {
my ($self, $ean, $condition) = @_;
return unless $ean;
my $asin = $self->get_asin_for_ean($ean);
return unless $asin;
return $self->get_lowest_price_for_asin($asin, $condition);
}
sub get_lowest_price_for_asin {
my ($self, $asin, $condition) = @_;
die "Wrong usage, missing argument asin" unless $asin;
my $listing;
try { $listing = $self->client
->GetLowestOfferListingsForASIN(
ASINList => [ $asin ],
MarketplaceId => $self->marketplace_id,
ExcludeMe => 1,
ItemCondition => $condition || 'New',
);
}
catch { die Dumper($_) };
return unless $listing && @$listing;
my $lowest;
foreach my $item (@$listing) {
my $current = $item->{Price}->{LandedPrice}->{Amount};
$lowest ||= $current;
if ($current < $lowest) {
$lowest = $current;
}
}
return $lowest;
}
=head2 shipping_confirmation_feed(@shipped_orders)
Return a feed string with the shipping confirmation. A list of
L<Amazon::MWS::XML::ShippedOrder> object must be passed.
=cut
sub shipping_confirmation_feed {
my ($self, @shipped_orders) = @_;
die "Missing Amazon::MWS::XML::ShippedOrder argument" unless @shipped_orders;
my $feeder = $self->generic_feeder;
my $counter = 1;
my @messages;
foreach my $order (@shipped_orders) {
push @messages, {
MessageID => $counter++,
OrderFulfillment => $order->as_shipping_confirmation_hashref,
};
}
return $feeder->create_feed(OrderFulfillment => \@messages);
}
=head2 send_shipping_confirmation($shipped_orders)
Schedule the shipped orders (an L<Amazon::MWS::XML::ShippedOrder>
object) for the uploading.
=head2 order_already_shipped($shipped_order)
Check if the shipped orders (an L<Amazon::MWS::XML::ShippedOrder> was
already notified as shipped looking into our table, returning the row
with the order.
To see the status, check shipping_confirmation_ok (already done),
shipping_confirmation_error (faulty), shipping_confirmation_job_id (pending).
=cut
sub order_already_shipped {
my ($self, $order) = @_;
my $condition = $self->_condition_for_shipped_orders($order);
my $sth = $self->_exe_query($self->sqla->select(amazon_mws_orders => '*', $condition));
if (my $row = $sth->fetchrow_hashref) {
die "Multiple results found in amazon_mws_orders for " . Dumper($condition)
if $sth->fetchrow_hashref;
return $row;
}
else {
return;
}
}
sub send_shipping_confirmation {
my ($self, @orders) = @_;
my @orders_to_notify;
foreach my $ord (@orders) {
if (my $report = $self->order_already_shipped($ord)) {
if ($report->{shipping_confirmation_ok}) {
print "Skipping ship-confirm for order $report->{amazon_order_id} $report->{shop_order_id}: already notified\n";
}
elsif (my $error = $report->{shipping_confirmation_error}) {
if ($self->reset_all_errors) {
warn "Submitting again previously failed job $report->{amazon_order_id} $report->{shop_order_id}\n";
push @orders_to_notify, $ord;
}
else {
warn "Skipping ship-confirm for order $report->{amazon_order_id} $report->{shop_order_id} with error $error\n";
}
}
elsif ($report->{shipping_confirmation_job_id}) {
print "Skipping ship-confirm for order $report->{amazon_order_id} $report->{shop_order_id}: pending\n";
}
else {
push @orders_to_notify, $ord;
}
}
else {
die "It looks like you are trying to send a shipping confirmation "
. " without prior order acknowlegdement. "
. "At least in the amazon_mws_orders there is no trace of "
. "$report->{amazon_order_id} $report->{shop_order_id}";
}
}
return unless @orders_to_notify;
my $feed_content = $self->shipping_confirmation_feed(@orders_to_notify);
# here we have only one feed to upload and check
my $job_id = $self->prepare_feeds(shipping_confirmation => [{
name => 'shipping_confirmation',
content => $feed_content,
}]);
# and store the job id in the table
foreach my $ord (@orders_to_notify) {
$self->_exe_query($self->sqla->update(amazon_mws_orders => {
shipping_confirmation_job_id => $job_id,
shipping_confirmation_error => undef,
},
$self->_condition_for_shipped_orders($ord)));
}
}
sub _condition_for_shipped_orders {
my ($self, $order) = @_;
die "Missing order" unless $order;
my %condition = (shop_id => $self->_unique_shop_id);
if (my $amazon_order_id = $order->amazon_order_id) {
$condition{amazon_order_id} = $amazon_order_id;
}
elsif (my $order_id = $order->merchant_order_id) {
$condition{shop_order_id} = $order_id;
}
else {
die "Missing amazon_order_id or merchant_order_id";
}
return \%condition;
}
=head2 orders_waiting_for_shipping
Return a list of hashref with two keys, C<amazon_order_id> and
C<shop_order_id> for each order which is waiting confirmation.
This is implemented looking into amazon_mws_orders where there is no
shipping confirmation job id.
The confirmed flag (which means we acknowledged the order) is ignored
to avoid stuck order_ack jobs to prevent the shipping confirmation.
=cut
sub orders_waiting_for_shipping {
my $self = shift;
my $sth = $self->_exe_query($self->sqla->select('amazon_mws_orders',
[qw/amazon_order_id
shop_order_id/],
{
shop_id => $self->_unique_shop_id,
shipping_confirmation_job_id => undef,
# do not stop the unconfirmed to be considered
# confirmed => 1,
}));
my @out;
while (my $row = $sth->fetchrow_hashref) {
push @out, $row;
}
return @out;
}
=head2 product_needs_upload($sku, $timestamp)
Lookup the product $sku with timestamp $timestamp and return the sku
if the product needs to be uploaded or can be safely skipped. This
method is stateless and doesn't alter anything.
=cut
sub product_needs_upload {
my ($self, $sku, $timestamp) = @_;
my $debug = $self->debug;
return unless $sku;
my $forced = $self->_force_hashref;
# if it's forced, we have nothing to check, just pass it.
if ($forced->{$sku}) {
print "Forcing $sku as requested\n" if $debug;
return $sku;
}
$timestamp ||= 0;
my $existing = $self->existing_products;
if (exists $existing->{$sku}) {
if (my $exists = $existing->{$sku}) {
my $status = $exists->{status} || '';
if ($status eq 'ok') {
if ($exists->{timestamp_string} eq $timestamp) {
return;
}
else {
return $sku;
}
}
elsif ($status eq 'redo') {
return $sku;
}
elsif ($status eq 'failed') {
if ($self->reset_all_errors) {
return $sku;
}
elsif (my $reset = $self->_reset_error_structure) {
# option for this error was passed.
my $error = $exists->{error_code};
my $match = $reset->{codes}->{$error};
if (($match && $reset->{negate}) or
(!$match && !$reset->{negate})) {
# was passed !this error or !random , so do not reset
print "Skipping failed item $sku with error code $error\n" if $debug;
return;
}
else {
# otherwise reset
print "Resetting error for $sku with error code $error\n" if $debug;
return $sku;
}
}
else {
print "Skipping failed item $sku\n" if $debug;
return;
}
}
elsif ($status eq 'pending') {
print "Skipping pending item $sku\n" if $debug;
return;
}
die "I shouldn't have reached this point with status <$status>";
}
}
print "$sku wasn't uploaded so far, scheduling it\n" if $debug;
return $sku;
}
=head2 orders_in_shipping_job($job_id)
Lookup the C<amazon_mws_orders> table and return a list of
C<amazon_order_id> for the given shipping confirmation job. INTERNAL.
=cut
sub orders_in_shipping_job {
my ($self, $job_id) = @_;
die unless $job_id;
my $sth = $self->_exe_query($self->sqla->select(amazon_mws_orders => [qw/amazon_order_id/],
{
shipping_confirmation_job_id => $job_id,
shop_id => $self->_unique_shop_id,
}));
my @orders;
while (my $row = $sth->fetchrow_hashref) {
push @orders, $row->{amazon_order_id};
}
return @orders;
}
=head2 put_product_on_error(sku => $sku, timestamp_string => $timestamp, error_code => $error_code, error_msg => $error)
Register a custom error for the product $sku with error $error and
$timestamp as the timestamp string. The error is optional, and will be
"shop error" if not provided. The error code will be 1 if not provided.
=cut
sub put_product_on_error {
my ($self, %product) = @_;
die "Missing sku" unless $product{sku};
die "Missing timestamp" unless defined $product{timestamp_string};
my %identifier = (
shop_id => $self->_unique_shop_id,
sku => $product{sku},
);
my %errors = (
status => 'failed',
error_msg => $product{error_msg} || 'shop error',
error_code => $product{error_code} || 1,
timestamp_string => $product{timestamp_string},
);
# check if we have it
my $sth = $self->_exe_query($self->sqla
->select('amazon_mws_products',
[qw/sku/], { %identifier }));
if ($sth->fetchrow_hashref) {
$sth->finish;
print "Updating $product{sku} with error $product{error_msg}\n";
$self->_exe_query($self->sqla->update('amazon_mws_products',
\%errors, \%identifier));
}
lib/Amazon/MWS/Uploader.pm view on Meta::CPAN
sub job_timed_out {
my ($self, $job_row) = @_;
my $task = $job_row->{task};
die "Missing task in $job_row->{amws_job_id}" unless $task;
my $started = $job_row->{job_started_epoch};
die "Missing job_started_epoch in $job_row->{amws_job_id}" unless $started;
my $now = time();
my $timeout;
if ($task eq 'order_ack') {
$timeout = $self->order_ack_days_timeout * 60 * 60 * 24;
}
else {
$timeout = $self->job_hours_timeout * 60 * 60;
}
die "Something is off, timeout not defined" unless defined $timeout;
return unless $timeout;
my $elapsed = $now - $started;
if ($elapsed > $timeout) {
return $elapsed;
}
else {
return;
}
}
sub _print_or_warn_error {
my ($self, @args) = @_;
my $action;
if (@args) {
if ($self->quiet) {
$action = 'print';
print @args;
}
else {
$action = 'warn';
warn @args;
}
}
return ($action, @args);
}
=head2 purge_old_jobs($limit)
Eventually the jobs and feed tables grow and never get purged. You can
call this method to remove from the db all the feeds older than
C<order_ack_days_timeout> (30 by default).
To avoid too much load on the db, you can set the limit to purge the
jobs. Defaults to 500. Set it to 0 to disable it.
=cut
sub purge_old_jobs {
my ($self, $limit) = @_;
unless (defined $limit) {
$limit = 500;
}
my $range = time() - $self->order_ack_days_timeout * 60 * 60 * 24;
my @and = (
task => [qw/product_deletion
upload/],
job_started_epoch => { '<', $range },
[ -or => {
aborted => 1,
success => 1,
},
],
);
if (my $shop_id = $self->shop_id) {
push @and, shop_id => $shop_id;
}
my $sth = $self->_exe_query($self->sqla
->select(amazon_mws_jobs => [qw/amws_job_id shop_id/],
[ -and => \@and ] ));
my @purge_jobs;
my $count = 0;
while (my $where = $sth->fetchrow_hashref) {
if ($limit) {
last if $count++ > $limit;
}
push @purge_jobs, $where;
}
$sth->finish;
if (@purge_jobs) {
$self->_exe_query($self->sqla->delete(amazon_mws_feeds => \@purge_jobs));
$self->_exe_query($self->sqla->delete(amazon_mws_jobs => \@purge_jobs));
while (@purge_jobs) {
my $feed = shift @purge_jobs;
my $dir = path($self->feed_dir)->child($feed->{shop_id}, $feed->{amws_job_id});
if ($dir->exists) {
print "Removing " . $dir->canonpath . "\n"; # unless $self->quiet;
$dir->remove_tree;
}
else {
print "$dir doesn't exist\n";
}
}
}
else {
print "Nothing to purge\n" unless $self->quiet;
}
}
1;
( run in 1.041 second using v1.01-cache-2.11-cpan-b16cb0d3907 )