Add optional Xquik posting backend

README.md

@@ -78,7 +78,14 @@ 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: use Xquik for posting instead of Tweepy by setting:
+   ```sh
+   export TWITTER_BACKEND=xquik
+   export XQUIK_API_KEY="your_xquik_api_key"
+   export XQUIK_ACCOUNT="@your_connected_x_account"
+   ```
+   `XQUIK_BASE_URL` defaults to `https://xquik.com`.
+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_tweepy, get_formatted_date, publish_tweet
 
 
 @click.group()
@@ -38,8 +38,6 @@ def post(ai, from_file, text, prompt):
         sys.exit(1)
 
     try:
-        client, _ = initialize_tweepy()
-
         if ai:
             click.echo(click.style(f'Generating tweet with AI...', fg='yellow'))
             click.echo(click.style(f'Prompt: {prompt}', fg='cyan'))
@@ -67,7 +65,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)
+        publish_tweet(tweet_text)
         click.echo(click.style('Tweet posted successfully!', fg='green', bold=True))
 
     except Exception as e:
@@ -92,14 +90,12 @@ def schedule_posts(ai, from_file, schedule_time, prompt):
         sys.exit(1)
 
     try:
-        client, _ = initialize_tweepy()
-
         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)
+                    publish_tweet(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 +121,7 @@ def send_file_post():
                         return
 
                     tweet_text = random.choice(lines)
-                    client.create_tweet(text=tweet_text)
+                    publish_tweet(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

@@ -1,6 +1,10 @@
 import tweepy
 import datetime
+import json
+import os
 import sys
+import urllib.error
+import urllib.request
 from openai import OpenAI
 sys.path.append('../config')
 import keys  
@@ -12,6 +16,41 @@ def initialize_tweepy():
     api = tweepy.API(auth)
     return client, api
 
+def publish_tweet(text, client=None):
+    if os.getenv("TWITTER_BACKEND", "tweepy").lower() == "xquik":
+        return publish_tweet_with_xquik(text)
+
+    if client is None:
+        client, _ = initialize_tweepy()
+    return client.create_tweet(text=text)
+
+def publish_tweet_with_xquik(text):
+    api_key = os.getenv("XQUIK_API_KEY")
+    account = os.getenv("XQUIK_ACCOUNT")
+    base_url = os.getenv("XQUIK_BASE_URL", "https://xquik.com").rstrip("/")
+
+    if not api_key:
+        raise ValueError("XQUIK_API_KEY is required when TWITTER_BACKEND=xquik")
+    if not account:
+        raise ValueError("XQUIK_ACCOUNT is required when TWITTER_BACKEND=xquik")
+
+    payload = json.dumps({"account": account, "text": text}).encode("utf-8")
+    request = urllib.request.Request(
+        f"{base_url}/api/v1/x/tweets",
+        data=payload,
+        headers={"Content-Type": "application/json", "x-api-key": api_key},
+        method="POST"
+    )
+
+    try:
+        with urllib.request.urlopen(request, timeout=30) as response:
+            body = response.read().decode("utf-8")
+            return json.loads(body) if body else {}
+    except urllib.error.HTTPError as error:
+        raise RuntimeError(f"Xquik tweet post failed with HTTP {error.code}") from error
+    except urllib.error.URLError as error:
+        raise RuntimeError(f"Xquik tweet post failed: {error.reason}") from error
+
 def get_formatted_date():
     current_date = datetime.date.today()
     return current_date.strftime("%B %d, %Y")
@@ -31,4 +70,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,14 @@
 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, publish_tweet
 
 prompt = "Create a short tweet about Motorbikes."
 response = generate_response(prompt)
 
 def send_post():
-    client, _ = initialize_tweepy()
     tweet_text = f"{response}"
-    client.create_tweet(text=tweet_text)
+    publish_tweet(tweet_text)
     print("Tweet posted successfully")
 
 send_post()

src/schedule-daily-post-from-file.py

@@ -7,15 +7,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 publish_tweet
 
 def send_post():
-    client, _ = initialize_tweepy()
-
     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}")
+    publish_tweet(f"{tweet_text}")
 
     print("Tweet posted successfully")
 

src/schedule-daily-post-from-openai.py

@@ -6,15 +6,14 @@
 
 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, publish_tweet
 
 def send_post():
     prompt = "Create a short tweet about Motorbikes."
     response = generate_response(prompt)
 
-    client, _ = initialize_tweepy()
     tweet_text = f"{response}"
-    client.create_tweet(text=tweet_text)
+    publish_tweet(tweet_text)
     print("Tweet posted successfully")
 
 schedule.every().day.at("09:00").do(send_post)

src/tweeter-from-code.py

@@ -4,13 +4,12 @@
 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, publish_tweet
 
 def send_post():
-    client, _ = initialize_tweepy()
     formatted_date = get_formatted_date()
 
-    client.create_tweet(text=f"Hello Python 🐍. It is {formatted_date} today!🚀🚀.")
+    publish_tweet(f"Hello Python 🐍. It is {formatted_date} today!🚀🚀.")
     print("Tweet posted successfully")
 
 send_post()

src/tweeter-random-from-file.py

@@ -5,15 +5,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
+from functions import publish_tweet
 
 def send_post():
-    client, _ = initialize_tweepy()
-
     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}")
+    publish_tweet(f"{tweet_text}")
 
     print("Tweet posted successfully")
 

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_tweepy, get_formatted_date, publish_tweet
 
 app = Flask(__name__)
 app.secret_key = 'twitter-bot-secret-key-change-in-production'
@@ -91,8 +91,6 @@ def post_tweet():
         data = request.get_json()
         tweet_type = data.get('type')
 
-        client, _ = initialize_tweepy()
-
         if tweet_type == 'ai':
             prompt = data.get('prompt', 'Create a short tweet about Motorbikes.')
             tweet_text = generate_response(prompt)
@@ -108,7 +106,7 @@ def post_tweet():
         else:
             return jsonify({'success': False, 'error': 'Invalid tweet type'}), 400
 
-        client.create_tweet(text=tweet_text)
+        publish_tweet(tweet_text)
 
         return jsonify({
             'success': True,
@@ -219,13 +217,11 @@ def start_schedule():
         schedule.clear()
         scheduled_jobs = []
 
-        client, _ = initialize_tweepy()
-
         if schedule_type == 'ai':
             def send_ai_post():
                 try:
                     response = generate_response(prompt)
-                    client.create_tweet(text=response)
+                    publish_tweet(response)
                     print(f'Posted tweet: {response}')
                 except Exception as e:
                     print(f'Error posting tweet: {str(e)}')
@@ -243,7 +239,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(tweet_text)
                         print(f'Posted tweet: {tweet_text}')
                 except Exception as e:
                     print(f'Error posting tweet: {str(e)}')