Работа с API Twitter: использование стримов в реальном времени

() translation by (you can also view the original English article)

Final product image — What You'll Be Creating

Хотя REST API Twitter подходит для многих приложений, если вам нужны мгновенные обновления и доступ к более широкому набору уведомлений, то необходим API Twitter Streaming. Например, только потоковый API скажет вам, когда другой пользователь будет использовать один из ваших твитов.

Использование Streaming API требует постоянного, постоянного подключения между вашим веб-сервером и Twitter. Этот тип реализации может быть незнакомым многим разработчикам PHP. Как только появятся твиты, Twitter уведомляет ваш сервер в режиме реального времени, позволяя вам хранить их в своей базе данных без задержки и опроса REST API. Использование Streaming API также не зависит от ограничений по скорости API в Twitter.

Вот как это работает:

Существует три варианта API-интерфейсов Twitter Streaming:

Публичный стрим. Он позволяет вашему приложению отслеживать публичные данные в Twitter, например, публичные твиты, фильтры хештэга и т.д.
Пользовательский стрим. Он позволяет отслеживать поток твитов пользователя в режиме реального времени. Третья часть этой серии будет посвящена пользовательскому потоку.
Потоки сайтов. Потоки сайтов позволяют вашему приложению отслеживать каналы Twitter в режиме реального времени для большого числа пользователей.

Задача вашей потоковой реализации заключается в том, чтобы как можно быстрее регистрировать входящие события и обрабатывать их в фоновом режиме с использованием REST API по мере необходимости для сбора более детальных данных. Потоки сайтов требуют предварительного одобрения от Twitter, которые, скорее всего, зарезервированы для крупных компаний и разработчиков.

К счастью, есть бесплатная библиотека с открытым исходным кодом Phirehose, которая реализует большинство требований к потоковым API. В этом руководстве будет описано, как интегрировать Phirehose в наше приложение с открытым исходным кодом Birdcage.

Библиотека Phirehose

Phirehose - это просто фантастическая PHP реализация для требований Stream Stream API, написанная Fenn Bailey. Как он сам описывает ее, Phirehose:

обеспечивают простой интерфейс Streaming API для PHP-приложений
выполняет рекомендации по потоковым API для обработки ошибок, повторного подключения и т. д.
направляет клиентов к правильному управлению потоковыми API
работает независимо от расширений PHP (то есть разделяемой памяти, PCNTL и т. д.),

Я нашел, что библиотека работает совершенно безупречно. Здесь вы найдёте больше документации по Phirehose.

Он предназначен для поддержания связи с Twitter и ответа на фид данных Twitter при неограниченном использовании и без перерывов. Он не предназначен для обработки детальный обработки твитов и гидратации данных, о которых мы говорили во второй части этой серии. Это можно сделать отдельно.

Запуск Phirehose

Как правило, вы не можете запускать типичную задачу cron в Интернете как неопределенную операцию keep-alive. Лучше создать демона командной строки.

Одной из мощных возможностей Yii является возможность запуска консольных приложений из командной строки. Это позволит нам запустить приложение командной строки keep-alive

Создание консольной команды Yii

В каталоге /app/, за пределами веб-доступного корня, мы добавим файл stream.php, который запускает нашу консольную консольную команду Phirehose:

<?php
defined('YII_DEBUG') or define('YII_DEBUG',true);
$yii=dirname(__FILE__).'/../framework/yii.php';
$config=dirname(__FILE__).'/protected/config/main.php';
require_once($yii);
Yii::createConsoleApplication($config)->run();

Затем мы создадим сам командный файл StreamCommand.php в каталоге /app/protected/commands:

<?php
class StreamCommand extends CConsoleCommand
{    
    // test with php ./app/stream.php Stream
    public function run($args)
    {
        // get Twitter user account keys
        $result = Account::model()->findByPk(1);
        $c = new Consumer($result['oauth_token'],$result['oauth_token_secret'],Phirehose::METHOD_USER);
        // load Twitter App keys
        $app = UserSetting::model()->loadPrimarySettings();
        $c->consumerKey = $app['twitter_key'];
        $c->consumerSecret = $app['twitter_secret'];
        $c->consume();
    }
}

Он запустит процесс Phirehose, Consumer, используя наше приложение Twitter и пользовательские ключи.

Примечание. Для примера потоковой записи Birdcage мы предполагаем, что зарегистрирована только одна учетная запись Twitter и жестко прописываем свои учетные данные, например. account_id = 1.

Интеграция с Phirehose

Чтобы интегрировать Phirehose в Birdcage, я переместил OAuthPhirehose.php и UserstreamPhirehose.php в каталог /app /protected/components. В моем файле конфигурации main.php я добавил phirehose в список загруженных компонентов:

    'preload'=>array(
    'log',
	  'bootstrap',
	  'mailgun',
	  'phirehose',
  	  'advanced'
	  ),

Затем я создал миграцию базы данных, чтобы создать таблицу для хранения необработанных данных из потока Twitter:

class m140919_193106_create_stream_table extends CDbMigration
{
     protected $MySqlOptions = 'ENGINE=InnoDB CHARSET=utf8 COLLATE=utf8_unicode_ci';
     public $tablePrefix;
     public $tableName;

     public function before() {
       $this->tablePrefix = Yii::app()->getDb()->tablePrefix;
       if ($this->tablePrefix <> '')
         $this->tableName = $this->tablePrefix.'stream';
     }

       public function safeUp()
   	{
   	  $this->before();
      $this->createTable($this->tableName, array(
               'id' => 'pk',
               'tweet_id' => 'bigint(20) unsigned NOT NULL',
               'code' => 'text  NULL',
               'is_processed' => 'tinyint default 0',
               'created_at' => 'DATETIME NOT NULL DEFAULT 0',
               'modified_at' => 'TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP',
                 ), $this->MySqlOptions);
                 $this->createIndex('tweet_id', $this->tableName , 'tweet_id', true);               

   	}

Я также создал новую модель под названием Consumer.php, которая расширяет OauthPhirehose с помощью необходимого метода enqueueStatus.

Мы хотим минимизировать объем обработки, которую необходимо выполнить в режиме реального времени. По сути, мы просто хотим записать данные, полученные из Twitter в нашу базу данных, и ничего больше. Мы можем выполнять остальную обработку в наших собственных фоновых задачах, не замедляя потоковое соединение Phirehose. Моя функция просто берет данные твитов из входящего потока и сохраняет их в таблице потоков:

<?php
  class Consumer extends OauthPhirehose
  {
    // This function is called automatically by the Phirehose class
    // when a new tweet is received with the JSON data in $status
    public function enqueueStatus($status) {
      $stream_item = json_decode($status);
      if (!(isset($stream_item->id_str))) { return;}
      $s = new Stream;
      $s->tweet_id = $stream_item->id_str;
      $s->code = base64_encode(serialize($stream_item));
      $s->is_processed=0;
      $s->created_at = new CDbExpression('NOW()');          
      $s->modified_at =new CDbExpression('NOW()');          
      $s->save();
      var_dump($stream_item);
    }
  }
  ?>

Мы будем полагаться на фоновые задачи, выполняемые нашим DaemonController для обработки данных в модели Bircdcage Tweet. Это описано ниже.

Активирование Phirehose

Вы можете протестировать Phirehose с помощью консольной PHP команды:

php ./app/stream.php Stream

Twitter отправит поток информации о последователях для учетной записи пользователя, а затем данные в реальном времени по мере поступления.

Чтобы активировать Phirehose как keep-alive, всегда включенную консольную команду, мы будем использовать команду nohup, например no hangup и перенаправляя вывод на dev/null:

nohup php ./app/stream.php Stream> / dev / null 2> & 1 &

Ubuntu ответит идентификатором задания вашего процесса для будущего его мониторинга и завершения:

[1] 9768

Если вы хотите проверить, что процесс запущен, просканируйте список задач для идентификатора задания:

ps -e все | grep 9768

Вы должны увидеть что-то вроде этого:

0 1000 9768 9743 20 0 273112 16916 poll_s S pts / 0 0:00 php ./app/stream.php Stream

И вы можете завершить Phirehose, убив id задания:

kill 9768

По моему собственному опыту, Phirehose безупречно работал с этой техникой.

Обработка потоковых данных

Нам также необходимо создать фоновый процесс в Birdcage, который будет обрабатывать потоковые данные в наших таблицах Twitter, Mention, URL и Hashtag, как если бы они появились в REST API.

Измените настройки файла twitter.ini для использования потоков:

1	twitter_stream = true

И мы можем использовать одно и то же задание cron из второй части для выполнения этой операции:

# To define the time you can provide concrete values for
# minute (m), hour (h), day of month (dom), month (mon),
# and day of week (dow) or use '*' in these fields (for 'any').# 
# Notice that tasks will be started based on the cron's system
# daemon's notion of time and timezones.
# 
# For example, you can run a backup of all your user accounts
# at 5 a.m every week with:
# 0 5 * * 1 tar -zcf /var/backups/home.tgz /home/
# 
# m h  dom mon dow   command
*/5 * * * * wget -O /dev/null http://birdcage.yourdomain.com/daemon/index

Затем, когда вызывается DaemonController, он активирует метод Stream model():

    public function actionIndex() {
	  // if not using twitter streams, we'll process tweets by REST API
	  if (!Yii::app()->params['twitter_stream']) {
	    Tweet::model()->getStreams();	    
	  } else {
	    Stream::model()->process();
	  }
  }

Метод процесса распаковывает данные закодированного потока и анализирует каждую запись так же, как мы сделали с ответом из REST API:

    public function process() {
	  // get unprocessed tweets from stream engine
	  // to do 
	  $account_id = 1;
	  $items = Stream::model()->unprocessed()->findAll();
	  foreach ($items as $i) {
	    $tweet = unserialize(base64_decode($i['code']));
      Tweet::model()->parse($account_id,$tweet);
      $this->setStatus($i['id'],self::STREAM_PROCESSED);      
	  }
	}

Birdcage в настоящее время игнорирует данные из потока, который не является твитом, например. уведомления, прямые сообщения и т. д. Я оставлю это вам для расширения - или вы можете проверить мое расширенное приложение Birdhouse.

В заключение

Надеюсь, вы нашли эту серию из трех статей о Twitter API полезной и интересной. К настоящему времени вы узнали об OAuth, REST API, Streaming API, создании базы данных для Twitter, обработке временной шкалы с использованием обоих типов API-интерфейсов, правильном подсчете символов в твитах и т.д.

Пожалуйста, размещайте любые комментарии, исправления или дополнительные идеи ниже. Вы можете найти другие мои уроки Tuts + на моей странице автора или подписаться на меня в Twitter @reifman.