import datetime
import csv
import json
-import requests_oauthlib
from postgresqleu.util.middleware import RedirectException
from postgresqleu.util.db import exec_to_list, exec_to_dict, exec_no_result
from postgresqleu.util.lists import flatten_list
from postgresqleu.util.decorators import superuser_required
+from postgresqleu.util.messaging.twitter import TwitterSetup
from models import Conference, ConferenceSeries
from models import AccessToken
if request.POST.get('activate_twitter', '') == '1':
# Fetch the oauth codes and re-render the form
try:
- oauth = requests_oauthlib.OAuth1Session(settings.TWITTER_CLIENT, settings.TWITTER_CLIENTSECRET)
- fetch_response = oauth.fetch_request_token('https://api.twitter.com/oauth/request_token')
- auth_url = oauth.authorization_url('https://api.twitter.com/oauth/authorize')
+ (auth_url, ownerkey, ownersecret) = TwitterSetup.get_authorization_data()
+ request.session['ownerkey'] = ownerkey
+ request.session['ownersecret'] = ownersecret
except Exception, e:
messages.error(request, 'Failed to talk to twitter: %s' % e)
return HttpResponseRedirect('.')
- request.session['ownerkey'] = fetch_response.get('oauth_token')
- request.session['ownersecret'] = fetch_response.get('oauth_stoken_secret')
return render(request, 'confreg/admin_integ_twitter.html', {
'conference': conference,
messages.error(request, 'Missing data in session, cannot continue')
return HttpResponseRedirect('.')
try:
- oauth = requests_oauthlib.OAuth1Session(settings.TWITTER_CLIENT,
- settings.TWITTER_CLIENTSECRET,
- resource_owner_key=request.session.pop('ownerkey'),
- resource_owner_secret=request.session.pop('ownersecret'),
- verifier=request.POST.get('pincode'))
- tokens = oauth.fetch_access_token('https://api.twitter.com/oauth/access_token')
+ tokens = TwitterSetup.authorize(request.session.pop('ownerkey'),
+ request.session.pop('ownersecret'),
+ request.POST.get('pincode'),
+ )
except:
messages.error(request, 'Failed to get tokens from twitter.')
return HttpResponseRedirect('.')
from postgresqleu.newsevents.models import News
from postgresqleu.confreg.models import Conference, ConferenceNews, ConferenceTweetQueue
-import requests_oauthlib
-
-
-def make_twitter_post(tw, statusstr):
- r = tw.post('https://api.twitter.com/1.1/statuses/update.json', data={
- 'status': statusstr,
- })
- if r.status_code != 200:
- print("Failed to post to twitter: %s " % r)
- return False
- return True
+from postgresqleu.util.messaging.twitter import Twitter
class Command(BaseCommand):
help = 'Post to twitter'
articles = []
if articles:
- tw = requests_oauthlib.OAuth1Session(settings.TWITTER_CLIENT,
- settings.TWITTER_CLIENTSECRET,
- settings.TWITTER_NEWS_TOKEN,
- settings.TWITTER_NEWS_TOKENSECRET)
+ tw = Twitter()
for a in articles:
# We hardcode 30 chars for the URL shortener. And then 10 to cover the intro and spacing.
settings.SITEBASE,
slugify(a.title),
a.id)
- if make_twitter_post(tw, statusstr):
+ ok, msg = tw.post_tweet(statusstr)
+ if ok:
a.tweeted = True
a.save()
+ else:
+ print("Failed to post to twitter: %s" % msg)
# Don't post more often than once / 10 seconds, to not trigger flooding.
time.sleep(10)
twitter_timewindow_end__gt=n).extra(where=[
"EXISTS (SELECT 1 FROM confreg_conferencenews n WHERE n.conference_id=confreg_conference.id AND (NOT tweeted) AND datetime > now()-'7 days'::interval AND datetime < now()) OR EXISTS (SELECT 1 FROM confreg_conferencetweetqueue q WHERE q.conference_id=confreg_conference.id)"
]):
-
- tw = requests_oauthlib.OAuth1Session(settings.TWITTER_CLIENT,
- settings.TWITTER_CLIENTSECRET,
- c.twitter_token,
- c.twitter_secret)
+ tw = Twitter(c)
al = list(ConferenceNews.objects.filter(conference=c, tweeted=False, datetime__gt=datetime.now()-timedelta(days=7), datetime__lt=datetime.now(), conference__twittersync_active=True).order_by('datetime')[:1])
if al:
statusstr = u"{0} {1}##{2}".format(a.title[:250-40],
c.confurl,
a.id)
- if make_twitter_post(tw, statusstr):
+ ok, msg = tw.post_tweet(statusstr)
+ if ok:
a.tweeted = True
a.save()
continue
+ else:
+ print("Failed to post to twitter: %s" % msg)
tl = list(ConferenceTweetQueue.objects.filter(conference=c).order_by('datetime')[:1])
if tl:
t = tl[0]
- if make_twitter_post(tw, t.contents):
+ ok, msg = tw.post_tweet(statusstr)
+ if ok:
t.delete()
continue
+ else:
+ print("Failed to post to twitter: %s" % msg)
--- /dev/null
+from django.conf import settings
+
+import requests_oauthlib
+
+_cached_twitter_users = {}
+
+class Twitter(object):
+ def __init__(self, conference=None):
+ if conference:
+ token = conference.twitter_token
+ secret = conference.twitter_secret
+ else:
+ token = settings.TWITTER_NEWS_TOKEN
+ secret = settings.TWITTER_NEWS_TOKENSECRET
+
+ self.tw = requests_oauthlib.OAuth1Session(settings.TWITTER_CLIENT,
+ settings.TWITTER_CLIENTSECRET,
+ token,
+ secret)
+
+
+ def post_tweet(self, tweet):
+ r = self.tw.post('https://api.twitter.com/1.1/statuses/update.json', data={
+ 'status': tweet,
+ })
+ if r.status_code != 200:
+ return (False, r.text)
+ return (True, None)
+
+
+class TwitterSetup(object):
+ @classmethod
+ def get_authorization_data(self):
+ oauth = requests_oauthlib.OAuth1Session(settings.TWITTER_CLIENT, settings.TWITTER_CLIENTSECRET)
+ fetch_response = oauth.fetch_request_token('https://api.twitter.com/oauth/request_token')
+ auth_url = oauth.authorization_url('https://api.twitter.com/oauth/authorize')
+
+ return (auth_url,
+ fetch_response.get('oauth_token'),
+ fetch_response.get('oauth_token_secret'),
+ )
+
+ @classmethod
+ def authorize(self, ownerkey, ownersecret, pincode):
+ oauth = requests_oauthlib.OAuth1Session(settings.TWITTER_CLIENT,
+ settings.TWITTER_CLIENTSECRET,
+ resource_owner_key=ownerkey,
+ resource_owner_secret=ownersecret,
+ verifier=pincode)
+ tokens = oauth.fetch_access_token('https://api.twitter.com/oauth/access_token')
+
+ return tokens
+
projects / pgeu-website.git / commitdiff