diff options
| -rw-r--r-- | MANIFEST | 1 | ||||
| -rw-r--r-- | lib/PublicInbox/LeiImport.pm | 129 | ||||
| -rw-r--r-- | t/lei-import-imap.t | 28 | ||||
| -rw-r--r-- | t/lei-import-maildir.t | 4 | ||||
| -rw-r--r-- | t/lei_to_mail.t | 10 |
5 files changed, 130 insertions, 42 deletions
@@ -365,6 +365,7 @@ t/kqnotify.t t/lei-convert.t t/lei-daemon.t t/lei-externals.t +t/lei-import-imap.t t/lei-import-maildir.t t/lei-import.t t/lei-mirror.t diff --git a/lib/PublicInbox/LeiImport.pm b/lib/PublicInbox/LeiImport.pm index 32f3a467..62a2a412 100644 --- a/lib/PublicInbox/LeiImport.pm +++ b/lib/PublicInbox/LeiImport.pm @@ -29,7 +29,7 @@ sub import_done { # EOF callback for main daemon $imp->wq_wait_old(\&import_done_wait, $lei); } -sub do_import { +sub import_start { my ($lei) = @_; my $ops = { '!' => [ $lei->can('fail_handler'), $lei ], @@ -39,7 +39,7 @@ sub do_import { }; ($lei->{pkt_op_c}, $lei->{pkt_op_p}) = PublicInbox::PktOp->pair($ops); my $self = $lei->{imp}; - my $j = $lei->{opt}->{jobs} // scalar(@{$self->{argv}}) || 1; + my $j = $lei->{opt}->{jobs} // scalar(@{$self->{inputs}}) || 1; if (my $nrd = $lei->{nrd}) { # $j = $nrd->net_concurrency($j); TODO } else { @@ -50,8 +50,8 @@ sub do_import { my $op = delete $lei->{pkt_op_c}; delete $lei->{pkt_op_p}; $self->wq_io_do('import_stdin', []) if $self->{0}; - for my $x (@{$self->{argv}}) { - $self->wq_io_do('import_path_url', [], $x); + for my $input (@{$self->{inputs}}) { + $self->wq_io_do('import_path_url', [], $input); } $self->wq_close(1); $lei->event_step_init; # wait for shutdowns @@ -61,60 +61,91 @@ sub do_import { } sub call { # the main "lei import" method - my ($cls, $lei, @argv) = @_; + my ($cls, $lei, @inputs) = @_; my $sto = $lei->_lei_store(1); $sto->write_prepare($lei); + my ($nrd, @f, @d); $lei->{opt}->{kw} //= 1; - my $self = $lei->{imp} = bless { argv => \@argv }, $cls; + my $self = $lei->{imp} = bless { inputs => \@inputs }, $cls; if ($lei->{opt}->{stdin}) { - @argv and return - $lei->fail("--stdin and locations (@argv) do not mix"); + @inputs and return $lei->fail("--stdin and @inputs do not mix"); $lei->check_input_format or return; $self->{0} = $lei->{0}; - } else { - my @f; - for my $x (@argv) { - if (-f $x) { push @f, $x } - elsif (-d _) { require PublicInbox::MdirReader } - else { - require PublicInbox::NetReader; - $lei->{nrd} //= PublicInbox::NetReader->new; - $lei->{nrd}->add_url($x); + } + + # TODO: do we need --format for non-stdin? + my $fmt = $lei->{opt}->{'format'}; + # e.g. Maildir:/home/user/Mail/ or imaps://example.com/INBOX + for my $input (@inputs) { + my $input_path = $input; + if ($input =~ m!\A(?:imap|nntp)s?://!i) { + require PublicInbox::NetReader; + $nrd //= PublicInbox::NetReader->new; + $nrd->add_url($input); + } elsif ($input_path =~ s/\A([a-z0-9]+)://is) { + my $ifmt = lc $1; + if (($fmt // $ifmt) ne $ifmt) { + return $lei->fail(<<""); +--format=$fmt and `$ifmt:' conflict + } - } - if (@f) { $lei->check_input_format(\@f) or return } - if ($lei->{nrd} && (my @err = $lei->{nrd}->errors)) { - return $lei->fail(@err); - } + if (-f $input_path) { + require PublicInbox::MboxReader; + PublicInbox::MboxReader->can($ifmt) or return + $lei->fail("$ifmt not supported"); + } elsif (-d _) { + require PublicInbox::MdirReader; + $ifmt eq 'maildir' or return + $lei->fail("$ifmt not supported"); + } else { + return $lei->fail("Unable to handle $input"); + } + } elsif (-f $input) { push @f, $input + } elsif (-d _) { push @d, $input + } else { return $lei->fail("Unable to handle $input") } } - do_import($lei); + if (@f) { $lei->check_input_format(\@f) or return } + if (@d) { # TODO: check for MH vs Maildir, here + require PublicInbox::MdirReader; + } + $self->{inputs} = \@inputs; + return import_start($lei) if !$nrd; + + if (my $err = $nrd->errors) { + return $lei->fail($err); + } + $nrd->{quiet} = $lei->{opt}->{quiet}; + $lei->{nrd} = $nrd; + require PublicInbox::LeiAuth; + my $auth = $lei->{auth} = PublicInbox::LeiAuth->new($nrd); + $auth->auth_start($lei, \&import_start, $lei); } sub ipc_atfork_child { my ($self) = @_; + delete $self->{lei}->{imp}; # drop circular ref $self->{lei}->lei_atfork_child; $self->SUPER::ipc_atfork_child; } sub _import_fh { - my ($lei, $fh, $x) = @_; + my ($lei, $fh, $input, $ifmt) = @_; my $set_kw = $lei->{opt}->{kw}; - my $fmt = $lei->{opt}->{'format'}; eval { - if ($fmt eq 'eml') { + if ($ifmt eq 'eml') { my $buf = do { local $/; <$fh> } // - return $lei->child_error(1 >> 8, <<""); -error reading $x: $! + return $lei->child_error(1 << 8, <<""); +error reading $input: $! my $eml = PublicInbox::Eml->new(\$buf); _import_eml($eml, $lei->{sto}, $set_kw); } else { # some mbox (->can already checked in call); - my $cb = PublicInbox::MboxReader->can($fmt) // - die "BUG: bad fmt=$fmt"; + my $cb = PublicInbox::MboxReader->can($ifmt) // + die "BUG: bad fmt=$ifmt"; $cb->(undef, $fh, \&_import_eml, $lei->{sto}, $set_kw); } }; - $lei->child_error(1 >> 8, "<stdin>: $@") if $@; + $lei->child_error(1 << 8, "<stdin>: $@") if $@; } sub _import_maildir { # maildir_each_file cb @@ -122,27 +153,45 @@ sub _import_maildir { # maildir_each_file cb $sto->ipc_do('set_eml_from_maildir', $f, $set_kw); } +sub _import_imap { # imap_each cb + my ($url, $uid, $kw, $eml, $sto, $set_kw) = @_; + warn "$url $uid"; + $sto->ipc_do('set_eml', $eml, $set_kw ? @$kw : ()); +} + sub import_path_url { - my ($self, $x) = @_; + my ($self, $input) = @_; my $lei = $self->{lei}; + my $ifmt = lc($lei->{opt}->{'format'} // ''); # TODO auto-detect? - if (-f $x) { - open my $fh, '<', $x or return $lei->child_error(1 >> 8, <<""); -unable to open $x: $! + if ($input =~ m!\A(imap|nntp)s?://!i) { + $lei->{nrd}->imap_each($input, \&_import_imap, $lei->{sto}, + $lei->{opt}->{kw}); + return; + } elsif ($input =~ s!\A([a-z0-9]+):!!i) { + $ifmt = lc $1; + } + if (-f $input) { + open my $fh, '<', $input or return $lei->child_error(1 << 8, <<""); +unable to open $input: $! - _import_fh($lei, $fh, $x); - } elsif (-d _ && (-d "$x/cur" || -d "$x/new")) { - PublicInbox::MdirReader::maildir_each_file($x, + _import_fh($lei, $fh, $input, $ifmt); + } elsif (-d _ && (-d "$input/cur" || -d "$input/new")) { + return $lei->fail(<<EOM) if $ifmt && $ifmt ne 'maildir'; +$input appears to a be a maildir, not $ifmt +EOM + PublicInbox::MdirReader::maildir_each_file($input, \&_import_maildir, $lei->{sto}, $lei->{opt}->{kw}); } else { - $lei->fail("$x unsupported (TODO)"); + $lei->fail("$input unsupported (TODO)"); } } sub import_stdin { my ($self) = @_; - _import_fh($self->{lei}, $self->{0}, '<stdin>'); + my $lei = $self->{lei}; + _import_fh($lei, delete $self->{0}, '<stdin>', $lei->{opt}->{'format'}); } 1; diff --git a/t/lei-import-imap.t b/t/lei-import-imap.t new file mode 100644 index 00000000..ee308723 --- /dev/null +++ b/t/lei-import-imap.t @@ -0,0 +1,28 @@ +#!perl -w +# Copyright (C) 2021 all contributors <meta@public-inbox.org> +# License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt> +use strict; use v5.10.1; use PublicInbox::TestCommon; +require_git 2.6; +require_mods(qw(DBD::SQLite Search::Xapian)); +my ($ro_home, $cfg_path) = setup_public_inboxes; +my ($tmpdir, $for_destroy) = tmpdir; +my $sock = tcp_server; +my $cmd = [ '-imapd', '-W0', "--stdout=$tmpdir/1", "--stderr=$tmpdir/2" ]; +my $env = { PI_CONFIG => $cfg_path }; +my $td = start_script($cmd, $env, { 3 => $sock }) or BAIL_OUT("-imapd: $?"); +my $host_port = tcp_host_port($sock); +undef $sock; +test_lei({ tmpdir => $tmpdir }, sub { + lei_ok(qw(q bytes:1..)); + my $out = json_utf8->decode($lei_out); + is_deeply($out, [ undef ], 'nothing imported, yet'); + lei_ok('import', "imap://$host_port/t.v2.0"); + lei_ok(qw(q bytes:1..)); + $out = json_utf8->decode($lei_out); + ok(scalar(@$out) > 1, 'got imported messages'); + is(pop @$out, undef, 'trailing JSON null element was null'); + my %r; + for (@$out) { $r{ref($_)}++ } + is_deeply(\%r, { 'HASH' => scalar(@$out) }, 'all hashes'); +}); +done_testing; diff --git a/t/lei-import-maildir.t b/t/lei-import-maildir.t index 5842e19e..d2b059ad 100644 --- a/t/lei-import-maildir.t +++ b/t/lei-import-maildir.t @@ -23,8 +23,8 @@ test_lei(sub { is_deeply($r2, $res, 'idempotent import'); rename("$md/cur/x:2,S", "$md/cur/x:2,SR") or BAIL_OUT "rename: $!"; - ok($lei->(qw(import), $md), 'import Maildir after +answered'); - ok($lei->(qw(q -d none s:boolean)), 'lei q after +answered'); + lei_ok('import', "maildir:$md", \'import Maildir after +answered'); + lei_ok(qw(q -d none s:boolean), \'lei q after +answered'); $res = json_utf8->decode($lei_out); like($res->[0]->{'s'}, qr/use boolean/, 'got expected result'); is_deeply($res->[0]->{kw}, ['answered', 'seen'], 'keywords set'); diff --git a/t/lei_to_mail.t b/t/lei_to_mail.t index 6a571660..72b90700 100644 --- a/t/lei_to_mail.t +++ b/t/lei_to_mail.t @@ -139,6 +139,16 @@ test_lei(sub { is($res->[1], undef, 'only one result'); }); +test_lei(sub { + lei_ok('import', "$mbox:$fn", \'imported mbox:/path') or diag $lei_err; + lei_ok(qw(q s:x), \'lei q works') or diag $lei_err; + my $res = json_utf8->decode($lei_out); + my $x = $res->[0]; + is($x->{'s'}, 'x', 'subject imported') or diag $lei_out; + is_deeply($x->{'kw'}, ['seen'], 'kw imported') or diag $lei_out; + is($res->[1], undef, 'only one result'); +}); + for my $zsfx (qw(gz bz2 xz)) { # XXX should we support zst, zz, lzo, lzma? my $zsfx2cmd = PublicInbox::LeiToMail->can('zsfx2cmd'); SKIP: { |