File Coverage

blib/lib/CPAN/Testers/Data/Uploads.pm

Criterion	Covered	Total	%
statement	288	316	91.1
branch	86	124	69.3
condition	32	53	60.3
subroutine	40	40	100.0
pod	7	7	100.0
total	453	540	83.8

package CPAN::Testers::Data::Uploads;

use strict;
use warnings;

use vars qw($VERSION);
$VERSION = '0.20';
$|++;

#----------------------------------------------------------------------------
# Library Modules

use base qw(Class::Accessor::Fast);

use CPAN::DistnameInfo;
use CPAN::Testers::Common::DBUtils;
use CPAN::Testers::Common::Article;
use Config::IniFiles;
use DBI;
use File::Basename;
use File::Find::Rule;
use File::Path;
use File::Slurp;
use Getopt::Long;
use IO::AtomicFile;
use IO::File;
use Net::NNTP;

#----------------------------------------------------------------------------
# Variables

my (%backups);
use constant    LASTMAIL    => '_lastmail';
use constant    LOGFILE     => '_uploads.log';
use constant    JOURNAL     => '_journal.sql';

my %phrasebook = (
    'FindAuthor'        => 'SELECT * FROM ixlatest WHERE author=?',

    'FindDistVersion'   => 'SELECT type FROM uploads WHERE author=? AND dist=? AND version=?',
    'InsertDistVersion' => 'INSERT INTO uploads (type,author,dist,version,filename,released) VALUES (?,?,?,?,?,?)',
    'UpdateDistVersion' => 'UPDATE uploads SET type=? WHERE author=? AND dist=? AND version=?',
    'FindDistTypes'     => 'SELECT * FROM uploads WHERE type=?',
    'DeleteAll'         => 'DELETE FROM uploads',
    'SelectAll'         => 'SELECT * FROM uploads',

    'DeleteAllIndex'    => 'DELETE FROM ixlatest',
    'DeleteIndex'       => 'DELETE FROM ixlatest WHERE dist=? AND author=?',
    'FindIndex'         => 'SELECT * FROM ixlatest WHERE dist=? AND author=?',
    'InsertIndex'       => 'INSERT INTO ixlatest (oncpan,author,version,released,dist) VALUES (?,?,?,?,?)',
    'AmendIndex'        => 'UPDATE ixlatest SET oncpan=? WHERE author=? AND version=? AND dist=?',
    'UpdateIndex'       => 'UPDATE ixlatest SET oncpan=?,version=?,released=? WHERE dist=? AND author=?',
    'BuildAuthorIndex'  => 'SELECT x.author,x.version,x.released,x.dist,x.type FROM (SELECT dist, MAX(released) AS mv FROM uploads WHERE author=? GROUP BY dist) AS y INNER JOIN uploads AS x ON x.dist=y.dist AND x.released=y.mv ORDER BY released',
    'GetAllAuthors'     => 'SELECT distinct(author) FROM uploads',

    'InsertRequest'     => 'INSERT INTO page_requests (type,name,weight) VALUES (?,?,5)',

    'ParseFailed'       => 'REPLACE INTO uploads_failed (source,type,dist,version,file,pause,created) VALUES (?,?,?,?,?,?,?)',

    # SQLite backup
    'CreateTable'       => 'CREATE TABLE uploads (type text, author text, dist text, version text, filename text, released int)',
);

my $extn = qr/\.(tar\.(gz|bz2)|tgz|zip)$/;

my %oncpan = (
    'backpan'   => 2,
    'cpan'      => 1,
    'upload'    => 1
);

#----------------------------------------------------------------------------
# The Application Programming Interface

sub new {
    my $class = shift;

    my $self = {};
    bless $self, $class;

    $self->_init_options(@_);
    return $self;
}

sub DESTROY {
    my $self = shift;
}

__PACKAGE__->mk_accessors(
    qw( uploads backpan cpan logfile logclean lastfile journal
        mgenerate mupdate mbackup mreindex ));

sub process {
    my $self = shift;
    $self->generate()       if($self->mgenerate);
    $self->reindex()        if($self->mreindex);
    $self->update()         if($self->mupdate);
    $self->backup()         if($self->mbackup);
}

sub generate {
    my $self = shift;
    my $db = $self->uploads;

    $self->_log("Restarting uploads database");
    $db->do_query($phrasebook{'DeleteAll'});

    $self->_log("Creating BACKPAN entries");
    my @files = File::Find::Rule->file()->name($extn)->in($self->backpan);
    $self->_parse_archive('backpan',$_)   for(@files);

    $self->_log("Creating CPAN entries");
    @files = File::Find::Rule->file()->name($extn)->in($self->cpan);
    $self->_parse_archive('cpan',$_)   for(@files);
}

sub reindex {
    my $self = shift;
    my $db = $self->uploads;

    $self->_log("Reindexing by author");

    my $next = $db->iterator('hash',$phrasebook{'GetAllAuthors'});
    while(my $author = $next->()) {
        $self->_log(".. author = $author->{author}");
        my @rows = $db->get_query('hash',$phrasebook{'BuildAuthorIndex'},$author->{author});
        for my $row (@rows) {
        $self->_log(".... dist = $row->{dist}, latest = $row->{version}");
            $db->do_query($phrasebook{'DeleteIndex'},$row->{dist},$row->{author});
            $db->do_query($phrasebook{'InsertIndex'},$oncpan{$row->{type}},$row->{author},$row->{version},$row->{released},$row->{dist});
        }
    }

    $self->_log("Reindexing authors done");
}

sub update {
    my $self = shift;
    my $db = $self->uploads;

    $self->_open_journal();

    # get list of db known CPAN distributions
    my @rows = $db->get_query('hash',$phrasebook{'FindDistTypes'},'cpan');
    my %cpan = map {$_->{filename} => $_} @rows;

    # get currently mirrored CPAN entries
    $self->_log("Updating CPAN entries");
    my @files = File::Find::Rule->file()->name($extn)->in($self->cpan);
    for(@files) {
        if(my $file = $self->_parse_archive('cpan',$_,1)) {
            delete $cpan{$file};
        } else {
            #$self->_log(".. cannot parse: $_");
        }
    }

    # demote any distributions no longer on CPAN
    $self->_log("Updating BACKPAN entries");
    for my $file (keys %cpan) {
        #$self->_log("backpan => $cpan{$file}->{dist} => $cpan{$file}->{version} => $cpan{$file}->{author} => $cpan{$file}->{released}");
        $self->_write_journal('UpdateDistVersion','backpan',$cpan{$file}->{author},$cpan{$file}->{dist},$cpan{$file}->{version});
        $db->do_query($phrasebook{'AmendIndex'},2,$cpan{$file}->{author},$cpan{$file}->{version},$cpan{$file}->{dist});
    }

    # read NNTP
    $self->_log("Updating NNTP entries");
    my ($nntp,$num,$first,$last) = $self->_nntp_connect();
    my $lastid = $self->_lastid();
    return    if($last <= $lastid);

    $self->_log(".. from $lastid to $last");
    for(my $id = $lastid+1; $id <= $last; $id++) {
        #$self->_log("NNTP ID = $id");
        my $article = join "", @{$nntp->article($id) || []};
        next    unless($article);
        my $object = CPAN::Testers::Common::Article->new($article);
        next    unless($object);
        $self->_log("... [$id] subject=".($object->subject()));

        my ($name,$version,$cpanid,$date,$filename);
        if($object->parse_upload()) {
            $name      = $object->distribution;
            $version   = $object->version;
            $cpanid    = $object->author;
            $date      = $object->epoch;
            $filename  = $object->filename;
        }

        #$self->_log("... name=$name");
        #$self->_log("... version=$version");
        #$self->_log("... cpanid=$cpanid");
        #$self->_log("... date=$date");

        next  unless($name && $version && $cpanid && $date);
        #$self->_log("upload => $name => $version => $cpanid => $date");

        $self->_update_index($cpanid,$version,$date,$name,1);
        my @rows = $db->get_query('array',$phrasebook{'FindDistVersion'},$cpanid,$name,$version);
        next    if(@rows);
        $self->_write_journal('InsertDistVersion','upload',$cpanid,$name,$version,$filename,$date);
    }

    $self->_lastid($last);
    $self->_close_journal();
}

sub backup {
    my $self = shift;
    my $db = $self->uploads;

    if(my @journals = $self->_find_journals()) {
        for my $driver (keys %backups) {
            if($driver =~ /(CSV|SQLite)/i && !$backups{$driver}{'exists'}) {
                $backups{$driver}{db}->do_query($phrasebook{'CreateTable'});
                $backups{$driver}{'exists'} = 1;
            }
        }

        for my $journal (@journals) {
            next    if($journal =~ /TMP$/); # don't process active journals
            $self->_log("Processing journal $journal");
            my $lines = $self->_read_journal($journal);
            for my $line (@$lines) {
                my ($phrase,@args) = @$line;
                for my $driver (keys %backups) {
                    $backups{$driver}{db}->do_query($phrasebook{$phrase},@args);
                }
            }

           $self->_done_journal($journal);
        }
        $self->_log("Processed journals");
    } else {
        for my $driver (keys %backups) {
            if($backups{$driver}{'exists'}) {
                $backups{$driver}{db}->do_query($phrasebook{'DeleteAll'});
            } elsif($driver =~ /(CSV|SQLite)/i) {
                $backups{$driver}{db}->do_query($phrasebook{'CreateTable'});
                $backups{$driver}{'exists'} = 1;
            }
        }

        $self->_log("Backup via DBD drivers");

        my $rows = $db->iterator('array',$phrasebook{'SelectAll'});
        while(my $row = $rows->()) {
            for my $driver (keys %backups) {
                $backups{$driver}{db}->do_query($phrasebook{'InsertDistVersion'},@$row);
            }
        }
    }

    # handle the CSV exception
    if($backups{CSV}) {
        $self->_log("Backup to CSV file");
        $backups{CSV}{db} = undef;  # close db handle
        my $fh1 = IO::File->new('uploads','r') or die "Cannot read temporary database file 'uploads'\n";
        my $fh2 = IO::File->new($backups{CSV}{dbfile},'w+') or die "Cannot write to CSV database file $backups{CSV}{dbfile}\n";
        while(<$fh1>) { print $fh2 $_ }
        $fh1->close;
        $fh2->close;
        unlink('uploads');
    }
}

sub help {
    my ($self,$full,$mess) = @_;

    print "\n$mess\n\n" if($mess);

    if($full) {
        print <

Usage: $0 --config= [-g] [-r] [-u] [-b] [-h] [-v]
        [--logfile=] [--logclean]
        [--lastmail=] [--journal=]

  --config=   database configuration file
  -g                generate new database
  -r                reindex database (*)
  -u                update existing database
  -b                backup database to portable files
  -h                this help screen
  -v                program version
  --logfile=  trace log file
  --logclean        overwrite exisiting log file
  --lastmail= last id file
  --journal=  SQL journal file path

Notes:
  * A generate request automatically includes a reindex.

HERE

    }

    print "$0 v$VERSION\n\n";
    exit(0);
}

#----------------------------------------------------------------------------
# Private Methods

sub _parse_archive {
    my ($self,$type,$file,$update) = @_;
    my $db = $self->uploads;
    my $dist = CPAN::DistnameInfo->new($file);

    my $name      = $dist->dist;      # "CPAN-DistnameInfo"
    my $version   = $dist->version;   # "0.02"
    my $cpanid    = $dist->cpanid;    # "GBARR"
    my $filename  = $dist->filename;  # "CPAN-DistnameInfo-0.02.tar.gz"
    my $date      = (stat($file))[9];

    unless($name && defined $version && $cpanid && $date) {
    	#$self->_log("PARSE: FAIL file=$file, $type => $name => $version => $cpanid => $date => $filename");
        $file =~ s!/opt/projects/CPAN/!!;
        $db->do_query($phrasebook{'ParseFailed'},$file,$type,$name,$version,$filename,$cpanid,$date);
    	return;
    }
    #$self->_log("$type => $name => $version => $cpanid => $date");

    my @rows = $db->get_query('array',$phrasebook{'FindDistVersion'},$cpanid,$name,$version);
    if(@rows) {
        if($type ne $rows[0]->[0]) {
            $db->do_query($phrasebook{'UpdateDistVersion'},$type,$cpanid,$name,$version);
            $self->_update_index($cpanid,$version,$date,$name,$oncpan{$type})
                if($update && $type ne 'backpan');
        }
    } else {
        $db->do_query($phrasebook{'InsertDistVersion'},$type,$cpanid,$name,$version,$filename,$date);
        $self->_update_index($cpanid,$version,$date,$name,$oncpan{$type})   if($update);
    }

    return $filename;
}

sub _update_index {
    my ($self,$author,$version,$date,$name,$oncpan) = @_;
    my $db = $self->uploads;

    my @index = $db->get_query('hash',$phrasebook{'FindIndex'},$name,$author);
    if(@index) {
        if($date > $index[0]->{released}) {
            $db->do_query($phrasebook{'UpdateIndex'},$oncpan,$version,$date,$name,$author);
            $self->_log("... index update [$author,$version,$date,$name,$oncpan]");
        }
    } else {
        $db->do_query($phrasebook{'InsertIndex'},$oncpan,$author,$version,$date,$name);
        $self->_log("... index insert [$author,$version,$date,$name,$oncpan]");
    }

    # add to page_requests table to update letter index pages and individual pages
    $db->do_query($phrasebook{'InsertRequest'},'ixauth',substr($author,0,1));
    $db->do_query($phrasebook{'InsertRequest'},'ixdist',substr($name,0,1));
    $db->do_query($phrasebook{'InsertRequest'},'author',$author);
    $db->do_query($phrasebook{'InsertRequest'},'distro',$name);
}

sub _nntp_connect {
    # connect to NNTP server
    my $nntp = Net::NNTP->new("nntp.perl.org") or die "Cannot connect to nntp.perl.org";
    my ($num,$first,$last) = $nntp->group("perl.cpan.uploads");

    return ($nntp,$num,$first,$last);
}

sub _lastid {
    my ($self,$id) = @_;
    my $f = $self->lastfile;

    unless( -f $f) {
        mkpath(dirname($f));
        overwrite_file( $f, 0 );
        $id ||= 0;
    }

    if($id) { overwrite_file( $f, $id ); }
    else    { $id = read_file($f); }

    return $id;
}

# generate atomic journal file name
sub _open_journal {
    my $self = shift;
    my @now  = localtime(time);
    my $file = sprintf "%s.%04d%02d%02d%02d%02d%02d", $self->journal, $now[5]+1900,$now[4]+1,$now[3],$now[2],$now[1],$now[0];
    $self->{current} = IO::AtomicFile->new($file,'w+') or die "Cannot write to journal file [$file]: $!\n";
}

sub _write_journal {
    my ($self,$phrase,@args) = @_;
    my $fh = $self->{current};

    print $fh "$phrase," . join(',',@args) . "\n";

    my $db = $self->uploads;
    $db->do_query($phrasebook{$phrase},@args);
}

sub _close_journal {
    my $self = shift;
    $self->{current}->close;
}

sub _find_journals {
    my $self = shift;
    my @files = glob($self->journal . '.*');
    return @files;
}

sub _read_journal {
    my ($self,$journal) = @_;
    my @lines;

    my $fh = IO::File->new($journal,'r') or die "Cannot read journal file [$journal]: $!\n";
    while(<$fh>) {
        my @fields = split(/,/);
        push @lines, \@fields;
    }
    $fh->close;
    return \@lines;
}

sub _done_journal {
    my ($self,$journal) = @_;
    my $cmd = "mv $journal logs";
    system($cmd);
}

sub _init_options {
    my $self = shift;
    my %hash  = @_;
    my %options;

    GetOptions( \%options,
        'config=s',
        'generate|g',
        'update|u',
        'reindex|r',
        'backup|b',
        'journal|j=s',
        'logfile|l=s',
        'logclean=s',
        'lastfile=s',
        'help|h',
        'version|v'
    );

    # default to API settings if no command line option
    for(qw(config generate update reindex fast backup help version)) {
        $options{$_} ||= $hash{$_}  if(defined $hash{$_});
    }

    $self->help(1)  if($options{help});
    $self->help(0)  if($options{version});

    $self->help(1,"Must specify at least one option from 'generate' (-g), 'reindex' (-r),\n'update' (-u)  and/or 'backup' (-b)")
                                                                        unless($options{generate} || $options{update} || $options{backup} || $options{reindex});
    $self->help(1,"Must specific the configuration file")               unless(   $options{config});
    $self->help(1,"Configuration file [$options{config}] not found")    unless(-f $options{config});

    # load configuration
    my $cfg = Config::IniFiles->new( -file => $options{config} );

    # configure sources
    if($options{generate}) {
        my $dir = $cfg->val('MASTER','BACKPAN');
        $self->help(1,"No source location for 'BACKPAN' in config file")    if(!   $dir);
        $self->help(1,"Cannot find source location for 'BACKPAN': [$dir]")  if(!-d $dir);
        $self->backpan($dir);
        $self->mgenerate(1);
        $self->mreindex(1);
    }
    if($options{generate} || $options{update}) {
        my $dir = $cfg->val('MASTER','CPAN');
        $self->help(1,"No source location for 'CPAN' in config file")   if(!   $dir);
        $self->help(1,"Cannot find source location for 'CPAN': [$dir]") if(!-d $dir);
        $self->cpan($dir);
    }
    if($options{reindex}) {
        $self->mreindex(1);
    }

    $self->mupdate(1)   if($options{update});
    $self->logfile(  $hash{logfile}  || $options{logfile}  || $cfg->val('MASTER','logfile'  ) || LOGFILE  );
    $self->logclean( $hash{logclean} || $options{logclean} || $cfg->val('MASTER','logclean' ) || 0        );
    $self->lastfile( $hash{lastfile} || $options{lastfile} || $cfg->val('MASTER','lastfile' ) || LASTMAIL );
    $self->journal(  $hash{journal}  || $options{journal}  || $cfg->val('MASTER','journal'  ) || JOURNAL  );

    # configure upload DB
    $self->help(1,"No configuration for UPLOADS database") unless($cfg->SectionExists('UPLOADS'));
    my %opts = map {$_ => ($cfg->val('UPLOADS',$_) || undef)} qw(driver database dbfile dbhost dbport dbuser dbpass);
    my $db = CPAN::Testers::Common::DBUtils->new(%opts);
    $self->help(1,"Cannot configure UPLOADS database") unless($db);
    $self->uploads($db);

    # configure backup DBs
    if($options{backup}) {
        $self->help(1,"No configuration for BACKUPS with backup option")    unless($cfg->SectionExists('BACKUPS'));

        my %available_drivers = map { $_ => 1 } DBI->available_drivers;
        my @drivers = $cfg->val('BACKUPS','drivers');
        for my $driver (@drivers) {
            unless($available_drivers{$driver}) {
                warn "No DBI support for '$driver', ignoring\n";
                next;
            }

            $self->help(1,"No configuration for backup option '$driver'")   unless($cfg->SectionExists($driver));

            my %opt = map {$_ => ($cfg->val($driver,$_) || undef)} qw(driver database dbfile dbhost dbport dbuser dbpass);
            $backups{$driver}{'exists'} = $driver =~ /SQLite/i ? -f $opt{database} : 1;

            # CSV is a bit of an oddity!
            if($driver =~ /CSV/i) {
                $backups{$driver}{'exists'} = 0;
                $backups{$driver}{'dbfile'} = $opt{dbfile};
                $opt{dbfile} = 'uploads';
                unlink($opt{dbfile});
            }

            $backups{$driver}{db} = CPAN::Testers::Common::DBUtils->new(%opt);
            $self->help(1,"Cannot configure BACKUPS database for '$driver'")   unless($backups{$driver}{db});
        }

        $self->mbackup(1)   if(keys %backups);
    }
}

sub _log {
    my $self = shift;
    my $log = $self->logfile or return;
    mkpath(dirname($log))   unless(-f $log);

    my $mode = $self->logclean ? 'w+' : 'a+';
    $self->logclean(0);

    my @dt = localtime(time);
    my $dt = sprintf "%04d/%02d/%02d %02d:%02d:%02d", $dt[5]+1900,$dt[4]+1,$dt[3],$dt[2],$dt[1],$dt[0];

    my $fh = IO::File->new($log,$mode) or die "Cannot write to log file [$log]: $!\n";
    print $fh "$dt ", @_, "\n";
    $fh->close;
}

q!Will code for a damn fine Balti!;

__END__