root/bin/pl-thread

#!/usr/bin/perl -w

=head1 NAME

pl-thread - add message to threaded archive, or rebuild archive thread index

=head1 SYNOPSIS

pl-thread [--rebuild] LIST@DOMAIN [MESSAGE ...]

=head1 AUTHOR

Anselm Lingnau <anselm@anselms.net>

=head1 COPYRIGHT AND LICENSE

Copyright 2011 by Anselm Lingnau. This program is free software; you
may redistribute it and/or modify it under the terms of the GNU
General Public License as published by the Free Software Foundation;
either version 2 of the License, or (at your option) any later version.

This program 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 this program; if not, refer to
<URL:http://www.gnu.org/copyleft/gpl.html>. You may also obtain it by
writing to the Free Software Foundation, Inc., 59 Temple Place - Suite
330, Boston, MA 02111-1307, USA.

=cut

use strict;

use IO::File;
use Email::Address;
use Email::Date;
use Time::Piece;
use DBI;
use Encode qw/encode decode/;

use Lancelot::GetOpt qw/GetOptions/;

use Lancelot::Message;
use Lancelot::DB;

use Lancelot::Log qw/log/;

use Lancelot::Module::archive_store;

my $ddate = 0;
my $rebuild = 0;
my %opts;

GetOptions(\%opts, -argvmin => 1, "no list address specified",
                   "rebuild!");

my $listaddr = shift;

my ($list, $domain) = split /\@/, $listaddr, 2;

my $msgid = "<$list-%d\@msgid.$domain>";

my (@head, %threads, %last);
my (%subjects, %senders, %sender_info, %message, %last_subject);
my ($subject_count, $sender_count);
my $prev_date = 0;

my (@month, %month_index, $m_idx);

my $db = new Lancelot::DB $listaddr, { user => $opts{user} }
    or die "couldn't find mailing list $listaddr\n";

if ($opts{rebuild}) {
    rebuild_index($db);
} elsif (@ARGV) {
    foreach my $fn (@ARGV) {
        my $f = IO::File->new($fn);
        add_message(join "", <$f>);
        $f->close();
    }
} else {
    add_message(join "", <STDIN>);
}

sub rebuild_index {
    my $dir = $db->get_config("archive.directory");
    unless ($dir) {
        log "warn", "this list is not archived (no archive.directory)";
        return;
    }
    $dir = File::Spec->join($db->get_listdir(), $dir) if $dir !~ m!^/!;
    my $dbh = Lancelot::Module::archive_store::init_database($dir);

    print STDERR "Reading messages\n";
    my $min = 0;
    my $max = $db->get_parameter("message");
    # my $max = $min+200;
    foreach my $i ($min .. $max) {
        print STDERR "\n$i" if $i % 1000 == 0;
        read_message_bulk($dir, $i);
    }
    print STDERR "\n";
    print STDERR "Identifying threads\n";
    identify_threads(0, $max);
    print STDERR scalar(@head) . " threads found\n";
    output_threads($dbh);
    print STDERR "Done.\n";
}

sub read_message_bulk {
    my ($dir, $n) = @_;
    my $fn = sprintf("%s/%d/%02d", $dir, int($n/100), $n % 100);
    if (-e $fn) {
        my $f = IO::File->new($fn);
        my $msg = Lancelot::Message->new(join "", <$f>);
        $f->close;

        process_message_bulk($n, $msg);
    }
}

my @m;

sub process_message_bulk {
    my ($n, $msg) = @_;

    $m[$n]->{date} = Lancelot::Module::archive_store::determine_date($msg) || gmtime;
    if ($prev_date != 0 && (abs($m[$n]->{date}->epoch - $prev_date) > 15*24*3600)) {
        print STDERR "Unusual date: $n: $m[$n]->{date} (vs. $prev_date)\n";
    }
    $prev_date = $m[$n]->{date}->epoch;

    $m[$n]->{msgid} = $msg->header("Message-ID") || sprintf $msgid, $n;
    $message{$m[$n]->{msgid}} = $n;

    my $subject = Lancelot::Module::archive_store::simplify_subject($msg->header("Subject"));
    $m[$n]->{subject} = store_subject($subject);
    my $sender = $msg->header("From") || "Anonymous <anonymous\@$domain>";
    $m[$n]->{sender} = store_sender($sender);

    $m[$n]->{parent} = Lancelot::Module::archive_store::find_parent(undef, $n, $msg, $m[$n]->{subject},
                                   \%last_subject, \%message);
    $m[$n]->{root} = find_root($n);
    my ($year, $month) = ($m[$n]->{date}->year, $m[$n]->{date}->mon);
    my $m_id;
    unless ($m_id = $month_index{"$year-$month"}) {
        push @month, [ ++$m_idx, $year, $month ];
        $m_id = $month_index{"$year-$month"} = $month[$#month]->[0];
    }
    $m[$n]->{month_id} = $m_id;
    $m[$m[$n]->{root}]->{lastmsg_id} = $n;
}

sub find_root {
    my ($n) = @_;

    my $parent = $n;
    while ($n != -1) {
        $parent = $n;
        $n = $m[$n]->{parent};
    }
    return $parent;
}

sub store_subject {
    my ($s) = @_;
    return $subjects{$s} if defined $subjects{$s};
    return $subjects{$s} = $subject_count++;
}

sub store_sender {
    my @a = Email::Address->parse($_[0]);
    my $s = $a[0] ? $a[0]->name : "Anonymous";
    my $a = $a[0] ? $a[0]->address : "anonymous\@$domain";
    my $f = $a[0] ? $a[0]->format : qq|"$s" <$a>|;
    return $senders{$a} if defined $senders{$a};
    $sender_info{$a} = [$s, $a];
    return $senders{$a} = $sender_count++;
}

sub identify_threads {
    my ($from, $to) = @_;
    my %kids;

    # print STDERR "identify_threads: from=$from to=$to\n";
    for (my $i = $from; $i <= $to; $i++) {
        next unless $m[$i];
        push @head, $i if $i == $m[$i]->{root};
        push @{$kids{$m[$i]->{parent}}}, $i if $m[$i]->{parent} >= 0;
    }

    foreach my $h (@head) {
        my $tt = {};
        $threads{$h} = [];
        Lancelot::Module::archive_store::visit_kids(\%kids, $h, 0, 0,
                                                    $threads{$h}, $tt);
        $last{$h} = ${$threads{$h}}[-1];
        $m[$_]->{level} = $tt->{$_}->{level} foreach (keys %$tt);
    }
}

sub output_threads {
    my ($dbh) = @_;
    my $prev = -1;

    $dbh->begin_work;

    print STDERR "Writing senders ($sender_count) ...\n";
    my $sth = $dbh->prepare("INSERT INTO sender VALUES(?,?,?,?)");
    foreach my $a (keys %senders) {
        my $s = $sender_info{$a}->[0];
        $s = decode("MIME-Header", $s) if $s =~ /=\?.*\?[qb]\?/i;
        $s = encode("utf-8", $s);
        my $s_id = $db->get_subscription_id($a);
        $sth->execute($senders{$a}, $s, $a, $s_id);
    }

    print STDERR "Writing subjects ($subject_count) ...\n";
    $sth = $dbh->prepare("INSERT INTO subject VALUES(?,?)");
    foreach my $k (keys %subjects) {
        my $s = $k;
        $s = encode("utf-8", $s);
        $sth->execute($subjects{$k}, $s) or print STDERR "$k: " . $dbh->error;
    }

    print STDERR "Writing month records ...\n";
    $sth = $dbh->prepare("INSERT INTO month VALUES(?,?,?)");
    foreach my $m (@month) {
        $sth->execute(@$m);
    }

    print STDERR "\nWriting messages and threads ...\n";
    $sth = $dbh->prepare("INSERT INTO message VALUES(?,?,?,?,?,?,?,?,?,?,?)");
    my $tth = $dbh->prepare("INSERT INTO thread VALUES(?,?,?,?,?,?,?)");

    foreach my $hh (0 .. $#head) {
        my $h = $head[$hh];
        my $count = 0;
        my @t = @{$threads{$h}};
        unshift @t, $hh ? $last{$head[$hh-1]} : -1;
        push @t, $hh < $#head ? $head[$hh+1] : -1;
        foreach my $i (1 .. $#t-1) {
            my $k = $t[$i];
            my $m = $m[$k];
            $sth->execute($k, $m->{root}, $count, $m->{parent}, $m->{level},
                          $t[$i-1], $t[$i+1], $m->{sender}, $m->{date}->epoch,
                          $m->{subject}, $m->{msgid});
            $count++;
        }
        $tth->execute($h, $m[$t[1]]->{month_id}, $m[$t[1]]->{subject},
                      $m[$t[1]]->{date}->epoch, $m[$t[1]]->{lastmsg_id},
                      $hh ? $head[$hh-1] : -1, $hh < $#head ? $head[$hh+1] : -1);
    }

    print STDERR "Committing database ...\n";
    $dbh->commit;
}