<commit_before>package MediaWords::Job::FetchTopicTweets;

=head1 NAME

MediaWords::Job::FetchTopicTweets - fetch tweets relevant to a topic and store them in the topic_tweets table

=head1 DESCRIPTION

Use the Crimson Hexagon API to lookup tweets relevant to a topic, then fetch each of those tweets from twitter.

=cut

use strict;
use warnings;

use Moose;
with 'MediaWords::AbstractJob';

use Modern::Perl "2015";
use MediaWords::CommonLibs;

use List::MoreUtils;
use Net::Twitter;
use Date::Parse;

use MediaWords::DB;
use MediaWords::Util::DateTime;
use MediaWords::Util::JSON;
use MediaWords::Util::Web;

my $_api_host = undef;

=head1 METHODS

=cut

# fetch the list of tweets from the ch api.  return only 500 posts unless $fetch_10k_posts is true.
sub _fetch_ch_posts ($$)
{
    my ( $ch_monitor_id, $day ) = @_;

    my $ua = MediaWords::Util::Web::UserAgentDetermined();
    $ua->max_size( 100 * 1024 * 1024 );
    $ua->timeout( 90 );
    $ua->timing( '1,2,4,8,16,32,64,128,256,512' );

    my $key = MediaWords::Util::Config::get_config->{ crimson_hexagon }->{ key };
    LOGDIE( "no crimson hexagon key in mediawords.yml at //crimson_hexagon/key." ) unless ( $key );

    my $next_day = MediaWords::Util::SQL::increment_day( $day );

    my $url = _get_ch_api_url() . "?auth=$key&id=$ch_monitor_id&start=$day&end=$next_day&extendLimit=true";

    DEBUG( "CH URL: $url" );

    my $response = $ua->get( $url );

    if ( !$response->is_success )
    {
        LOGDIE( "error fetching posts: " . $response->as_string );
    }

    my $decoded_content = $response->decoded_content;

    my $data;
    eval { $data = MediaWords::Util::JSON::decode_json( $decoded_content ); };

    LOGDIE( "Unable to parse json from url $url: $@ " . substr( $decoded_content, 0, 1024 ) . " ..." ) unless ( $data );

    LOGDIE( "Unknown response status: '$data->{ status }'" ) unless ( $data->{ status } eq 'success' );

    return $data;
}

# add a row to topic_tweet_day if it does not already exist.  fetch data for new row from CH
sub _add_topic_tweet_single_day ($$$)
{
    my ( $db, $topic, $day ) = @_;

    my $topic_tweet_day = $db->query( <<SQL, $topic->{ topics_id }, $day )->hash;
select * from topic_tweet_days where topics_id = \$1 and day = \$2
SQL

    return if ( $topic_tweet_day && $topic_tweet_day->{ tweets_fetched } );

    # if we have a ttd but had not finished fetching tweets, delete it and start over
    $db->delete_by_id( 'topic_tweet_days', $topic_tweet_day->{ topic_tweet_days_id } ) if ( $topic_tweet_day );

    my $ch_posts = _fetch_ch_posts( $topic->{ ch_monitor_id }, $day );

    my $tweet_count = $ch_posts->{ totalPostsAvailable };

    my $num_ch_tweets = scalar( @{ $ch_posts->{ posts } } );

    $topic_tweet_day = $db->create(
        'topic_tweet_days',
        {
            topics_id      => $topic->{ topics_id },
            day            => $day,
            tweet_count    => $tweet_count,
            num_ch_tweets  => $num_ch_tweets,
            tweets_fetched => 'false'
        }
    );

    $topic_tweet_day->{ ch_posts } = $ch_posts;

    return $topic_tweet_day;
}

# for each day within the topic date range, find or create a topic_tweet_day row.
sub _add_topic_tweet_days ($$)
{
    my ( $db, $topic ) = @_;

    my $twitter = _get_twitter_handle();

    my $date = $topic->{ start_date };
    while ( $date le $topic->{ end_date } )
    {
        my $topic_tweet_day = _add_topic_tweet_single_day( $db, $topic, $date );
        _fetch_tweets_for_day( $db, $twitter, $topic, $topic_tweet_day ) if ( $topic_tweet_day );

        $date = MediaWords::Util::SQL::increment_day( $date );
    }

}

# given a set of ch_posts, fetch data from twitter about each tweet and attach it under the $ch->{ tweet } field
sub _add_tweets_to_ch_posts
{
    my ( $twitter, $ch_posts ) = @_;

    DEBUG( "fetching tweets for " . scalar( @{ $ch_posts } ) . " tweets" );

    LOGDIE( "more than 100 posts in $ch_posts" ) unless ( scalar( @{ $ch_posts } ) <= 500 );

    my $ch_post_lookup = {};
    for my $ch_post ( @{ $ch_posts } )
    {
        LOGDIE( "Unable to parse id from tweet url: $ch_post->{ url }" ) unless ( $ch_post->{ url } =~ m~/status/(\d+)~ );
        my $tweet_id = $1;

        $ch_post->{ tweet_id } = $tweet_id;
        $ch_post_lookup->{ $tweet_id } = $ch_post;
    }

    my $tweet_ids = [ keys( %{ $ch_post_lookup } ) ];

    my $tweets;
    my $twitter_retries = 10;
    while ( !$tweets && $twitter_retries-- )
    {
        eval {
            $tweets = $twitter->lookup_statuses( { id => $tweet_ids, include_entities => 'true', trim_user => 'false' } );
        };
        if ( !$tweets )
        {
            DEBUG( "twitter fetch error.  waiting 60 seconds before retry ..." );
            sleep( 60 );
        }
    }

    die( "unable to fetch tweets: $@ " ) unless ( $tweets );

    for my $tweet ( @{ $tweets } )
    {
        if ( my $ch_post = $ch_post_lookup->{ $tweet->{ id } } )
        {
            $ch_post->{ tweet } = $tweet;
        }
        else
        {
            LOGDIE( "no post found for tweet id $tweet->{ id }" );
        }
    }

    map { DEBUG( "no tweet fetched for url $_->{ url }" ); } ( grep { !$_->{ tweet } } @{ $ch_posts } );
}

# using the data in ch_post, store the tweet in topic_tweets and its urls in topic_tweet_urls
sub _store_tweet_and_urls($$$$)
{
    my ( $db, $topic, $topic_tweet_day, $ch_post ) = @_;

    my $created_at = $ch_post->{ tweet }->{ created_at };
    my $publish_date =
      $created_at
      ? MediaWords::Util::SQL::get_sql_date_from_epoch( Date::Parse::str2time( $created_at ) )<commit_msg>use backoff timing for twitter fetch errors<commit_after>package MediaWords::Job::FetchTopicTweets;

=head1 NAME

MediaWords::Job::FetchTopicTweets - fetch tweets relevant to a topic and store them in the topic_tweets table

=head1 DESCRIPTION

Use the Crimson Hexagon API to lookup tweets relevant to a topic, then fetch each of those tweets from twitter.

=cut

use strict;
use warnings;

use Moose;
with 'MediaWords::AbstractJob';

use Modern::Perl "2015";
use MediaWords::CommonLibs;

use List::MoreUtils;
use Net::Twitter;
use Date::Parse;

use MediaWords::DB;
use MediaWords::Util::DateTime;
use MediaWords::Util::JSON;
use MediaWords::Util::Web;

my $_api_host = undef;

=head1 METHODS

=cut

# fetch the list of tweets from the ch api.  return only 500 posts unless $fetch_10k_posts is true.
sub _fetch_ch_posts ($$)
{
    my ( $ch_monitor_id, $day ) = @_;

    my $ua = MediaWords::Util::Web::UserAgentDetermined();
    $ua->max_size( 100 * 1024 * 1024 );
    $ua->timeout( 90 );
    $ua->timing( '1,2,4,8,16,32,64,128,256,512' );

    my $key = MediaWords::Util::Config::get_config->{ crimson_hexagon }->{ key };
    LOGDIE( "no crimson hexagon key in mediawords.yml at //crimson_hexagon/key." ) unless ( $key );

    my $next_day = MediaWords::Util::SQL::increment_day( $day );

    my $url = _get_ch_api_url() . "?auth=$key&id=$ch_monitor_id&start=$day&end=$next_day&extendLimit=true";

    DEBUG( "CH URL: $url" );

    my $response = $ua->get( $url );

    if ( !$response->is_success )
    {
        LOGDIE( "error fetching posts: " . $response->as_string );
    }

    my $decoded_content = $response->decoded_content;

    my $data;
    eval { $data = MediaWords::Util::JSON::decode_json( $decoded_content ); };

    LOGDIE( "Unable to parse json from url $url: $@ " . substr( $decoded_content, 0, 1024 ) . " ..." ) unless ( $data );

    LOGDIE( "Unknown response status: '$data->{ status }'" ) unless ( $data->{ status } eq 'success' );

    return $data;
}

# add a row to topic_tweet_day if it does not already exist.  fetch data for new row from CH
sub _add_topic_tweet_single_day ($$$)
{
    my ( $db, $topic, $day ) = @_;

    my $topic_tweet_day = $db->query( <<SQL, $topic->{ topics_id }, $day )->hash;
select * from topic_tweet_days where topics_id = \$1 and day = \$2
SQL

    return if ( $topic_tweet_day && $topic_tweet_day->{ tweets_fetched } );

    # if we have a ttd but had not finished fetching tweets, delete it and start over
    $db->delete_by_id( 'topic_tweet_days', $topic_tweet_day->{ topic_tweet_days_id } ) if ( $topic_tweet_day );

    my $ch_posts = _fetch_ch_posts( $topic->{ ch_monitor_id }, $day );

    my $tweet_count = $ch_posts->{ totalPostsAvailable };

    my $num_ch_tweets = scalar( @{ $ch_posts->{ posts } } );

    $topic_tweet_day = $db->create(
        'topic_tweet_days',
        {
            topics_id      => $topic->{ topics_id },
            day            => $day,
            tweet_count    => $tweet_count,
            num_ch_tweets  => $num_ch_tweets,
            tweets_fetched => 'false'
        }
    );

    $topic_tweet_day->{ ch_posts } = $ch_posts;

    return $topic_tweet_day;
}

# for each day within the topic date range, find or create a topic_tweet_day row.
sub _add_topic_tweet_days ($$)
{
    my ( $db, $topic ) = @_;

    my $twitter = _get_twitter_handle();

    my $date = $topic->{ start_date };
    while ( $date le $topic->{ end_date } )
    {
        my $topic_tweet_day = _add_topic_tweet_single_day( $db, $topic, $date );
        _fetch_tweets_for_day( $db, $twitter, $topic, $topic_tweet_day ) if ( $topic_tweet_day );

        $date = MediaWords::Util::SQL::increment_day( $date );
    }

}

# given a set of ch_posts, fetch data from twitter about each tweet and attach it under the $ch->{ tweet } field
sub _add_tweets_to_ch_posts
{
    my ( $twitter, $ch_posts ) = @_;

    DEBUG( "fetching tweets for " . scalar( @{ $ch_posts } ) . " tweets" );

    LOGDIE( "more than 100 posts in $ch_posts" ) unless ( scalar( @{ $ch_posts } ) <= 500 );

    my $ch_post_lookup = {};
    for my $ch_post ( @{ $ch_posts } )
    {
        LOGDIE( "Unable to parse id from tweet url: $ch_post->{ url }" ) unless ( $ch_post->{ url } =~ m~/status/(\d+)~ );
        my $tweet_id = $1;

        $ch_post->{ tweet_id } = $tweet_id;
        $ch_post_lookup->{ $tweet_id } = $ch_post;
    }

    my $tweet_ids = [ keys( %{ $ch_post_lookup } ) ];

    my $tweets;
    my $twitter_retries = 0;
    while ( !$tweets && ( ++$twitter_retries <= 10 ) )
    {
        eval {
            $tweets = $twitter->lookup_statuses( { id => $tweet_ids, include_entities => 'true', trim_user => 'false' } );
        };
        if ( !$tweets )
        {
            my $sleep = 2 * ( $twitter_retries**2 );
            DEBUG( "twitter fetch error.  waiting $sleep seconds before retry ..." );
            sleep( $sleep );
        }
    }

    die( "unable to fetch tweets: $@ " ) unless ( $tweets );

    for my $tweet ( @{ $tweets } )
    {
        if ( my $ch_post = $ch_post_lookup->{ $tweet->{ id } } )
        {
            $ch_post->{ tweet } = $tweet;
        }
        else
        {
            LOGDIE( "no post found for tweet id $tweet->{ id }" );
        }
    }

    map { DEBUG( "no tweet fetched for url $_->{ url }" ); } ( grep { !$_->{ tweet } } @{ $ch_posts } );
}

# using the data in ch_post, store the tweet in topic_tweets and its urls in topic_tweet_urls
sub _store_tweet_and_urls($$$$)
{
    my ( $db, $topic, $topic_tweet_day, $ch_post ) = @_;

    my $created_at = $ch_post->{ tweet }->{ created_at };
    my $publish_date =
      $created_at
      ? MediaWords::Util::SQL::get_sql_date_from_epoch( Date::Parse::str2time( $created_at ) )
{'fraction_non_alphanumeric': 0.14997354963851173, 'fraction_numerical': 0.01137365543995768, 'mean_word_length': 2.888584161810079, 'pattern_counts': {'":': 0, '<': 10, '<?xml version=': 0, '>': 97, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '352386', 'n_tokens_mistral': 4298, 'n_tokens_neox': 4042, 'n_words': 1035}