Skip to content

Instantly share code, notes, and snippets.

@nnsnodnb
Last active February 18, 2016 10:49
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
Star You must be signed in to star a gist
Save nnsnodnb/bfc3e4846352435f7baa to your computer and use it in GitHub Desktop.
ToDoアプリにツイートからタスクを追加
#!/usr/bin/env python3
# coding: utf-8
from tweepy.streaming import StreamListener, Stream
from tweepy.auth import OAuthHandler
from tweepy.api import API
from datetime import timedelta
import os
import mysql.connector
def get_oauth():
consumer_key = "********************"
consumer_secret = "********************"
access_key = "********************"
access_secret = "********************"
auth = OAuthHandler(consumer_key, consumer_secret)
auth.set_access_token(access_key, access_secret)
return auth
def insert_data(text):
status_text = text.split(" ")
connect = mysql.connector.connect(db="todo", host="127.0.0.1", port=3306, user="user", passwd="pass")
cur = connect.cursor()
stmt = ("INSERT INTO task (TITLE , SCHEDULE_DATE , RANK , FINISH_CHECK)" "VALUES(%s , %s , %s , %s)")
if "_" in status_text[0]:
tmp = status_text[0].replace("_" , " ")
status_text[0] = tmp
data = (status_text[0] , status_text[1] , status_text[2] , '0')
cur.execute(stmt, data)
connect.commit()
cur.close()
connect.close()
class AbstractedlyListener(StreamListener):
""" Let's stare abstractedly at the User Streams ! """
def on_status(self, status):
status.created_at += timedelta(hours=9)
if status.author.screen_name == 'nnsnodnb' and "#ひやかし帳" in status.text:
insert_data(status.text)
if __name__ == '__main__':
auth = get_oauth()
stream = Stream(auth, AbstractedlyListener(), secure=True)
stream.userstream()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment