s434708/pbr-ayct-backend
1
2
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

delete twitter account and campaign

Browse Source
This commit is contained in:
Łukasz Jędyk 2022-04-12 17:55:52 +02:00
parent 12a946c88a
commit 9d95c8a411
4 changed files with 204 additions and 99 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

196
ayct_backend/campaign/__init__.py
View File

@ -2,14 +2,13 @@ import requests
import json
from requests_oauthlib import OAuth1Session
from flask import Blueprint, current_app, request, jsonify
from rsa import verify
from ayct_backend.twitter.models import *
from ayct_backend.campaign.models import *
from ayct_backend.firebase import verify_token
campaign = Blueprint('campaign', __name__)
@campaign.route('/campaign', methods=['GET'])
@campaign.route('/', methods=['GET'])
def get_twitter_campaigns():
decoded_token = verify_token(request.headers)
if not decoded_token:
@ -23,19 +22,17 @@ def get_twitter_campaigns():
for campaign in twitter_campaigns:
campaign_data = {}
campaign_data['campaign_id'] = campaign.id
campaign_data['campaign_name'] = campaign.campaign_name
campaign_data['twitter_account_id'] = campaign.twitter_account_id
campaign_data['user_input'] = campaign.user_input
campaign_data['generated_content'] = campaign.generated_content
campaign_data['twitter_post_id'] = campaign.twitter_post_id
campaigns.append(campaign_data)
return jsonify({
"twitter_campaigns": campaigns
}), 200
@campaign.route('/campaign', methods=['POST'])
def add_twitter_account():
@campaign.route('/', methods=['POST'])
def add_twitter_campaign():
decoded_token = verify_token(request.headers)
if not decoded_token:
return "Not authorised!", 401
@ -43,61 +40,132 @@ def add_twitter_account():
user_id = decoded_token['sub']
content_type = request.headers.get('Content-Type')
if (content_type == 'application/json'):
request_json = request.json
if not request_json['campaign_name'] or not request_json['user_input'] or not request_json['twitter_account_id']:
return "Invalid request!", 400
else:
consumer_key = current_app.config["TWITTER_CONSUMER_KEY"]
consumer_secret = current_app.config["TWITTER_CONSUMER_SECERT"]
twitter_account = TwitterAccount.query.filter_by(twitter_account_id=request_json['twitter_account_id']).first()
# generate campaign content
core_url = 'http://65.108.80.28:4999/generate'
payload = {
"data": request_json['user_input'],
"length": 420
}
response = requests.post(core_url, headers={"Content-Type":"application/json"}, data=json.dumps(payload))
generated_content = response.content.decode("utf-8").replace('<|endoftext|>', '').replace(' ', ' ')
trimmed_text = generated_content[:270]
dot_index = trimmed_text.rfind(".")
space_index = trimmed_text.rfind(" ")
if dot_index > 0:
tweet_text = trimmed_text[:dot_index + 1].capitalize()
else:
tweet_text = trimmed_text[:space_index + 1].capitalize()
# create post on twitter
oauth = OAuth1Session(
consumer_key,
client_secret=consumer_secret,
resource_owner_key=twitter_account.access_token,
resource_owner_secret=twitter_account.access_token_secret,
)
response = oauth.post(
"https://api.twitter.com/2/tweets",
json={"text": tweet_text},
).json()['data']
# save campaign to database
new_twitter_campaign = TwitterCampaign(
campaign_name = request_json['campaign_name'],
user_id = user_id,
twitter_account_id = request_json['twitter_account_id'],
user_input = request_json['user_input'],
generated_content = tweet_text,
twitter_post_id = response['id']
)
campaign_db.session.add(new_twitter_campaign)
campaign_db.session.commit()
return "Campaign succesfully created.", 201
else:
if content_type != 'application/json':
return "Content-type not supported!", 400
request_json = request.json
if not request_json['campaign_name'] or not request_json['user_input'] or not request_json['twitter_account_id']:
return "Invalid request!", 400
consumer_key = current_app.config["TWITTER_CONSUMER_KEY"]
consumer_secret = current_app.config["TWITTER_CONSUMER_SECERT"]
twitter_account = TwitterAccount.query.filter_by(twitter_account_id=request_json['twitter_account_id']).first()
# generate campaign content
core_url = 'http://65.108.80.28:4999/generate'
payload = {
"data": request_json['user_input'],
"length": 420
}
response = requests.post(core_url, headers={"Content-Type":"application/json"}, data=json.dumps(payload))
generated_content = response.content.decode("utf-8").replace('<|endoftext|>', '').replace(' ', ' ')
trimmed_text = generated_content[:270]
dot_index = trimmed_text.rfind(".")
space_index = trimmed_text.rfind(" ")
if dot_index > 0:
tweet_text = trimmed_text[:dot_index + 1].capitalize()
else:
tweet_text = trimmed_text[:space_index + 1].capitalize()
# create post on twitter
oauth = OAuth1Session(
consumer_key,
client_secret=consumer_secret,
resource_owner_key=twitter_account.access_token,
resource_owner_secret=twitter_account.access_token_secret,
)
response = oauth.post(
"https://api.twitter.com/2/tweets",
json={"text": tweet_text},
).json()['data']
# save campaign to database
twitter_campaign = TwitterCampaign(
user_id = user_id,
campaign_name = request_json['campaign_name'],
twitter_account_id = request_json['twitter_account_id'],
user_input = request_json['user_input'],
)
campaign_db.session.add(twitter_campaign)
campaign_db.session.flush()
campaign_db.session.refresh()
twitter_campaign_post = TwitterPost(
campaign_id = twitter_campaign.id,
user_id = user_id,
post_content = tweet_text,
twitter_post_id = response['id']
)
campaign_db.session.add(twitter_campaign_post)
campaign_db.session.commit()
return "Campaign succesfully created.", 201
@campaign.route('/', methods=['DELETE'])
def delete_twitter_campaign():
decoded_token = verify_token(request.headers)
if not decoded_token:
return "Not authorised!", 401
user_id = decoded_token['sub']
content_type = request.headers.get('Content-Type')
if content_type != 'application/json':
return "Content-type not supported!", 400
request_json = request.json
if not request_json['campaign_id']:
return "Invalid request!", 400
twitter_campaign = TwitterAccount.query.filter_by(user_id=user_id, id=request_json['campaign_id'])
if not twitter_campaign:
return "Account not found!", 404
twitter_db.session.delete(twitter_campaign)
twitter_db.session.commit()
return "Twitter campaign succesfully deleted.", 200
@campaign.route('/details', methods=['GET'])
def get_twitter_campaign_details():
decoded_token = verify_token(request.headers)
if not decoded_token:
return "Not authorised!", 401
user_id = decoded_token['sub']
content_type = request.headers.get('Content-Type')
if content_type != 'application/json':
return "Content-type not supported!", 400
request_json = request.json
if not request_json['campaign_id']:
return "Invalid request!", 400
twitter_campaign = TwitterCampaign.query.filter_by(user_id=user_id, id=request_json['campaign_id']).first()
campaign_posts = TwitterPost.query.filter_by(user_id=user_id, campaign_id=request_json['campaign_id'])
posts = []
for post in campaign_posts:
post_data = {}
post_data['post_id'] = post.post_id
post_data['campaign_id'] = post.campaign_id
post_data['post_content'] = post.post_content
post_data['twitter_post_id'] = post.twitter_post_id
posts.append(post_data)
return jsonify({
"campaign_id": twitter_campaign.id,
"campaign_name": twitter_campaign.campaign_name,
"user_input": twitter_campaign.user_input,
"twitter_account_id": twitter_campaign.twitter_account_id,
"posts": posts
}), 200

11
ayct_backend/campaign/models.py
View File

@ -5,10 +5,17 @@ campaign_db = SQLAlchemy()
class TwitterCampaign(campaign_db.Model):
__tablename__ = 'twitter_campaign'
campaign_id = campaign_db.Column(campaign_db.Integer, primary_key=True)
id = campaign_db.Column(campaign_db.Integer, primary_key=True)
user_id = campaign_db.Column(campaign_db.String(64), nullable=False)
campaign_name = campaign_db.Column(campaign_db.String(64), nullable=False)
twitter_account_id = campaign_db.Column(campaign_db.String(32), nullable=False)
user_input = campaign_db.Column(campaign_db.String(100), nullable=False)
generated_content = campaign_db.Column(campaign_db.String(300), nullable=False)
class TwitterPost(campaign_db.Model):
__tablename__ = 'twitter_post'
id = campaign_db.Column(campaign_db.Integer, primary_key=True)
campaign_id = campaign_db.Column(campaign_db.Integer, nullable=False)
user_id = campaign_db.Column(campaign_db.String(64), nullable=False)
post_content = campaign_db.Column(campaign_db.String(300), nullable=False)
twitter_post_id = campaign_db.Column(campaign_db.String(32), unique=True, nullable=False)

93
ayct_backend/twitter/__init__.py
View File

@ -36,35 +36,66 @@ def add_twitter_account():
user_id = decoded_token['sub']
content_type = request.headers.get('Content-Type')
if (content_type == 'application/json'):
json = request.json
if not json['veryfier'] or not json['oauth_token'] or not json['oauth_token_secret']:
return "Invalid request!", 400
else:
consumer_key = current_app.config["TWITTER_CONSUMER_KEY"]
consumer_secret = current_app.config["TWITTER_CONSUMER_SECERT"]
access_token_url = "https://api.twitter.com/oauth/access_token"
oauth = OAuth1Session(
consumer_key,
client_secret=consumer_secret,
resource_owner_key=json['oauth_token'],
resource_owner_secret=json['oauth_token_secret'],
verifier=json['veryfier'],
)
oauth_tokens = oauth.fetch_access_token(access_token_url)
new_twitter_account = TwitterAccount(
user_id = user_id,
twitter_account_id = oauth_tokens['user_id'],
username = oauth_tokens['screen_name'],
access_token = oauth_tokens['oauth_token'],
access_token_secret = oauth_tokens['oauth_token_secret']
)
twitter_db.session.add(new_twitter_account)
twitter_db.session.commit()
return "Twitter account succesfully added.", 201
else:
if content_type != 'application/json':
return "Content-type not supported!", 400
json = request.json
if not json['veryfier'] or not json['oauth_token'] or not json['oauth_token_secret']:
return "Invalid request!", 400
consumer_key = current_app.config["TWITTER_CONSUMER_KEY"]
consumer_secret = current_app.config["TWITTER_CONSUMER_SECERT"]
access_token_url = "https://api.twitter.com/oauth/access_token"
oauth = OAuth1Session(
consumer_key,
client_secret=consumer_secret,
resource_owner_key=json['oauth_token'],
resource_owner_secret=json['oauth_token_secret'],
verifier=json['veryfier'],
)
oauth_tokens = oauth.fetch_access_token(access_token_url)
twitter_account = TwitterAccount.query.filter_by(user_id=user_id, twitter_account_id=oauth_tokens['user_id'])
if twitter_account:
return "Account already exists!", 400
twitter_account = TwitterAccount(
user_id = user_id,
twitter_account_id = oauth_tokens['user_id'],
username = oauth_tokens['screen_name'],
access_token = oauth_tokens['oauth_token'],
access_token_secret = oauth_tokens['oauth_token_secret']
)
twitter_db.session.add(twitter_account)
twitter_db.session.commit()
return "Twitter account succesfully added.", 201
@twitter.route('/account', methods=['DELETE'])
def delete_twitter_account():
decoded_token = verify_token(request.headers)
if not decoded_token:
return "Not authorised!", 401
user_id = decoded_token['sub']
content_type = request.headers.get('Content-Type')
if content_type != 'application/json':
return "Content-type not supported!", 400
request_json = request.json
if not request_json['twitter_account_id']:
return "Invalid request!", 400
twitter_account = TwitterAccount.query.filter_by(user_id=user_id, twitter_account_id=request_json['twitter_account_id'])
if not twitter_account:
return "Account not found!", 404
twitter_db.session.delete(twitter_account)
twitter_db.session.commit()
return "Twitter account succesfully deleted.", 200

3
ayct_backend/twitter/models.py
View File

@ -1,4 +1,3 @@
from enum import unique
from flask_sqlalchemy import SQLAlchemy
twitter_db = SQLAlchemy()
@ -6,7 +5,7 @@ twitter_db = SQLAlchemy()
class TwitterAccount(twitter_db.Model):
__tablename__ = 'twitter_account'
account_id = twitter_db.Column(twitter_db.Integer, primary_key=True)
id = twitter_db.Column(twitter_db.Integer, primary_key=True)
user_id = twitter_db.Column(twitter_db.String(64), nullable=False)
twitter_account_id = twitter_db.Column(twitter_db.String(32), nullable=False)
username = twitter_db.Column(twitter_db.String(16), nullable=False)