Add opt-in Xquik posting backend

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
 

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:

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()
+    return response_message.strip()

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()

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")
 

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)

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()

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")
 

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

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()

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)}')