stream.pl

#!/usr/bin/perl

use strict;
use warnings;
use Switch;

use lib qw(/home/aerith/local/bin/twitter/lib);

use JSON::Syck;
use Config::Pit;
use AnyEvent::Twitter::Stream;

use Util::Twitter::Stream::DB;
use Util::Twitter::Stream::DB::Schema;

binmode STDIN, ":utf8";
binmode STDOUT, ":utf8";
binmode STDERR, ":utf8";

my $config = Config::Pit::pit_get('aerith.twitter.com');
my $connected = 0;

while (1) {
        PROCESS: {
                my $condvar  = AnyEvent->condvar;
                my $streamer = AnyEvent::Twitter::Stream->new(
                        consumer_key    => $config->{consumer_key},
                        consumer_secret => $config->{consumer_secret},
                        token           => $config->{access_token},
                        token_secret    => $config->{access_token_secret},
                        method   => 'userstream',
                        on_tweet => sub {
                                $connected = 1;
                        },
                        on_event => sub {
                                $connected = 1;

                                my $data = shift;
                                my $name = $data->{event};

                                switch ($data->{event}) {
                                        case [qw/favorite unfavorite/] {
                                                my $favorite = Util::Twitter::Stream::DB->single($name => {tweet_id => $data->{target_object}->{id_str}});

                                                unless (defined $favorite) {
                                                        Util::Twitter::Stream::DB->insert($name => {
                                                                target_id_str => $data->{target}->{id_str} || '',
                                                                source_id_str => $data->{source}->{id_str} || '',
                                                                target_screen_name => $data->{target}->{screen_name} || '',
                                                                source_screen_name => $data->{source}->{screen_name} || '',
                                                                tweet_id => $data->{target_object}->{id_str} || 0,
                                                                tweet_content => $data->{target_object}->{text} || '',
                                                                utc_offset => $data->{target_object}->{utc_offset} || 0,
                                                                created_at => $data->{target_object}->{created_at} || '',
                                                                data => YAML::Syck::Dump($data->{target_object}) || ''
                                                        });
                                                }

                                                {
                                                        my $state = Util::Twitter::Stream::DB->single(favorite_state => {tweet_id => $data->{target_object}->{id_str}});
                                                        my $starred = $name eq 'favorite' ? 1 : 0 ;

                                                        if (defined $state) {
                                                                $state->update({is_starred => $starred});
                                                        } else {
                                                                Util::Twitter::Stream::DB->insert(favorite_state => {
                                                                        target_id_str => $data->{target}->{id_str} || '',
                                                                        source_id_str => $data->{source}->{id_str} || '',
                                                                        target_screen_name => $data->{target}->{screen_name} || '',
                                                                        source_screen_name => $data->{source}->{screen_name} || '',
                                                                        tweet_id => $data->{target_object}->{id_str} || 0,
                                                                        tweet_content => $data->{target_object}->{text} || '',
                                                                        utc_offset => $data->{target_object}->{utc_offset} || 0,
                                                                        created_at => $data->{target_object}->{created_at} || '',
                                                                        is_starred => $starred
                                                                });
                                                        }
                                                }
                                        }
                                }

                                print YAML::Syck::Dump($data);
                        },
                        on_keepalive => sub {
                                $connected = 0;
                        },
                        on_error => sub {
                                $connected = 0;
                                my $error = shift;
                                warn "ERROR: $error\n";
                                $condvar->send;
                        },
                        on_eof => sub {
                                $connected = 0;
                                $condvar->send;
                        },
                );

                $condvar->recv;

                undef $streamer;
        }

        my $wait = $connected ? 0 : 5 ;

        my $wait_cv = AE::cv;
        my $wait_t = AE::timer $wait, 0, $wait_cv;

        $wait_cv->recv;

        goto PROCESS;
}

__DATA__