--- trunk/bin/irc-logger.pl 2008/03/12 18:14:57 117 +++ trunk/bin/irc-logger.pl 2008/03/14 15:26:33 125 @@ -2,7 +2,7 @@ use strict; $|++; -use POE qw(Component::IRC Component::Server::HTTP); +use POE qw(Component::IRC Component::Server::HTTP Component::Client::HTTP); use HTTP::Status; use DBI; use Regexp::Common qw /URI/; @@ -18,6 +18,9 @@ use Carp qw/confess/; use XML::Feed; use DateTime::Format::Flexible; +use IPC::DirQueue; +use File::Slurp; +use Encode; =head1 NAME @@ -47,6 +50,8 @@ ## CONFIG +my $debug = 0; + my $irc_config = { nick => 'irc-logger', server => 'irc.freenode.net', @@ -54,6 +59,8 @@ ircname => 'Anna the bot: try /msg irc-logger help', }; +my $queue_dir = './queue'; + my $HOSTNAME = `hostname -f`; chomp($HOSTNAME); @@ -75,7 +82,7 @@ my @channels = ( $CHANNEL ); -warn "# config = ", dump( $irc_config ), $/; +warn "## config = ", dump( $irc_config ) if $debug; my $NICK = $irc_config->{nick} or die "no nick?"; @@ -106,6 +113,8 @@ GetOptions( 'import-dircproxy:s' => \$import_dircproxy, 'log:s' => \$log_path, + 'queue:s' => \$queue_dir, + 'debug!' => \$debug, ); #$SIG{__DIE__} = sub { @@ -118,6 +127,15 @@ open(STDOUT, '>', $log_path) && warn "log to $log_path: $!\n"; +# queue + +if ( ! -d $queue_dir ) { + warn "## creating queue directory $queue_dir"; + mkdir $queue_dir or die "can't create queue directory $queue_dir: $!"; +} + +my $dq = IPC::DirQueue->new({ dir => $queue_dir }); + # HTML formatters my %escape = ('<'=>'<', '>'=>'>', '&'=>'&', '"'=>'"'); @@ -600,7 +618,7 @@ $a->{me} ||= 0; $a->{time} ||= strftime($TIMESTAMP,localtime()); - _log + _log "ARCHIVE", $a->{channel}, " ", $a->{me} ? "***" . $a->{nick} : "<" . $a->{nick} . ">", " " . $a->{message}; @@ -650,18 +668,34 @@ my $_stat; +POE::Component::Client::HTTP->spawn( + Alias => 'rss-fetch', + Timeout => 30, +); + +=head2 rss_parse_xml + + rss_parse_xml({ + url => 'http://www.example.com/rss', + send_rss_msgs => 42, + }); -sub rss_fetch { +=cut + +sub rss_parse_xml { my ($args) = @_; + warn "## rss_parse_xml ",dump( @_ ) if $debug; + # how many messages to send out when feed is seen for the first time? - my $send_rss_msgs = 1; + my $send_rss_msgs = $args->{send_rss_msgs}; + $send_rss_msgs = 1 if ! defined $send_rss_msgs; - _log "RSS fetch", $args->{url}; + _log "RSS fetch first $send_rss_msgs items from", $args->{url}; - my $feed = XML::Feed->parse(URI->new( $args->{url} )); + my $feed = XML::Feed->parse( \$args->{xml} ); if ( ! $feed ) { - _log("can't fetch RSS ", $args->{url}); + _log "can't fetch RSS ", $args->{url}, XML::Feed->errstr; return; } @@ -671,8 +705,10 @@ for my $entry ($feed->entries) { $total++; + my $seen_times = $_stat->{rss}->{seen}->{$args->{channel}}->{$feed->link}->{$entry->id}++; # seen allready? - next if $_stat->{rss}->{seen}->{$args->{channel}}->{$feed->link}->{$entry->id}++ > 0; + warn "## $seen_times ",$entry->id if $debug; + next if $seen_times > 0; sub prefix { my ($txt,$var) = @_; @@ -701,21 +737,29 @@ if ( my $tags = $entry->category ) { $tags =~ s!^\s+!!; $tags =~ s!\s*$! !; - $tags =~ s!\s+!// !g; + $tags =~ s!,?\s+!// !g; $msg .= prefix( ' ' , $tags ); } - if ( $args->{kernel} && $send_rss_msgs ) { + if ( $seen_times == 0 && $send_rss_msgs ) { $send_rss_msgs--; if ( ! $args->{private} ) { # FIXME bug! should be save_message -# save_message( channel => $args->{channel}, me => 1, nick => $NICK, message => $msg ); - $sth_insert_log->execute( $args->{channel}, 1, $NICK, $msg, 'now()' ); + save_message( channel => $args->{channel}, me => 1, nick => $NICK, message => $msg ); +# $sth_insert_log->execute( $args->{channel}, 1, $NICK, $msg, 'now()' ); } my ( $type, $to ) = ( 'notice', $args->{channel} ); ( $type, $to ) = ( 'privmsg', $args->{nick} ) if $args->{private}; - _log(">> $type $to", $msg); - $args->{kernel}->post( $irc => $type => $to, $msg ); + + _log("RSS generated $type to $to:", $msg); + # XXX enqueue message to send later + sub enqueue_post { + my $post = dump( @_ ); + warn "## queue_post $post\n" if $debug; + $dq->enqueue_string( $post ); + } + enqueue_post( $type => $to => $msg ); + $updates++; } } @@ -725,29 +769,37 @@ $sql .= qq{where id = } . $args->{id}; eval { $dbh->do( $sql ) }; - _log "RSS got $total items of which $updates new"; + _log "RSS got $total items of which $updates new from", $args->{url}; return $updates; } sub rss_fetch_all { - my $kernel = shift; + my ( $kernel, $send_rss_msgs ) = @_; + warn "## rss_fetch_all -- send_rss_msgs: $send_rss_msgs\n" if $debug; my $sql = qq{ select id, url, name, channel, nick, private from feeds where active is true }; # limit to newer feeds only if we are not sending messages out - $sql .= qq{ and last_update + delay < now() } if $kernel; + $sql .= qq{ and last_update + delay < now() } if defined ( $_stat->{rss}->{fetch} ); my $sth = $dbh->prepare( $sql ); $sth->execute(); warn "# ",$sth->rows," active RSS feeds\n"; my $count = 0; while (my $row = $sth->fetchrow_hashref) { - $row->{kernel} = $kernel if $kernel; - $count += rss_fetch( $row ); + $row->{send_rss_msgs} = $send_rss_msgs if defined $send_rss_msgs; + $_stat->{rss}->{fetch}->{ $row->{url} } = $row; + $kernel->post( + 'rss-fetch', + 'request', + 'rss_response', + HTTP::Request->new( GET => $row->{url} ), + ); + warn "## queued rss-fetch ", dump( $row ) if $debug; } - return "OK, fetched $count posts from " . $sth->rows . " feeds"; + return "OK, scheduled " . $sth->rows . " feeds for refresh"; } @@ -755,16 +807,22 @@ my $kernel = shift; $_stat->{rss}->{last_poll} ||= time(); my $dt = time() - $_stat->{rss}->{last_poll}; - warn "## rss_check_updates $dt > $rss_min_delay\n"; if ( $dt > $rss_min_delay ) { + warn "## rss_check_updates $dt > $rss_min_delay\n"; $_stat->{rss}->{last_poll} = time(); _log rss_fetch_all( $kernel ); } + # XXX send queue messages + while ( my $job = $dq->pickup_queued_job() ) { + my $data = read_file( $job->get_data_path ) || die "can't load ", $job->get_data_path, ": $!"; + my @data = eval $data; + _log "IRC post from queue:", @data; + $kernel->post( $irc => @data ); + $job->finish; + warn "## done queued job: ",dump( @data ) if $debug; + } } -# seed rss seen cache so we won't send out all items on startup -_log rss_fetch_all; - POE::Session->create( inline_states => { _start => sub { $_[KERNEL]->post( $irc => register => 'all' ); @@ -775,11 +833,13 @@ my $poco_object = $sender->get_heap(); _log "connected to",$poco_object->server_name(); $kernel->post( $sender => join => $_ ) for @channels; + # seen RSS cache, so don't send out messages + _log rss_fetch_all( $kernel, 0 ); undef; }, - irc_255 => sub { # server is done blabbing - $_[KERNEL]->post( $irc => join => $CHANNEL); - }, +# irc_255 => sub { # server is done blabbing +# $_[KERNEL]->post( $irc => join => $CHANNEL); +# }, irc_public => sub { my $kernel = $_[KERNEL]; my $nick = (split /!/, $_[ARG0])[0]; @@ -829,6 +889,7 @@ my $nick = (split /!/, $_[ARG0])[0]; my $msg = $_[ARG2]; my $channel = $_[ARG1]->[0]; + warn "# ARG = ",dump( @_[ARG0,ARG1,ARG2] ) if $debug; my $res = "unknown command '$msg', try /msg $NICK help!"; my @out; @@ -1073,7 +1134,7 @@ _log ">> registreted, so IDENTIFY"; $_[KERNEL]->post( $irc => privmsg => 'nickserv', "IDENTIFY $NICK" ); } else { - warn "## ignore $m\n"; + warn "## ignore $m\n" if $debug; } }, irc_snotice => sub { @@ -1092,6 +1153,19 @@ ""; 0; # false for signals }, + rss_response => sub { + my ($request_packet, $response_packet) = @_[ARG0, ARG1]; + my $request_object = $request_packet->[0]; + my $response_object = $response_packet->[0]; + + my $row = delete( $_stat->{rss}->{fetch}->{ $request_object->uri } ); + if ( $row ) { + $row->{xml} = $response_object->content; + rss_parse_xml( $row ); + } else { + warn "## can't find rss->fetch for ", $request_object->uri; + } + }, }, ); @@ -1370,7 +1444,7 @@

See history of all messages.

}; - $response->content( $html ); + $response->content( decode('utf-8',$html) ); warn "<< ", $request->method, " ", $request->uri, " created ", length($html), " bytes\n"; return RC_OK; }