diff --git a/README.md b/README.md index a4e7351..4564b7e 100644 --- a/README.md +++ b/README.md @@ -78,7 +78,13 @@ The bot now includes a powerful CLI for easy management! See the [CLI Usage](#-c access_token_secret = "GET_KEY_FROM_developer.twitter.com/apps" openai_key = "GET_YOUR_OPENAI_API_KEY_FROM_https://platform.openai.com/api-keys" ``` -3. Customize `data/tweets.txt` with your tweets. (SKIP: If not using tweet from file) +3. Optional: set these environment variables to post tweets through Xquik instead of the default Tweepy path: + ```sh + TWITTER_BACKEND=xquik + XQUIK_API_KEY=your_xquik_api_key + XQUIK_ACCOUNT=your_x_handle_or_account_id + ``` +4. Customize `data/tweets.txt` with your tweets. (SKIP: If not using tweet from file) ## 🔧 Usage diff --git a/cli.py b/cli.py index d0e0709..c32d2ed 100755 --- a/cli.py +++ b/cli.py @@ -15,7 +15,7 @@ sys.path.append(os.path.join(os.path.dirname(__file__), 'src')) import keys -from src.functions import generate_response, initialize_tweepy, get_formatted_date +from src.functions import generate_response, initialize_posting_client, initialize_tweepy, post_tweet @click.group() @@ -38,7 +38,7 @@ def post(ai, from_file, text, prompt): sys.exit(1) try: - client, _ = initialize_tweepy() + client, _ = initialize_posting_client() if ai: click.echo(click.style(f'Generating tweet with AI...', fg='yellow')) @@ -67,8 +67,7 @@ def post(ai, from_file, text, prompt): tweet_text = text click.echo(click.style(f'Posting: {tweet_text}', fg='green')) - client.create_tweet(text=tweet_text) - click.echo(click.style('Tweet posted successfully!', fg='green', bold=True)) + click.echo(click.style(post_tweet(client, tweet_text), fg='green', bold=True)) except Exception as e: click.echo(click.style(f'Error posting tweet: {str(e)}', fg='red')) @@ -92,14 +91,14 @@ def schedule_posts(ai, from_file, schedule_time, prompt): sys.exit(1) try: - client, _ = initialize_tweepy() + client, _ = initialize_posting_client() if ai: def send_ai_post(): try: click.echo(click.style(f'\nGenerating and posting tweet...', fg='yellow')) response = generate_response(prompt) - client.create_tweet(text=response) + post_tweet(client, response) click.echo(click.style(f'Tweet posted: {response}', fg='green')) click.echo(click.style(f'Next post scheduled for tomorrow at {schedule_time}', fg='cyan')) except Exception as e: @@ -125,7 +124,7 @@ def send_file_post(): return tweet_text = random.choice(lines) - client.create_tweet(text=tweet_text) + post_tweet(client, tweet_text) click.echo(click.style(f'\nTweet posted: {tweet_text}', fg='green')) click.echo(click.style(f'Next post scheduled for tomorrow at {schedule_time}', fg='cyan')) except Exception as e: diff --git a/src/functions.py b/src/functions.py index 4c4e43e..557ebff 100644 --- a/src/functions.py +++ b/src/functions.py @@ -4,6 +4,10 @@ from openai import OpenAI sys.path.append('../config') import keys +try: + from .xquik_client import post_tweet_with_xquik, uses_xquik_twitter_backend +except ImportError: + from xquik_client import post_tweet_with_xquik, uses_xquik_twitter_backend def initialize_tweepy(): @@ -12,6 +16,17 @@ def initialize_tweepy(): api = tweepy.API(auth) return client, api +def initialize_posting_client(): + if uses_xquik_twitter_backend(): + return None, None + return initialize_tweepy() + +def post_tweet(client, tweet_text): + if uses_xquik_twitter_backend(): + return post_tweet_with_xquik(tweet_text) + client.create_tweet(text=tweet_text) + return "Tweet posted successfully" + def get_formatted_date(): current_date = datetime.date.today() return current_date.strftime("%B %d, %Y") @@ -31,4 +46,4 @@ def generate_response(prompt): ) response_message = response.choices[0].message.content - return response_message.strip() \ No newline at end of file + return response_message.strip() diff --git a/src/instantly-tweet-from-openai.py b/src/instantly-tweet-from-openai.py index 4f2d86d..9b74492 100644 --- a/src/instantly-tweet-from-openai.py +++ b/src/instantly-tweet-from-openai.py @@ -2,15 +2,15 @@ import os sys.path.append(os.path.join(os.path.dirname(__file__), '..', 'config')) import keys -from functions import generate_response, initialize_tweepy, get_formatted_date +from functions import generate_response, initialize_posting_client, post_tweet prompt = "Create a short tweet about Motorbikes." response = generate_response(prompt) def send_post(): - client, _ = initialize_tweepy() + client, _ = initialize_posting_client() tweet_text = f"{response}" - client.create_tweet(text=tweet_text) + post_tweet(client, tweet_text) print("Tweet posted successfully") send_post() diff --git a/src/schedule-daily-post-from-file.py b/src/schedule-daily-post-from-file.py index 8f97a02..08beab6 100644 --- a/src/schedule-daily-post-from-file.py +++ b/src/schedule-daily-post-from-file.py @@ -7,15 +7,15 @@ sys.path.append(os.path.join(os.path.dirname(__file__), '..', 'config')) import keys sys.path.append(os.path.join(os.path.dirname(__file__), '..')) -from functions import initialize_tweepy, get_formatted_date +from functions import initialize_posting_client, post_tweet def send_post(): - client, _ = initialize_tweepy() + client, _ = initialize_posting_client() with open(os.path.join(os.path.dirname(__file__), '..', 'data', 'tweets.txt'), 'r') as file: lines = file.readlines() tweet_text = random.choice(lines).strip() - client.create_tweet(text=f"{tweet_text}") + post_tweet(client, f"{tweet_text}") print("Tweet posted successfully") diff --git a/src/schedule-daily-post-from-openai.py b/src/schedule-daily-post-from-openai.py index 3e3e718..f8458d0 100644 --- a/src/schedule-daily-post-from-openai.py +++ b/src/schedule-daily-post-from-openai.py @@ -6,15 +6,15 @@ sys.path.append(os.path.join(os.path.dirname(__file__), '..', 'config')) import keys -from functions import generate_response, initialize_tweepy, get_formatted_date +from functions import generate_response, initialize_posting_client, post_tweet def send_post(): prompt = "Create a short tweet about Motorbikes." response = generate_response(prompt) - client, _ = initialize_tweepy() + client, _ = initialize_posting_client() tweet_text = f"{response}" - client.create_tweet(text=tweet_text) + post_tweet(client, tweet_text) print("Tweet posted successfully") schedule.every().day.at("09:00").do(send_post) diff --git a/src/tweeter-from-code.py b/src/tweeter-from-code.py index d758c09..6dc8269 100644 --- a/src/tweeter-from-code.py +++ b/src/tweeter-from-code.py @@ -4,13 +4,13 @@ sys.path.append(os.path.join(os.path.dirname(__file__), '..', 'config')) import keys sys.path.append(os.path.join(os.path.dirname(__file__), '..')) -from functions import initialize_tweepy, get_formatted_date +from functions import get_formatted_date, initialize_posting_client, post_tweet def send_post(): - client, _ = initialize_tweepy() + client, _ = initialize_posting_client() formatted_date = get_formatted_date() - client.create_tweet(text=f"Hello Python 🐍. It is {formatted_date} today!🚀🚀.") + post_tweet(client, f"Hello Python 🐍. It is {formatted_date} today!🚀🚀.") print("Tweet posted successfully") send_post() diff --git a/src/tweeter-random-from-file.py b/src/tweeter-random-from-file.py index 47b3bcc..f548782 100644 --- a/src/tweeter-random-from-file.py +++ b/src/tweeter-random-from-file.py @@ -5,15 +5,15 @@ sys.path.append(os.path.join(os.path.dirname(__file__), '..', 'config')) import keys sys.path.append(os.path.join(os.path.dirname(__file__), '..')) -from functions import initialize_tweepy +from functions import initialize_posting_client, post_tweet def send_post(): - client, _ = initialize_tweepy() + client, _ = initialize_posting_client() with open(os.path.join(os.path.dirname(__file__), '..', 'data', 'tweets.txt'), 'r') as file: lines = file.readlines() tweet_text = random.choice(lines).strip() - client.create_tweet(text=f"{tweet_text}") + post_tweet(client, f"{tweet_text}") print("Tweet posted successfully") diff --git a/src/xquik_client.py b/src/xquik_client.py new file mode 100644 index 0000000..87d3f0e --- /dev/null +++ b/src/xquik_client.py @@ -0,0 +1,59 @@ +import json +import os +from urllib import error, request + + +XQUIK_TWEET_URL = os.environ.get("XQUIK_API_URL", "https://xquik.com/api/v1/x/tweets") + + +def uses_xquik_twitter_backend(): + return os.environ.get("TWITTER_BACKEND", "twitter").strip().lower() == "xquik" + + +def format_xquik_result(response_data): + tweet_url = response_data.get("url") + if tweet_url: + return f"Posted to Twitter/X via Xquik: {tweet_url}" + + write_action_id = response_data.get("writeActionId") + if write_action_id: + return f"Xquik accepted the tweet. Write action ID: {write_action_id}" + + return "Xquik accepted the tweet." + + +def post_tweet_with_xquik(tweet_text): + api_key = os.environ.get("XQUIK_API_KEY") + account = os.environ.get("XQUIK_ACCOUNT") + + if not api_key: + raise RuntimeError("Missing XQUIK_API_KEY.") + if not account: + raise RuntimeError("Missing XQUIK_ACCOUNT.") + + payload = json.dumps({ + "account": account, + "text": tweet_text, + }).encode("utf-8") + + post_request = request.Request( + XQUIK_TWEET_URL, + data=payload, + headers={ + "Authorization": f"Bearer {api_key}", + "Content-Type": "application/json", + }, + method="POST", + ) + + try: + with request.urlopen(post_request, timeout=30) as response: + response_body = response.read().decode("utf-8") + response_data = json.loads(response_body) if response_body else {} + if response.getcode() in (200, 202): + return format_xquik_result(response_data) + raise RuntimeError(f"Xquik returned HTTP {response.getcode()}.") + except error.HTTPError as exc: + raise RuntimeError(f"Xquik returned HTTP {exc.code}.") from exc + except error.URLError as exc: + raise RuntimeError(f"Could not reach Xquik: {exc.reason}") from exc diff --git a/tests/test_xquik_client.py b/tests/test_xquik_client.py new file mode 100644 index 0000000..edd15ca --- /dev/null +++ b/tests/test_xquik_client.py @@ -0,0 +1,60 @@ +import io +import os +import unittest +from unittest import mock + +from src import xquik_client + + +class XquikClientTest(unittest.TestCase): + def test_backend_defaults_to_twitter(self): + with mock.patch.dict(os.environ, {}, clear=True): + self.assertFalse(xquik_client.uses_xquik_twitter_backend()) + + def test_backend_can_select_xquik(self): + with mock.patch.dict(os.environ, {"TWITTER_BACKEND": "xquik"}, clear=True): + self.assertTrue(xquik_client.uses_xquik_twitter_backend()) + + def test_missing_account_is_clear(self): + with mock.patch.dict(os.environ, {"XQUIK_API_KEY": "key"}, clear=True): + with self.assertRaisesRegex(RuntimeError, "Missing XQUIK_ACCOUNT"): + xquik_client.post_tweet_with_xquik("hello") + + def test_accepted_write_action_is_reported(self): + env = { + "XQUIK_API_KEY": "key", + "XQUIK_ACCOUNT": "@example", + } + response = FakeResponse(b'{"writeActionId":"wa_123"}', 202) + + with mock.patch.dict(os.environ, env, clear=True): + with mock.patch.object(xquik_client.request, "urlopen", return_value=response) as urlopen: + result = xquik_client.post_tweet_with_xquik("hello") + + self.assertEqual(result, "Xquik accepted the tweet. Write action ID: wa_123") + sent_request = urlopen.call_args.args[0] + self.assertEqual(sent_request.headers["Authorization"], "Bearer key") + self.assertEqual(sent_request.headers["Content-type"], "application/json") + self.assertEqual(sent_request.data, b'{"account": "@example", "text": "hello"}') + + +class FakeResponse: + def __init__(self, body, status_code): + self._body = io.BytesIO(body) + self._status_code = status_code + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, traceback): + return False + + def read(self): + return self._body.read() + + def getcode(self): + return self._status_code + + +if __name__ == "__main__": + unittest.main() diff --git a/web.py b/web.py index 7b68e0c..d3405ef 100755 --- a/web.py +++ b/web.py @@ -18,7 +18,7 @@ sys.path.append(os.path.join(os.path.dirname(__file__), 'src')) import keys -from src.functions import generate_response, initialize_tweepy, get_formatted_date +from src.functions import generate_response, initialize_posting_client, initialize_tweepy, post_tweet as publish_tweet app = Flask(__name__) app.secret_key = 'twitter-bot-secret-key-change-in-production' @@ -91,7 +91,7 @@ def post_tweet(): data = request.get_json() tweet_type = data.get('type') - client, _ = initialize_tweepy() + client, _ = initialize_posting_client() if tweet_type == 'ai': prompt = data.get('prompt', 'Create a short tweet about Motorbikes.') @@ -108,11 +108,12 @@ def post_tweet(): else: return jsonify({'success': False, 'error': 'Invalid tweet type'}), 400 - client.create_tweet(text=tweet_text) + post_result = publish_tweet(client, tweet_text) return jsonify({ 'success': True, - 'tweet': tweet_text + 'tweet': tweet_text, + 'message': post_result }) except Exception as e: return jsonify({ @@ -219,13 +220,13 @@ def start_schedule(): schedule.clear() scheduled_jobs = [] - client, _ = initialize_tweepy() + client, _ = initialize_posting_client() if schedule_type == 'ai': def send_ai_post(): try: response = generate_response(prompt) - client.create_tweet(text=response) + publish_tweet(client, response) print(f'Posted tweet: {response}') except Exception as e: print(f'Error posting tweet: {str(e)}') @@ -243,7 +244,7 @@ def send_file_post(): tweets = get_tweets_from_file() if tweets: tweet_text = random.choice(tweets) - client.create_tweet(text=tweet_text) + publish_tweet(client, tweet_text) print(f'Posted tweet: {tweet_text}') except Exception as e: print(f'Error posting tweet: {str(e)}')