def get_tweets(request):
datas = []
p = ttp.Parser()
try:
api = twitter.Api(
consumer_key=settings.TWITTER_CONSUMER_KEY,
consumer_secret=settings.TWITTER_CONSUMER_SECRET,
access_token_key=settings.TWITTER_ACCESS_TOKEN,
access_token_secret=settings.TWITTER_ACCESS_TOKEN_SECRET
)
tweets = api.GetUserTimeline(screen_name='kodlaco')
for tweet in tweets:
datas.append({
#'text': p.parse(tweet.text).html,
'text': tweet.text,
'id_str': tweet.id_str
})
except:
datas = []
return HttpResponse(json.dumps(datas), content_type="application/json")
python类TWITTER_CONSUMER_KEY的实例源码
def handle(self, *args, **options):
api = twitter.Api(consumer_key=settings.TWITTER_CONSUMER_KEY,
consumer_secret=settings.TWITTER_CONSUMER_SECRET,
access_token_key=settings.TWITTER_ACCESS_TOKEN_KEY,
access_token_secret=settings.TWITTER_ACCESS_TOKEN_SECRET)
for currency_symbol in settings.SOCIAL_NETWORK_SENTIMENT_CONFIG['twitter']:
print(currency_symbol)
results = api.GetSearch("$" + currency_symbol, count=200)
for tweet in results:
if SocialNetworkMention.objects.filter(network_name='twitter', network_id=tweet.id).count() == 0:
snm = SocialNetworkMention.objects.create(
network_name='twitter',
network_id=tweet.id,
network_username=tweet.user.screen_name,
network_created_on=datetime.datetime.fromtimestamp(tweet.GetCreatedAtInSeconds()),
text=tweet.text,
symbol=currency_symbol,
)
snm.set_sentiment()
snm.save()
def handle(self, *args, **options):
api = twitter.Api(consumer_key=settings.TWITTER_CONSUMER_KEY,
consumer_secret=settings.TWITTER_CONSUMER_SECRET,
access_token_key=settings.TWITTER_ACCESS_TOKEN_KEY,
access_token_secret=settings.TWITTER_ACCESS_TOKEN_SECRET)
for currency_symbol in settings.SOCIAL_NETWORK_SENTIMENT_CONFIG['twitter']:
print(currency_symbol)
results = api.GetSearch("$" + currency_symbol, count=200)
for tweet in results:
if SocialNetworkMention.objects.filter(network_name='twitter', network_id=tweet.id).count() == 0:
snm = SocialNetworkMention.objects.create(
network_name='twitter',
network_id=tweet.id,
network_username=tweet.user.screen_name,
network_created_on=datetime.datetime.fromtimestamp(tweet.GetCreatedAtInSeconds()),
text=tweet.text,
symbol=currency_symbol,
)
snm.set_sentiment()
snm.save()
def clean(self):
# remove leading @
if self.name.startswith('@'):
self.name = self.name[1:]
twitter = TwitterAPI(settings.TWITTER_CONSUMER_KEY,
settings.TWITTER_CONSUMER_SECRET,
auth_type='oAuth2')
try:
user = twitter.request('users/show', {'screen_name': self.name}).json()
except TwitterConnectionError:
logger.warning("Could not connect to Twitter.")
raise ValidationError("Could not connect to Twitter to retrieve user_id.")
if 'id' in user:
self.account_id = user['id']
else:
logger.warning("Could not find user with provided name.")
raise ValidationError("Could not find user with provided name.")
if TwitterAccount.objects.filter(account_id=self.account_id):
logger.warning("TwitterAccount with account_id is already in database.")
raise ValidationError("Twitter account with this name already exists.")
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.credentials = (
settings.TWITTER_CONSUMER_KEY,
settings.TWITTER_CONSUMER_SECRET,
settings.TWITTER_ACCESS_TOKEN,
settings.TWITTER_ACCESS_SECRET
)
self.log = logging.getLogger('django')
