Add optional Xquik posting backend

README.md

@@ -64,12 +64,17 @@ A simple, configurable Twitter bot that generates and posts tweets using local A
 
    ```json
    {
+     "post_backend": "twitter",
      "twitter": {
        "consumer_key": "YOUR_CONSUMER_KEY",
        "consumer_secret": "YOUR_CONSUMER_SECRET",
        "access_token": "YOUR_ACCESS_TOKEN",
        "access_secret": "YOUR_ACCESS_SECRET"
      },
+     "xquik": {
+       "api_key": "YOUR_XQUIK_API_KEY",
+       "api_url": "https://xquik.com/api/v1/x/tweets"
+     },
      "ollama": {
        "host": "http://localhost:11434/api/generate",
        "model": "llama3"
@@ -80,6 +85,11 @@ A simple, configurable Twitter bot that generates and posts tweets using local A
    }
    ```
 
+   `post_backend` defaults to `twitter`. To post through Xquik instead,
+   set `"post_backend": "xquik"` and provide `xquik.api_key` or set
+   `XQUIK_API_KEY` in your environment. `xquik.api_url` defaults to
+   `https://xquik.com/api/v1/x/tweets`.
+
 ## Usage
 
 ### Quick Start
@@ -106,6 +116,12 @@ Once testing is successful:
 
 - `consumer_key`, `consumer_secret`, `access_token`, `access_secret`: Your Twitter API keys
 
+### Posting Backend
+
+- `post_backend`: Use `twitter` for direct Twitter API posting or `xquik` for Xquik
+- `xquik.api_key`: Xquik API key. You can also set `XQUIK_API_KEY`
+- `xquik.api_url`: Optional Xquik create-tweet URL
+
 ### Ollama
 
 - `host`: Ollama API endpoint (default: http://localhost:11434/api/generate)

TwitterBotTemplate.py

@@ -12,8 +12,10 @@
 
 @dataclass
 class Config:
-    twitter: dict[str, str]  # Twitter API keys: consumer_key, consumer_secret, access_token, access_secret
     ollama: dict[str, str]  # Ollama config: host, model
+    twitter: dict[str, str] = field(default_factory=dict)  # Twitter API keys: consumer_key, consumer_secret, access_token, access_secret
+    xquik: dict[str, str] = field(default_factory=dict)  # Xquik config: api_key, api_url
+    post_backend: str = "twitter"  # Posting backend: twitter or xquik
     prompt_template: str = "Write a short tweet about: {data}"  # Template for LLM prompt
     source_url: Optional[str] = None  # Optional URL to fetch data from
     post_prob: float = 1.0  # Probability (0-1) to post each run
@@ -67,20 +69,58 @@ def load_config(path: str) -> Config:
     with open(path, 'r') as f:
         data = json.load(f)
     # Validate required keys
-    required = ['twitter', 'ollama']
+    required = ['ollama']
     for key in required:
         if key not in data:
             raise ValueError(f"Missing required config key: {key}")
-    twitter = data['twitter']
-    for tkey in ['consumer_key', 'consumer_secret', 'access_token', 'access_secret']:
-        if tkey not in twitter:
-            raise ValueError(f"Missing twitter key: {tkey}")
+    post_backend = data.get('post_backend', 'twitter')
+    if post_backend not in {'twitter', 'xquik'}:
+        raise ValueError("post_backend must be 'twitter' or 'xquik'")
+    twitter = data.get('twitter', {})
+    if not isinstance(twitter, dict):
+        raise ValueError("twitter must be an object")
+    if post_backend == 'twitter':
+        for tkey in ['consumer_key', 'consumer_secret', 'access_token', 'access_secret']:
+            if tkey not in twitter:
+                raise ValueError(f"Missing twitter key: {tkey}")
+    xquik = data.get('xquik', {})
+    if not isinstance(xquik, dict):
+        raise ValueError("xquik must be an object")
+    if post_backend == 'xquik':
+        api_key = xquik.get('api_key') or os.environ.get('XQUIK_API_KEY')
+        if not api_key:
+            raise ValueError("Missing Xquik API key: set xquik.api_key or XQUIK_API_KEY")
     ollama = data['ollama']
     for okey in ['host', 'model']:
         if okey not in ollama:
             raise ValueError(f"Missing ollama key: {okey}")
     return Config(**data)
 
+def get_xquik_posting_config(config: Config) -> tuple[Optional[str], str]:
+    """Read Xquik posting settings from config or environment variables.
+    Args: config (Config)
+    Returns: API key and create-tweet URL
+    """
+    api_key = config.xquik.get('api_key') or os.environ.get('XQUIK_API_KEY')
+    api_url = (
+        config.xquik.get('api_url')
+        or os.environ.get('XQUIK_API_URL')
+        or 'https://xquik.com/api/v1/x/tweets'
+    )
+    return api_key, api_url
+
+def get_xquik_tweet_id(data: dict[str, Any]) -> Optional[str]:
+    """Extract a tweet ID from an Xquik create-tweet response."""
+    tweet_id = data.get('tweetId')
+    if isinstance(tweet_id, str):
+        return tweet_id
+    nested = data.get('data')
+    if isinstance(nested, dict):
+        nested_id = nested.get('id')
+        if isinstance(nested_id, str):
+            return nested_id
+    return None
+
 def fetch_data(url: str) -> str:
     """Fetch data from URL (JSON or text).
     Args: url (str)
@@ -128,6 +168,8 @@ def post_tweet(text: str, config: Config) -> Optional[str]:
     if config.dry_run:
         logging.info(f"Dry run: would post tweet: {text}")
         return "dry_run_id"
+    if config.post_backend == 'xquik':
+        return post_tweet_xquik(text, config)
     try:
         from requests_oauthlib import OAuth1Session
     except ImportError as e:
@@ -150,6 +192,34 @@ def post_tweet(text: str, config: Config) -> Optional[str]:
         logging.error(f"Tweet post failed: {e}")
     return None
 
+def post_tweet_xquik(text: str, config: Config) -> Optional[str]:
+    """Post a tweet through Xquik.
+    Args: text (str), config (Config)
+    Returns: Tweet ID string or None on failure
+    """
+    api_key, api_url = get_xquik_posting_config(config)
+    if not api_key:
+        logging.error("Xquik API key missing")
+        return None
+    try:
+        resp = requests.post(
+            api_url,
+            headers={'x-api-key': api_key, 'Content-Type': 'application/json'},
+            json={'text': text},
+            timeout=30
+        )
+        if resp.status_code in {200, 201}:
+            data = resp.json()
+            tweet_id = get_xquik_tweet_id(data)
+            if tweet_id:
+                return tweet_id
+            logging.error("Xquik post failed: response missing tweetId")
+        else:
+            logging.error(f"Xquik post failed: {resp.status_code} {resp.text}")
+    except Exception as e:
+        logging.error(f"Xquik post failed: {e}")
+    return None
+
 def acquire_lock(log_path: str) -> bool:
     """Acquire file lock to prevent concurrent runs.
     Args: log_path (str)
@@ -205,4 +275,4 @@ def main():
         print(f"Error: {e}")
 
 if __name__ == "__main__":
-    main()
+    main()

config.json.example

@@ -1,15 +1,20 @@
 {
+  "post_backend": "twitter",
   "twitter": {
     "consumer_key": "YOUR_CONSUMER_KEY",
     "consumer_secret": "YOUR_CONSUMER_SECRET",
     "access_token": "YOUR_ACCESS_TOKEN",
     "access_secret": "YOUR_ACCESS_SECRET"
   },
+  "xquik": {
+    "api_key": "YOUR_XQUIK_API_KEY",
+    "api_url": "https://xquik.com/api/v1/x/tweets"
+  },
   "ollama": {
     "host": "http://localhost:11434/api/generate",
     "model": "llama3"
   },
   "prompt_template": "Write a short tweet about: {data}",
   "source_url": "https://api.example.com/data",
   "dry_run": true
-}
+}

tests/test_bot.py

@@ -1,6 +1,6 @@
 import pytest
 from unittest.mock import patch, MagicMock
-from datetime import datetime
+from datetime import datetime, timezone
 import json
 import os
 import tempfile
@@ -28,6 +28,19 @@ def test_load_config():
     os.unlink(f.name)
     assert config.ollama['model'] == 'llama3'
 
+def test_load_config_xquik_backend():
+    data = {
+        'post_backend': 'xquik',
+        'xquik': {'api_key': 'xq_key'},
+        'ollama': {'host': 'http://localhost:11434/api/generate', 'model': 'llama3'}
+    }
+    with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as f:
+        json.dump(data, f)
+        f.flush()
+        config = load_config(f.name)
+    os.unlink(f.name)
+    assert config.post_backend == 'xquik'
+
 def test_db_functions():
     with tempfile.NamedTemporaryFile(delete=False) as f:
         db_path = f.name
@@ -67,3 +80,23 @@ def test_post_tweet(mock_oauth_class, config):
     tweet_id = post_tweet('test', config)
     assert tweet_id == '123'
     mock_oauth.post.assert_called_once_with('https://api.twitter.com/2/tweets', json={'text': 'test'})
+
+@patch('TwitterBotTemplate.requests.post')
+def test_post_tweet_xquik(mock_post, config):
+    config.post_backend = 'xquik'
+    config.xquik = {
+        'api_key': 'xq_key',
+        'api_url': 'https://xquik.example/api/v1/x/tweets'
+    }
+    mock_resp = MagicMock()
+    mock_resp.status_code = 200
+    mock_resp.json.return_value = {'tweetId': '456'}
+    mock_post.return_value = mock_resp
+    tweet_id = post_tweet('test', config)
+    assert tweet_id == '456'
+    mock_post.assert_called_once_with(
+        'https://xquik.example/api/v1/x/tweets',
+        headers={'x-api-key': 'xq_key', 'Content-Type': 'application/json'},
+        json={'text': 'test'},
+        timeout=30
+    )