pbr-ayct-backend/ayct_backend/campaign/__init__.py
2022-05-17 22:25:20 +02:00

165 lines
5.4 KiB
Python

import requests
import json
from uuid import uuid4
from requests_oauthlib import OAuth1Session
from flask import Blueprint, current_app, request, jsonify
from ayct_backend.models import *
from ayct_backend.firebase import verify_token
campaign = Blueprint('campaign', __name__)
@campaign.route('/', methods=['GET'])
def get_twitter_campaigns():
decoded_token = verify_token(request.headers)
if not decoded_token:
return "Not authorised!", 401
user_id = decoded_token['sub']
twitter_campaigns = TwitterCampaign.query.filter_by(user_id=user_id).all()
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
campaigns.append(campaign_data)
return jsonify({
"twitter_campaigns": campaigns
}), 200
@campaign.route('/', methods=['POST'])
def add_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 'campaign_name' not in request_json or 'user_input' not in request_json or 'twitter_account_id' not in request_json:
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()
random_part = str(uuid4().hex)
tweet_text = "This is placeholder content generated by our application. Normal generation doesn't work." + random_part
# 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(
id = str(uuid4()),
user_id = user_id,
campaign_name = request_json['campaign_name'],
twitter_account_id = request_json['twitter_account_id'],
user_input = request_json['user_input'],
)
twitter_campaign_post = TwitterPost(
id = str(uuid4()),
campaign_id = twitter_campaign.id,
user_id = user_id,
post_content = tweet_text,
twitter_post_id = response['id']
)
db.session.add(twitter_campaign)
db.session.add(twitter_campaign_post)
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 'campaign_id' not in request_json:
return "Invalid request!", 400
twitter_campaign = TwitterCampaign.query.filter_by(user_id=user_id, id=request_json['campaign_id']).first()
if not twitter_campaign:
return "Capmaign not found!", 404
db.session.delete(twitter_campaign)
db.session.commit()
return "Twitter campaign succesfully deleted.", 200
@campaign.route('/<campaign_id>', methods=['GET'])
def get_twitter_campaign_details(campaign_id):
decoded_token = verify_token(request.headers)
if not decoded_token:
return "Not authorised!", 401
user_id = decoded_token['sub']
twitter_campaign = TwitterCampaign.query.filter_by(user_id=user_id, id=campaign_id).first()
campaign_posts = TwitterPost.query.filter_by(user_id=user_id, campaign_id=campaign_id).all()
posts = []
for post in campaign_posts:
post_data = {}
post_data['post_id'] = 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