lib/App/MPD/Feeder.pm

   1 package App::MPD::Feeder;
   2
   3 use strict;
   4 use warnings;
   5 use utf8;
   6
   7 use App::MPD::Feeder::Options;
   8 use DBD::Pg;
   9 use DBI;
  10 use Getopt::Long;
  11 use IO::Async::Signal;
  12 use Log::Any qw($log);
  13 use Net::Async::MPD;
  14 use Object::Pad;
  15 use Syntax::Keyword::Try;
  16
  17
  18 class App::MPD::Feeder {
  19     has $cfg_file :reader;
  20     has $opt :reader;
  21     has $db;
  22     has $db_generation;
  23     has $db_needs_update :writer = 1;
  24     has $mpd :reader;
  25
  26 use constant DEFAULT_CONFIG_FILE => '/etc/mpd-feeder/mpd-feeder.conf';
  27
  28     ADJUST {
  29         Getopt::Long::Configure('pass_through');
  30         Getopt::Long::GetOptions('cfg|config=s' => \$cfg_file);
  31         Getopt::Long::Configure('no_pass_through');
  32
  33         $cfg_file //= DEFAULT_CONFIG_FILE if -e DEFAULT_CONFIG_FILE;
  34
  35         $self->configure;
  36
  37         $db_needs_update = 0 if $opt->skip_db_update;
  38     }
  39
  40     method configure {
  41         my $new_opt = App::MPD::Feeder::Options->new;
  42
  43         $new_opt->parse_config_file($cfg_file) if $cfg_file;
  44
  45         $new_opt->parse_command_line;
  46
  47         Log::Any::Adapter->set( Stderr => log_level => $new_opt->log_level );
  48
  49         $opt = $new_opt;
  50     }
  51
  52     method connect_mpd {
  53         return if $mpd;
  54
  55         my %conn = ( auto_connect => 1 );
  56         $conn{host} = $opt->mpd_host if $opt->mpd_host;
  57         $conn{port} = $opt->mpd_port if $opt->mpd_port;
  58
  59         $mpd = Net::Async::MPD->new(%conn);
  60
  61         $mpd->loop->add(
  62             IO::Async::Signal->new(
  63                 name       => 'HUP',
  64                 on_receipt => sub {
  65                     $log->debug("SIGHUP received. Stopping loop");
  66                     $mpd->loop->stop('reload');
  67                 },
  68             )
  69         );
  70
  71         $mpd->loop->add(
  72             IO::Async::Signal->new(
  73                 name       => 'USR1',
  74                 on_receipt => sub {
  75                     $log->debug("SIGUSR1 received. Dumping configuration to STDERR");
  76                     my $old = select \*STDERR;
  77                     try {
  78                         $opt->dump;
  79                     }
  80                     finally {
  81                         select $old;
  82                     }
  83                 },
  84             )
  85         );
  86     }
  87
  88     method connect_db {
  89         return if $db;
  90
  91         $db = DBI->connect( "dbi:Pg:dbname=" . $opt->db_path,
  92             $opt->db_user, $opt->db_password,
  93             { RaiseError => 1, PrintError => 0, AutoCommit => 1 } );
  94
  95         $log->info( "Connected to database " . $opt->db_path );
  96         $db_generation = $self->db_get_option('generation');
  97         $log->debug("DB generation is $db_generation");
  98         $self->update_db;
  99     }
 100
 101     method db_get_option($name) {
 102         my $sth = $db->prepare_cached("select $name from options");
 103         $sth->execute;
 104         my @result = $sth->fetchrow_array;
 105         $sth->finish;
 106         undef $sth;
 107
 108         return $result[0];
 109     }
 110
 111     method db_set_option( $name, $value ) {
 112         my $sth = $db->prepare_cached("update options set $name = ?");
 113         $sth->execute($value);
 114     }
 115
 116     method db_store_song($song, $artist, $album) {
 117         return unless length($song) and length($artist) and length($album);
 118
 119         $db->prepare_cached(
 120             <<'SQL')->execute( $song, $artist, $album, $db_generation );
 121 INSERT INTO songs(path, artist, album, generation)
 122 VALUES($1, $2, $3, $4)
 123 ON CONFLICT ON CONSTRAINT songs_pkey DO
 124 UPDATE SET artist = $2
 125          , album = $3
 126          , generation = $4
 127 SQL
 128         $db->prepare_cached(<<'SQL')->execute( $artist, $album, $db_generation );
 129 INSERT INTO albums(artist, album, generation)
 130 VALUES($1, $2, $3)
 131 ON CONFLICT ON CONSTRAINT albums_pkey DO
 132 UPDATE SET generation = $3
 133 SQL
 134         $db->prepare_cached(<<'SQL')->execute( $artist, $db_generation );
 135 INSERT INTO artists(artist, generation)
 136 VALUES($1, $2)
 137 ON CONFLICT ON CONSTRAINT artists_pkey DO
 138 UPDATE SET generation = $2
 139 SQL
 140     }
 141
 142     method db_remove_stale_entries {
 143         my $sth =
 144             $db->prepare_cached('DELETE FROM songs WHERE generation <> ?');
 145         $sth->execute($db_generation);
 146         $log->debug( sprintf( "Deleted %d stale songs", $sth->rows ) );
 147
 148         $sth = $db->prepare_cached('DELETE FROM albums WHERE generation <> ?');
 149         $sth->execute($db_generation);
 150         $log->debug( sprintf( "Deleted %d stale albums", $sth->rows ) );
 151
 152         $sth =
 153             $db->prepare_cached('DELETE FROM artists WHERE generation <> ?');
 154         $sth->execute($db_generation);
 155         $log->debug( sprintf( "Deleted %d stale artists", $sth->rows ) );
 156     }
 157
 158     method db_note_song_qeued($item) {
 159         $db->prepare_cached(
 160             'UPDATE songs SET last_queued=current_timestamp WHERE path=?')
 161             ->execute( $item->{song} );
 162         $db->prepare_cached(
 163             'UPDATE artists SET last_queued=CURRENT_TIMESTAMP WHERE artist=?')
 164             ->execute( $item->{artist} );
 165         $db->prepare_cached(
 166             'UPDATE albums SET last_queued=CURRENT_TIMESTAMP WHERE artist=? AND album=?'
 167         )->execute( $item->{artist}, $item->{album} );
 168     }
 169
 170     method update_db($force = undef) {
 171         if (!$db_needs_update and !$force) {
 172             $log->debug("Skipping DB update");
 173             return;
 174         }
 175
 176         $log->info('Updating song database');
 177         $self->connect_mpd;
 178         $self->connect_db;
 179
 180         my $rows = $mpd->send('listallinfo')->get;
 181         try {
 182             $db->begin_work;
 183
 184             $db_generation++;
 185
 186             my $song_count;
 187
 188             foreach my $entry (@$rows) {
 189                 next unless exists $entry->{file};
 190                 $self->db_store_song( $entry->{file},
 191                     $entry->{AlbumArtist} // $entry->{Artist},
 192                     $entry->{Album} );
 193                 $song_count++;
 194             }
 195
 196             $log->info("Updated data about $song_count songs");
 197
 198             $self->db_remove_stale_entries;
 199
 200             $self->db_set_option( generation => $db_generation );
 201
 202             $db->commit;
 203
 204             $db_needs_update = 0;
 205         }
 206         catch {
 207             my $err = $@;
 208
 209             $db_generation--;
 210
 211             $db->rollback;
 212
 213             die $err;
 214         }
 215     }
 216
 217     method db_find_suitable_songs($num) {
 218         $self->connect_db;
 219         $self->update_db;
 220
 221         my @result;
 222         my $sql = <<SQL;
 223 SELECT s.path, s.artist, s.album
 224 FROM songs s
 225 JOIN artists ar ON ar.artist=s.artist
 226 JOIN albums al ON al.album=s.album AND al.artist=s.artist
 227 WHERE (s.last_queued IS NULL OR s.last_queued < CURRENT_TIMESTAMP - (? || ' seconds')::interval)
 228   AND (ar.last_queued IS NULL OR ar.last_queued < CURRENT_TIMESTAMP - (? || ' seconds')::interval)
 229   AND (al.last_queued IS NULL OR al.last_queued < CURRENT_TIMESTAMP - (? || ' seconds')::interval)
 230   AND NOT EXISTS (SELECT 1 FROM unwanted_artists uar WHERE uar.artist = s.artist)
 231   AND NOT EXISTS (SELECT 1 FROM unwanted_albums  ual WHERE ual.album  = s.album)
 232 ORDER BY random()
 233 LIMIT ?
 234 SQL
 235         my @params = (
 236             $opt->min_song_interval,  $opt->min_artist_interval,
 237             $opt->min_album_interval, $num,
 238         );
 239         my $sth = $db->prepare_cached($sql);
 240         $sth->execute(@params);
 241         while ( my @row = $sth->fetchrow_array ) {
 242             push @result,
 243                 { song => $row[0], artist => $row[1], album => $row[2] };
 244         }
 245         undef $sth;
 246
 247         if (scalar(@result) == $num and  $log->is_debug) {
 248             $sql =~ s/^SELECT .+$/SELECT COUNT(DISTINCT s.path)/m;
 249             $sql =~ s/^ORDER BY .+$//m;
 250             $sql =~ s/^LIMIT .+$//m;
 251             $log->debug($sql);
 252             my $sth = $db->prepare_cached($sql);
 253             pop @params;
 254             $sth->execute(@params);
 255             my $count = ($sth->fetchrow_array)[0];
 256             $sth->finish;
 257
 258             $sth = $db->prepare_cached('SELECT COUNT(*) FROM songs');
 259             $sth->execute;
 260             my $total = ($sth->fetchrow_array)[0];
 261             $log->debug(
 262                 sprintf(
 263                     "Number of songs meeting the criteria: %d out of total %d (%5.2f%%)",
 264                     $count, $total, 100.0 * $count / $total
 265                 )
 266             );
 267             $sth->finish;
 268
 269             $sql = <<SQL;
 270 SELECT COUNT(*)
 271 FROM songs s
 272 WHERE (s.last_queued IS NULL OR s.last_queued < CURRENT_TIMESTAMP - (? || ' seconds')::interval)
 273 UNION
 274 SELECT COUNT(*)
 275 FROM songs
 276 SQL
 277             $sth = $db->prepare_cached($sql);
 278             $sth->execute($opt->min_song_interval);
 279             $count = ($sth->fetchrow_array)[0];
 280             $total = ($sth->fetchrow_array)[0];
 281             $sth->finish;
 282
 283             $log->debug(
 284                 sprintf(
 285                     "Number of songs not queued soon: %d out of total %d (%5.2f%%)",
 286                     $count, $total, 100.0 * $count / $total
 287                 )
 288             );
 289             $sth->finish;
 290
 291             $sql = <<SQL;
 292 SELECT COUNT(*)
 293 FROM artists ar
 294 WHERE (ar.last_queued IS NULL OR ar.last_queued < CURRENT_TIMESTAMP - (? || ' seconds')::interval)
 295 UNION
 296 SELECT COUNT(*)
 297 FROM artists
 298 SQL
 299             $sth = $db->prepare_cached($sql);
 300             $sth->execute($opt->min_artist_interval);
 301             $count = ($sth->fetchrow_array)[0];
 302             $total = ($sth->fetchrow_array)[0];
 303             $log->debug(
 304                 sprintf(
 305                     "Number of artists not queued soon: %d out of total %d (%5.2f%%)",
 306                     $count, $total, 100.0 * $count / $total
 307                 )
 308             );
 309             $sth->finish;
 310
 311             $sql = <<SQL;
 312 SELECT COUNT(*)
 313 FROM albums al
 314 WHERE (al.last_queued IS NULL OR al.last_queued < CURRENT_TIMESTAMP - (? || ' seconds')::interval)
 315 UNION
 316 SELECT COUNT(*)
 317 FROM albums
 318 SQL
 319             $sth = $db->prepare_cached($sql);
 320             $sth->execute($opt->min_album_interval);
 321             $count = ($sth->fetchrow_array)[0];
 322             $total = ($sth->fetchrow_array)[0];
 323             $log->debug(
 324                 sprintf(
 325                     "Number of albums not queued soon: %d out of total %d (%5.2f%%)",
 326                     $count, $total, 100.0 * $count / $total
 327                 )
 328             );
 329             $sth->finish;
 330
 331             undef $sth;
 332         }
 333
 334         return @result;
 335     }
 336
 337     method db_add_unwanted_artist($artist) {
 338         $self->connect_db;
 339
 340         try {
 341             $db->do(
 342                 <<'SQL',
 343 INSERT INTO unwanted_artists(artist, generation)
 344 VALUES($1, $2)
 345 SQL
 346                 undef, $artist, $db_generation
 347             );
 348             return 1;
 349         }
 350         catch {
 351             my $err = $@;
 352
 353             $log->debug("PostgreSQL error: $err");
 354             $log->debug( "SQLSTATE = " . $db->state );
 355             return 0 if $db->state eq '23505';
 356
 357             die $err;
 358         }
 359     }
 360
 361     method db_del_unwanted_artist($artist) {
 362         $self->connect_db;
 363
 364         return 1 == $db->do(
 365             <<'SQL',
 366 DELETE FROM unwanted_artists
 367 WHERE artist = $1
 368 SQL
 369             undef, $artist
 370         );
 371     }
 372
 373     method queue_songs($num = undef, $callback = undef) {
 374         if (!defined $num) {
 375             $self->connect_mpd;
 376             $mpd->send('playlist')->on_done(
 377                 sub {
 378                     my $present = scalar @{ $_[0] };
 379
 380                     $log->notice( "Playlist contains $present songs. Wanted: "
 381                             . $opt->target_queue_length );
 382                     if ( $present < $opt->target_queue_length ) {
 383                         $self->queue_songs(
 384                             $opt->target_queue_length - $present, $callback );
 385                     }
 386                     else {
 387                         $callback->() if $callback;
 388                     }
 389                 }
 390             );
 391
 392             return;
 393         }
 394
 395         my @list = $self->db_find_suitable_songs($num);
 396
 397         die "Found no suitable songs" unless @list;
 398
 399         if ( @list < $num ) {
 400             $log->warn(
 401                 sprintf(
 402                     'Found only %d suitable songs instead of %d',
 403                     scalar(@list), $num
 404                 )
 405             );
 406         }
 407
 408         $log->info("About to add $num songs to the playlist");
 409
 410         my @paths;
 411         for my $song (@list) {
 412             my $path = $song->{song};
 413             $path =~ s/"/\\"/g;
 414             push @paths, $path;
 415         }
 416
 417         $log->debug( "Adding " . join( ', ', map {"«$_»"} @paths ) );
 418         my @commands;
 419         for (@paths) {
 420             push @commands, [ add => "\"$_\"" ];
 421         }
 422         $self->connect_mpd;
 423         my $f = $mpd->send( \@commands );
 424         $f->on_fail( sub { die @_ } );
 425         $f->on_done(
 426             sub {
 427                 $self->db_note_song_qeued($_) for @list;
 428                 $callback->(@_) if $callback;
 429             }
 430         );
 431     }
 432
 433     method prepare_to_wait_idle {
 434         $log->trace('declaring idle mode');
 435         $mpd->send('idle database playlist')->on_done(
 436             sub {
 437                 my $result = shift;
 438
 439                 if ( $result->{changed} eq 'database' ) {
 440                     $db_needs_update = 1;
 441                     $self->prepare_to_wait_idle;
 442                 }
 443                 elsif ( $result->{changed} eq 'playlist' ) {
 444                     $self->queue_songs( undef,
 445                         sub { $self->prepare_to_wait_idle } );
 446                 }
 447                 else {
 448                     use JSON;
 449                     $log->warn(
 450                         "Unknown result from idle: " . to_json($result) );
 451                     $self->prepare_to_wait_idle;
 452                 }
 453             }
 454         );
 455     }
 456
 457     method run {
 458         $mpd->on(
 459             close => sub {
 460                 die "Connection to MPD lost";
 461             }
 462         );
 463
 464         $self->prepare_to_wait_idle;
 465     }
 466
 467     method stop {
 468         undef $mpd;
 469
 470         if ($db) {
 471             if ($db->{ActiveKids}) {
 472                 $log->warn("$db->{ActiveKids} active DB statements");
 473                 for my $st ( @{ $db->{ChildHandles} } ) {
 474                     next unless $st->{Active};
 475                     while(my($k,$v) = each %$st) {
 476                         $log->debug("$k = ".($v//'<NULL>'));
 477                     }
 478                 }
 479             }
 480
 481             $db->disconnect;
 482             undef $db;
 483         }
 484     }
 485
 486     method run_loop {
 487         $self->connect_db;
 488
 489         for ( ;; ) {
 490             $self->queue_songs( undef, sub { $self->run } );
 491
 492             $log->debug("Entering event loop. PID=$$");
 493
 494             my $result = $mpd->loop->run;
 495             $log->trace( "Got loop result of " . ( $result // 'undef' ) );
 496
 497             if ( 'reload' eq $result ) {
 498                 $log->notice("disconnecting");
 499                 $self->stop;
 500
 501                 exec( "$0", '--config', $self->cfg_file, '--skip-db-update' );
 502             }
 503         }
 504     }
 505 }
 506