Skip to content

Commit 8793123

Browse files
mickeynmickeyn
authored
Merge pull request #16 from metacpan/mickey/latest
latest: port the rest of the logic
2 parents 507b380 + ce96fc9 commit 8793123

File tree

1 file changed

+110
-53
lines changed

1 file changed

+110
-53
lines changed

‎bin/latest.pl

Lines changed: 110 additions & 53 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@
77

88
use CPAN::DistnameInfo;
99
use Parse::CPAN::Packages::Fast;
10+
use Ref::Util qw< is_arrayref is_hashref >;
1011
use Regexp::Common qw< time >;
1112
use Time::Local qw< timelocal >;
1213

@@ -31,14 +32,14 @@
3132
my $minion;
3233
$minion = minion() if $queue;
3334

35+
my $node = config->{config}{es_test_node};
36+
3437
run();
3538

3639
sub run () {
3740
log_info {'Reading 02packages.details'};
3841

39-
40-
my $node = config->{config}{es_test_node};
41-
my $es = MetaCPAN::ES->new(
42+
my $es = MetaCPAN::ES->new(
4243
type => "file",
4344
node => $node
4445
);
@@ -54,7 +55,9 @@ ()
5455
my $dist = $packages->package($p)->distribution->dist;
5556
push @filter, $p if $dist && $dist eq $d;
5657
}
57-
log_info { "$distribution consists of " . scalar(@filter) . ' modules' };
58+
log_info {
59+
"$distribution consists of " . scalar(@filter) . ' modules'
60+
};
5861
}
5962

6063
# if we are just queueing a single distribution
@@ -98,7 +101,7 @@ ()
98101

99102
log_debug { sprintf( "Found %s modules", $scroll->total ) };
100103
log_debug { sprintf( "Found %s total modules", $found_total ) }
101-
if @$filter != $total and $filter == $module_filters->[-1];
104+
if @$filter != $total and $filter == $module_filters->[-1];
102105

103106
my $i = 0;
104107

@@ -109,18 +112,18 @@ ()
109112
my $file_data = $file->{_source};
110113

111114
# Convert module name into Parse::CPAN::Packages::Fast::Package object.
112-
my @modules =
113-
grep { defined }
114-
map { eval { $packages->package( $_->{name} ) } }
115-
@{ $file_data->{module} };
115+
my @modules = grep {defined}
116+
map {
117+
eval { $packages->package( $_->{name} ) }
118+
} @{ $file_data->{module} };
116119

117120
# For each of the packages in this file...
118-
foreach my $module ( @modules ) {
121+
foreach my $module (@modules) {
119122

120123
# Get P:C:P:F:Distribution (CPAN::DistnameInfo) object for package.
121124
my $dist = $module->distribution;
122125

123-
if ( $queue ) {
126+
if ($queue) {
124127
my $d = $dist->dist;
125128
_queue_latest($d)
126129
unless exists $queued_distributions{$d};
@@ -141,8 +144,10 @@ ()
141144

142145
# If multiple versions of a dist appear in 02packages
143146
# only mark the most recent upload as latest.
144-
next if ( $upgrade
145-
&& compare_dates( $upgrade->{date}, $file_data->{date} ) );
147+
next
148+
if ( $upgrade
149+
&& compare_dates( $upgrade->{date},
150+
$file_data->{date} ) );
146151
$upgrade{ $file_data->{distribution} } = $file_data;
147152
}
148153
elsif ( $file_data->{status} eq 'latest' ) {
@@ -152,7 +157,7 @@ ()
152157
}
153158
}
154159

155-
my $bulk = $es->bulk_helper( type => 'file' );
160+
my $bulk = $es->bulk( type => 'file' );
156161

157162
my %to_purge;
158163

@@ -164,7 +169,7 @@ ()
164169

165170
$to_purge{ $file_data->{download_url} } = 1;
166171

167-
reindex( $bulk, $file_data, 'latest' );
172+
_reindex( $bulk, $file_data, 'latest' );
168173
}
169174

170175
while ( my ( $release, $file_data ) = each %downgrade ) {
@@ -181,22 +186,16 @@ ()
181186

182187
$to_purge{ $file_data->{download_url} } = 1;
183188

184-
reindex( $bulk, $file_data, 'cpan' );
189+
_reindex( $bulk, $file_data, 'cpan' );
185190
}
186191

187192
$bulk->flush;
188-
$es->index_refresh();
193+
$es->index_refresh;
189194

190-
# Call Fastly to purge
191-
# purge_cpan_distnameinfos( [ map CPAN::DistnameInfo->new($_), keys %to_purge ] );
195+
# Call Fastly to purge
196+
# purge_cpan_distnameinfos( [ map CPAN::DistnameInfo->new($_), keys %to_purge ] );
192197
}
193198

194-
=head2
195-
196-
TODO: FIX: removing this comment causes syntax error ???!!!
197-
198-
=end
199-
200199
sub _add_module_filters ($filter) {
201200
my @module_filters;
202201
if (@$filter) {
@@ -224,13 +223,13 @@ ($filter)
224223
return +{
225224
query => {
226225
filtered => {
227-
query => { match_all => {} },
226+
query => { match_all => {} },
228227
filter => {
229228
bool => {
230229
must => [
231230
{
232231
nested => {
233-
path => 'module',
232+
path => 'module',
234233
query => { bool => { must => $filter } }
235234
}
236235
},
@@ -256,52 +255,110 @@ ( $dist = $distribution )
256255
);
257256
}
258257

259-
# Update the status for the release and all the files.
260-
sub reindex ( $bulk, $source, $status ) {
258+
sub _get_release ( $es, $author, $name ) {
259+
my $release = $es->search(
260+
body => {
261+
query => {
262+
bool => {
263+
must => [
264+
{ term => { author => $author } },
265+
{ term => { name => $name } },
266+
]
267+
}
268+
}
269+
},
270+
fields => [qw< id name >],
271+
);
261272

262-
=head2
273+
return {}
274+
unless is_arrayref( $release->{hits}{hits} )
275+
&& is_hashref( $release->{hits}{hits}[0] );
276+
277+
my $fields = $release->{hits}{hits}[0]{fields};
278+
279+
return +{
280+
id => $fields->{id},
281+
name => $fields->{name}[0],
282+
};
283+
}
284+
285+
sub _set_release_status ( $es, $release_id, $status ) {
286+
my $bulk = $es->bulk();
287+
$bulk->update( { id => $release_id, doc => { status => $status } } );
288+
$bulk->flush;
289+
}
290+
291+
# Update the status for the release and all the files.
292+
sub _reindex ( $bulk, $source, $status ) {
263293

264294
# Update the status on the release.
265-
my $release = $self->index->type('release')->get( {
266-
author => $source->{author},
267-
name => $source->{release},
268-
} );
295+
my $es_release = MetaCPAN::ES->new(
296+
type => "release",
297+
node => $node
298+
);
299+
300+
my $release
301+
= _get_release( $es_release, $source->{author}, $source->{release} );
302+
303+
unless ( keys %$release ) {
304+
log_info {
305+
sprintf( 'failed to fetch release: %s - %s',
306+
$source->{author}, $source->{release} )
307+
};
308+
return;
309+
}
310+
311+
_set_release_status( $es_release, $release->{id}, $status )
312+
unless $dry_run;
269313

270-
$release->_set_status($status);
271314
log_info {
272315
$status eq 'latest' ? 'Upgrading ' : 'Downgrading ',
273-
'release ', $release->name || q[];
316+
'release ', $release->{name}
274317
};
275-
$release->put unless ( $dry_run );
276318

277319
# Get all the files for the release.
278-
my $scroll = $self->index->type("file")->search_type('scan')->filter( {
279-
bool => {
280-
must => [
281-
{ term => { 'release' => $source->{release} } },
282-
{ term => { 'author' => $source->{author} } }
283-
]
284-
}
285-
} )->size(100)->source( [ 'status', 'file' ] )->raw->scroll;
320+
321+
my $es_file = MetaCPAN::ES->new(
322+
type => "file",
323+
node => $node
324+
);
325+
326+
my $scroll = $es_file->scroll(
327+
body => {
328+
query => {
329+
filtered => {
330+
query => { match_all => {} },
331+
filter => {
332+
bool => {
333+
must => [
334+
{
335+
term =>
336+
{ 'release' => $source->{release} }
337+
},
338+
{ term => { 'author' => $source->{author} } },
339+
],
340+
},
341+
},
342+
},
343+
},
344+
fields => [qw< name >],
345+
},
346+
);
286347

287348
while ( my $row = $scroll->next ) {
288-
my $source = $row->{_source};
289349
log_trace {
290-
$status eq 'latest' ? 'Upgrading ' : 'Downgrading ',
291-
'file ', $source->{name} || q[];
350+
sprintf( '%s file %s',
351+
( $status eq 'latest' ? 'Upgrading' : 'Downgrading' ),
352+
$row->{fields}{name}[0] )
292353
};
293354

294355
# Use bulk update to overwrite the status for X files at a time.
295356
$bulk->update( { id => $row->{_id}, doc => { status => $status } } )
296357
unless $dry_run;
297358
}
298-
299-
=end
300-
301359
}
302360

303-
sub compare_dates {
304-
my ( $self, $d1, $d2 ) = @_;
361+
sub compare_dates ( $d1, $d2 ) {
305362
for ( $d1, $d2 ) {
306363
if ( $_ =~ /$RE{time}{iso}{-keep}/ ) {
307364
$_ = timelocal( $7, $6, $5, $4, $3 - 1, $2 );

0 commit comments

Comments
 (0)