diff --git a/README.md b/README.md index 51bf702..33f6705 100644 --- a/README.md +++ b/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) diff --git a/TwitterBotTemplate.py b/TwitterBotTemplate.py index 561fcad..dab1348 100644 --- a/TwitterBotTemplate.py +++ b/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() \ No newline at end of file + main() diff --git a/config.json.example b/config.json.example index d47eabf..87a4dc3 100644 --- a/config.json.example +++ b/config.json.example @@ -1,10 +1,15 @@ { + "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" @@ -12,4 +17,4 @@ "prompt_template": "Write a short tweet about: {data}", "source_url": "https://api.example.com/data", "dry_run": true -} \ No newline at end of file +} diff --git a/tests/test_bot.py b/tests/test_bot.py index 4b97a8b..beaa7f4 100644 --- a/tests/test_bot.py +++ b/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 + )