lib/PublicInbox/LeiImport.pm

   1 # Copyright (C) 2021 all contributors <meta@public-inbox.org>
   2 # License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
   3
   4 # front-end for the "lei import" sub-command
   5 package PublicInbox::LeiImport;
   6 use strict;
   7 use v5.10.1;
   8 use parent qw(PublicInbox::IPC);
   9 use PublicInbox::Eml;
  10 use PublicInbox::InboxWritable qw(eml_from_path);
  11 use PublicInbox::PktOp qw(pkt_do);
  12
  13 sub _import_eml { # MboxReader callback
  14         my ($eml, $sto, $set_kw) = @_;
  15         $sto->ipc_do('set_eml', $eml, $set_kw ? $sto->mbox_keywords($eml) : ());
  16 }
  17
  18 sub import_done_wait { # dwaitpid callback
  19         my ($arg, $pid) = @_;
  20         my ($imp, $lei) = @$arg;
  21         $lei->child_error($?, 'non-fatal errors during import') if $?;
  22         my $ign = $lei->{sto}->ipc_do('done'); # PublicInbox::LeiStore::done
  23         $lei->dclose;
  24 }
  25
  26 sub import_done { # EOF callback for main daemon
  27         my ($lei) = @_;
  28         my $imp = delete $lei->{imp} or return;
  29         $imp->wq_wait_old(\&import_done_wait, $lei);
  30 }
  31
  32 sub net_merge_all { # via wq_broadcast
  33         my ($self, $net_new) = @_;
  34         my $net = $self->{lei}->{net};
  35         %$net = (%$net, %$net_new);
  36         pkt_do($self->{lei}->{pkt_op_p}, 'net_merge_done1') or
  37                 die "pkt_op_do net_merge_done1: $!";
  38 }
  39
  40 sub net_merge_continue { # first worker is done with auth
  41         my ($self, $net_new) = @_;
  42         $self->wq_broadcast('net_merge_all', $net_new);
  43 }
  44
  45 sub net_merge_complete {
  46         my ($self) = @_;
  47         for my $input (@{$self->{inputs}}) {
  48                 $self->wq_io_do('import_path_url', [], $input);
  49         }
  50         $self->wq_close(1);
  51 }
  52
  53 sub net_merge_done1 {
  54         my ($self) = @_;
  55         my $lei = $self->{lei};
  56         return if ++$lei->{nr_net_merge_done} != $self->{-wq_nr_workers};
  57         net_merge_complete($self);
  58 }
  59
  60 sub import_start {
  61         my ($lei) = @_;
  62         my $self = $lei->{imp};
  63         my $j = $lei->{opt}->{jobs} // scalar(@{$self->{inputs}}) || 1;
  64         if (my $net = $lei->{net}) {
  65                 # $j = $net->net_concurrency($j); TODO
  66         } else {
  67                 my $nproc = $self->detect_nproc;
  68                 $j = $nproc if $j > $nproc;
  69         }
  70         my $ops = { '' => [ \&import_done, $lei ] };
  71         my $auth = $lei->{auth};
  72         if ($auth) {
  73                 $ops->{net_merge} = [ \&net_merge_continue, $self ];
  74                 $ops->{net_merge_done1} = [ \&net_merge_done1, $self ];
  75         }
  76         $self->{-wq_nr_workers} = $j // 1; # locked
  77         my $op = $lei->workers_start($self, 'lei_import', undef, $ops);
  78         $self->wq_io_do('import_stdin', []) if $self->{0};
  79         net_merge_complete($self) if !$auth;
  80         while ($op && $op->{sock}) { $op->event_step }
  81 }
  82
  83 sub call { # the main "lei import" method
  84         my ($cls, $lei, @inputs) = @_;
  85         my $sto = $lei->_lei_store(1);
  86         $sto->write_prepare($lei);
  87         my ($net, @f, @d);
  88         $lei->{opt}->{kw} //= 1;
  89         my $self = $lei->{imp} = bless { inputs => \@inputs }, $cls;
  90         if ($lei->{opt}->{stdin}) {
  91                 @inputs and return $lei->fail("--stdin and @inputs do not mix");
  92                 $lei->check_input_format or return;
  93                 $self->{0} = $lei->{0};
  94         }
  95
  96         # TODO: do we need --format for non-stdin?
  97         my $fmt = $lei->{opt}->{'format'};
  98         # e.g. Maildir:/home/user/Mail/ or imaps://example.com/INBOX
  99         for my $input (@inputs) {
 100                 my $input_path = $input;
 101                 if ($input =~ m!\A(?:imap|nntp)s?://!i) {
 102                         require PublicInbox::NetReader;
 103                         $net //= PublicInbox::NetReader->new;
 104                         $net->add_url($input);
 105                 } elsif ($input_path =~ s/\A([a-z0-9]+)://is) {
 106                         my $ifmt = lc $1;
 107                         if (($fmt // $ifmt) ne $ifmt) {
 108                                 return $lei->fail(<<"");
 109 --format=$fmt and `$ifmt:' conflict
 110
 111                         }
 112                         if (-f $input_path) {
 113                                 require PublicInbox::MboxReader;
 114                                 PublicInbox::MboxReader->can($ifmt) or return
 115                                         $lei->fail("$ifmt not supported");
 116                         } elsif (-d _) {
 117                                 require PublicInbox::MdirReader;
 118                                 $ifmt eq 'maildir' or return
 119                                         $lei->fail("$ifmt not supported");
 120                         } else {
 121                                 return $lei->fail("Unable to handle $input");
 122                         }
 123                 } elsif (-f $input) { push @f, $input
 124                 } elsif (-d _) { push @d, $input
 125                 } else { return $lei->fail("Unable to handle $input") }
 126         }
 127         if (@f) { $lei->check_input_format(\@f) or return }
 128         if (@d) { # TODO: check for MH vs Maildir, here
 129                 require PublicInbox::MdirReader;
 130         }
 131         $self->{inputs} = \@inputs;
 132         if ($net) {
 133                 if (my $err = $net->errors) {
 134                         return $lei->fail($err);
 135                 }
 136                 $net->{quiet} = $lei->{opt}->{quiet};
 137                 $lei->{net} = $net;
 138                 require PublicInbox::LeiAuth;
 139                 $lei->{auth} = PublicInbox::LeiAuth->new($net);
 140         }
 141         import_start($lei);
 142 }
 143
 144 sub ipc_atfork_child {
 145         my ($self) = @_;
 146         my $lei = $self->{lei};
 147         delete $lei->{imp}; # drop circular ref
 148         $lei->lei_atfork_child;
 149         $self->SUPER::ipc_atfork_child;
 150         my $net = $lei->{net};
 151         if ($net && $self->{-wq_worker_nr} == 0) {
 152                 my $mics = $net->imap_common_init($lei);
 153                 PublicInbox::LeiAuth::net_merge($lei, $net);
 154                 $net->{mics_cached} = $mics;
 155         }
 156         undef;
 157 }
 158
 159 sub _import_fh {
 160         my ($lei, $fh, $input, $ifmt) = @_;
 161         my $set_kw = $lei->{opt}->{kw};
 162         eval {
 163                 if ($ifmt eq 'eml') {
 164                         my $buf = do { local $/; <$fh> } //
 165                                 return $lei->child_error(1 << 8, <<"");
 166 error reading $input: $!
 167
 168                         my $eml = PublicInbox::Eml->new(\$buf);
 169                         _import_eml($eml, $lei->{sto}, $set_kw);
 170                 } else { # some mbox (->can already checked in call);
 171                         my $cb = PublicInbox::MboxReader->can($ifmt) //
 172                                 die "BUG: bad fmt=$ifmt";
 173                         $cb->(undef, $fh, \&_import_eml, $lei->{sto}, $set_kw);
 174                 }
 175         };
 176         $lei->child_error(1 << 8, "<stdin>: $@") if $@;
 177 }
 178
 179 sub _import_maildir { # maildir_each_file cb
 180         my ($f, $sto, $set_kw) = @_;
 181         $sto->ipc_do('set_eml_from_maildir', $f, $set_kw);
 182 }
 183
 184 sub _import_imap { # imap_each cb
 185         my ($url, $uid, $kw, $eml, $sto, $set_kw) = @_;
 186         warn "$url $uid";
 187         $sto->ipc_do('set_eml', $eml, $set_kw ? @$kw : ());
 188 }
 189
 190 sub import_path_url {
 191         my ($self, $input) = @_;
 192         my $lei = $self->{lei};
 193         my $ifmt = lc($lei->{opt}->{'format'} // '');
 194         # TODO auto-detect?
 195         if ($input =~ m!\A(imap|nntp)s?://!i) {
 196                 $lei->{net}->imap_each($input, \&_import_imap, $lei->{sto},
 197                                         $lei->{opt}->{kw});
 198                 return;
 199         } elsif ($input =~ s!\A([a-z0-9]+):!!i) {
 200                 $ifmt = lc $1;
 201         }
 202         if (-f $input) {
 203                 open my $fh, '<', $input or return $lei->child_error(1 << 8, <<"");
 204 unable to open $input: $!
 205
 206                 _import_fh($lei, $fh, $input, $ifmt);
 207         } elsif (-d _ && (-d "$input/cur" || -d "$input/new")) {
 208                 return $lei->fail(<<EOM) if $ifmt && $ifmt ne 'maildir';
 209 $input appears to a be a maildir, not $ifmt
 210 EOM
 211                 PublicInbox::MdirReader::maildir_each_file($input,
 212                                         \&_import_maildir,
 213                                         $lei->{sto}, $lei->{opt}->{kw});
 214         } else {
 215                 $lei->fail("$input unsupported (TODO)");
 216         }
 217 }
 218
 219 sub import_stdin {
 220         my ($self) = @_;
 221         my $lei = $self->{lei};
 222         _import_fh($lei, delete $self->{0}, '<stdin>', $lei->{opt}->{'format'});
 223 }
 224
 225 1;