gistfile1.php

<?php
date_default_timezone_set('Asia/Tokyo');
$twitter_method = 'sample';
$twitter_user = 'user';
$twitter_pass = 'pass';

$db_name = 'twitter';
$db_user = 'user';
$db_pass = 'pass';

poll();

/**
 * 日本語のみを保存したいとかなら
 */
function filter($data){
    return $data;
}

class TwitterStream
{
    private $dbc;
    private $stmt_insert_status;
    private $stmt_insert_user;
    private $stmt_update_user;
    private $stmt_select_user;
    private $status_columns = array('id', 'user_id', 'text', 'source', 'truncated', 'geo', 'favorited',
        'in_reply_to_user_id', 'in_reply_to_status_id', 'in_reply_to_screen_name', 'created_at');
    private $user_columns = array('id', 'verified', 'profile_sidebar_fill_color', 'profile_text_color',
        'followers_count', 'protected', 'location', 'profile_background_color', 'utc_offset',
        'statuses_count', 'description', 'friends_count', 'profile_link_color', 'profile_image_url',
        'notifications', 'geo_enabled', 'profile_background_image_url', 'screen_name', 'profile_background_tile',
        'favorites_count', 'name', 'url', 'created_at', 'time_zone', 'profile_sidebar_border_color', 'following');

    public function save($status){
        $this->dbc->beginTransaction();
        $this->save_user($status['user']);
        foreach($this->status_columns as $c){
            if($c == 'user_id' || $c == 'user') continue;
            $value = isset($status[$c])? $status[$c]: null;
            if($c == 'created_at' && !empty($value)) $value = $this->format_date($value);
            if(is_array($value)) $value = json_encode($value);
            $this->stmt_insert_status->bindValue(':'. $c, $value);
        }
        $this->stmt_insert_status->bindValue(':user_id', $status['user']['id']);
        if(!$this->stmt_insert_status->execute()) var_dump($this->stmt_insert_status->errorInfo());
        $this->dbc->commit();
    }
    private function save_user($user){
        $update = false;
        if($this->stmt_select_user->execute(array($user['id']))){
            while($row = $this->stmt_select_user->fetch()){
                $update = true;
                break;
            }
        }
        foreach($this->user_columns as $c){
            $value = isset($user[$c])? $user[$c]: null;
            if($c == 'created_at' && !empty($value)) $value = $this->format_date($value);
            if($update){
                $this->stmt_update_user->bindValue(':'. $c, $value);
            } else {
                $this->stmt_insert_user->bindValue(':'. $c, $value);
            }
        }
        if($update){
            $this->stmt_update_user->execute();
        } else {
            $this->stmt_insert_user->execute();
        }
    }

    public function __construct(PDO $dbc){
        $this->stmt_insert_status = $dbc->prepare('INSERT INTO statuses ('. implode(', ', $this->status_columns). ') VALUES ('. $this->named_columns('status'). ')');
        $this->stmt_insert_user = $dbc->prepare('INSERT INTO user ('. implode(', ', $this->user_columns). ') VALUES ('. $this->named_columns('user'). ')');
        $sql = 'UPDATE user SET ';
        $columns = array();
        foreach($this->user_columns as $c) if($c != 'id') $columns[] = sprintf('%s = :%s', $c, $c);
        $sql .= implode(', ', $columns);
        $sql .= ' WHERE id = :id';
        $this->stmt_update_user = $dbc->prepare($sql);
        $this->stmt_select_user = $dbc->prepare('SELECT * FROM user WHERE id = ?');
        $this->dbc = $dbc;
    }
    public function __destruct(){
        $this->dbc = null;
    }
    private function format_date($date){
        return date('Y-m-d H:i:s', strtotime($date));
    }
    private function named_columns($n){
        $name = $n. '_columns';
        $columns = $this->{$name};
        foreach($columns as &$c){
            $c = ':'. $c;
        }
        return implode(', ', $columns);
    }
}

/**
 *
 */
function poll(){
    global $twitter_method, $twitter_user, $twitter_pass, $db_name, $db_user, $db_pass;

    try {
        $pdo = new PDO('mysql:host=localhost;dbname='. $db_name, $db_user, $db_pass);
        $dbc = new TwitterStream($pdo);
    } catch(Exception $e){
        printf("DB connection failed: %s\n", $e->getMessage());
        exit;
    }
    $stream = fopen(sprintf('http://%s:%[email protected]/1/statuses/%s.json',
        $twitter_user, $twitter_pass, $twitter_method), 'r');
    if(false === $stream){
        echo 'twitter connection failed', PHP_EOL;
        exit;
    }
    while($json = @fgets($stream)){
        $status = json_decode($json, true);
        if(isset($status['text'])){
            $status = filter($status);
            if(is_null($status)){
                continue;
            }
            try {
                $dbc->save($status);
            } catch(Exception $e){
                $pdo->rollBack();
                printf("status save failed: %s\n", $e->getMessage());
            }
        }
    }
    $pdo = $dbc = null;
    echo 'reconnecting...', PHP_EOL;
    poll();
}

gistfile2.sql

CREATE TABLE 'statuses'(
    'id' bigint UNSIGNED primary key,
    'user_id' int(14),
    'text' text,
    'source' varchar(255),
    'truncated' int(1),
    'geo' varchar(60),
    'favorited' int(1),
    'in_reply_to_user_id' int(14) UNSIGNED,
    'in_reply_to_status_id' bigint UNSIGNED,
    'in_reply_to_screen_name' varchar(100),
    'created_at' timestamp
);

CREATE TABLE 'user'(
    'id' int(14) UNSIGNED primary key,
    'verified' int(1),
    'profile_sidebar_fill_color' varchar(10),
    'profile_text_color' varchar(10),
    'followers_count' int(11) UNSIGNED,
    'protected' int(1),
    'location' varchar(255),
    'profile_background_color' varchar(10),
    'utc_offset' int(6),
    'statuses_count' int(11) UNSIGNED,
    'description' varchar(255),
    'friends_count' int(11) UNSIGNED,
    'profile_link_color' varchar(10),
    'profile_image_url' varchar(255),
    'notifications' varchar(255),
    'geo_enabled' int(1),
    'profile_background_image_url' varchar(255),
    'screen_name' varchar(100),
    'profile_background_tile' int(1),
    'favorites_count' int(11) UNSIGNED,
    'name' varchar(100),
    'url' varchar(255),
    'created_at' timestamp,
    'time_zone' varchar(20),
    'profile_sidebar_border_color' varchar(10),
    'following' int(1)
);