From b54bc9178a467da4d46b13a2a482ea756c6e361e Mon Sep 17 00:00:00 2001 From: kriptoburak Date: Wed, 24 Jun 2026 04:56:32 +0300 Subject: [PATCH] feat: add xquik target adapter --- .env.example | 7 +++++ README.md | 16 +++++----- adapters/__init__.py | 9 +++++- adapters/base.py | 8 ++++- adapters/instagram.py | 7 ++++- adapters/twitter.py | 15 ++++++--- adapters/xquik.py | 72 +++++++++++++++++++++++++++++++++++++++++++ config.py | 8 +++++ core/orchestrator.py | 7 ++++- main.py | 11 +++++++ 10 files changed, 144 insertions(+), 16 deletions(-) create mode 100644 adapters/xquik.py diff --git a/.env.example b/.env.example index bd873c6..cbeca42 100644 --- a/.env.example +++ b/.env.example @@ -24,6 +24,13 @@ X_TOTP_SECRET= # Optional: saved session cookies path (default: storage/twitter_cookies.json) X_COOKIES_FILE=storage/twitter_cookies.json +# ================================ +# XQUIK (Target - API) +# ================================ +XQUIK_API_KEY= +XQUIK_ACCOUNT=@your_account +XQUIK_CREATE_TWEET_URL=https://xquik.com/api/v1/x/tweets + # ================================ # INSTAGRAM (Target - unofficial, cookie-based) # ================================ diff --git a/README.md b/README.md index 5d6bd71..ce0207e 100644 --- a/README.md +++ b/README.md @@ -29,15 +29,16 @@ | :-- | :-- | :-- | :-- | | 🦋 Bluesky | Source | `atproto` | App password | | ✖️ X (Twitter) | Target | `twikit` | Cookies _(recommended)_ or login | +| ✖️ X via Xquik | Target | `requests` | API key | | 📸 Instagram | Target | `instagrapi` | Session / login | ## Features -- **Multi-target fan-out** — one Bluesky post → many platforms in a single pass. -- **Media-aware** — downloads and re-uploads images attached to the source post. -- **Resilient state** — tracks the last processed post; only advances state when a target succeeds, so nothing is silently dropped. -- **Drop-in adapters** — each platform is isolated behind a common interface; missing credentials simply skip that target. -- **Cookie-based X auth** — sidesteps X's Cloudflare-protected login (essential on VPS/datacenter IPs). +- **Multi-target fan-out** - one Bluesky post → many platforms in a single pass. +- **Media-aware** - downloads and re-uploads images attached to the source post. +- **Resilient state** - tracks the last processed post; only advances state when a target succeeds, so nothing is silently dropped. +- **Drop-in adapters** - each platform is isolated behind a common interface; missing credentials simply skip that target. +- **Cookie-based X auth** - sidesteps X's Cloudflare-protected login (essential on VPS/datacenter IPs). ## Installation @@ -63,13 +64,14 @@ cp .env.example .env | :-- | :--: | :-- | | `BSKY_HANDLE` | ✅ | Your Bluesky handle (`name.bsky.social`) | | `BSKY_APP_PASSWORD` | ✅ | Bluesky **app password** (not your login password) | -| `X_AUTH_TOKEN` / `X_CT0` | ▶️ | Browser-exported X cookies — **preferred** auth | +| `X_AUTH_TOKEN` / `X_CT0` | ▶️ | Browser-exported X cookies - **preferred** auth | | `X_USERNAME` / `X_EMAIL` / `X_PASSWORD` | ▶️ | X login fallback (blocked on datacenter IPs) | | `X_TOTP_SECRET` | ⬜ | X 2FA base32 secret, if enabled | +| `XQUIK_API_KEY` / `XQUIK_ACCOUNT` | ▶️ | Xquik API key and connected X account | | `IG_USERNAME` / `IG_PASSWORD` | ▶️ | Instagram credentials | | `POLL_INTERVAL_SECONDS` | ⬜ | How often to check Bluesky (default `180`) | -> ▶️ = required only for that target. Enable any subset of X / Instagram. +> ▶️ = required only for that target. Enable any subset of X / Xquik / Instagram. **Getting your X cookies:** log into X in a browser → DevTools → **Application → Cookies → `https://x.com`** → copy the `auth_token` and `ct0` values into `.env`. See [TROUBLESHOOTING.md](TROUBLESHOOTING.md) for why this is recommended. diff --git a/adapters/__init__.py b/adapters/__init__.py index 1afe8de..f155c67 100644 --- a/adapters/__init__.py +++ b/adapters/__init__.py @@ -1,7 +1,14 @@ # adapters/__init__.py from .bluesky_source import BlueskySource from .twitter import TwitterAdapter +from .xquik import XquikAdapter from .instagram import InstagramAdapter from .base import SocialAdapter -__all__ = ["BlueskySource", "TwitterAdapter", "InstagramAdapter", "SocialAdapter"] +__all__ = [ + "BlueskySource", + "TwitterAdapter", + "XquikAdapter", + "InstagramAdapter", + "SocialAdapter", +] diff --git a/adapters/base.py b/adapters/base.py index 7e58ade..fa2902d 100644 --- a/adapters/base.py +++ b/adapters/base.py @@ -17,13 +17,19 @@ def validate_credentials(self) -> bool: pass @abstractmethod - def post(self, text: str, media_paths: Optional[List[str]] = None) -> Dict[str, Any]: + def post( + self, + text: str, + media_paths: Optional[List[str]] = None, + media_urls: Optional[List[str]] = None, + ) -> Dict[str, Any]: """ Posts content to the platform. Args: text: The text content of the post. media_paths: An optional list of local file paths to images/videos to upload. + media_urls: An optional list of public media URLs from the source post. Returns: A dictionary containing the result of the operation: diff --git a/adapters/instagram.py b/adapters/instagram.py index a8767e2..766e0dc 100644 --- a/adapters/instagram.py +++ b/adapters/instagram.py @@ -99,7 +99,12 @@ def validate_credentials(self) -> bool: logger.error(f"❌ Instagram login failed: {e}") return False - def post(self, text: str, media_paths: Optional[List[str]] = None) -> Dict[str, Any]: + def post( + self, + text: str, + media_paths: Optional[List[str]] = None, + media_urls: Optional[List[str]] = None, + ) -> Dict[str, Any]: """Post to Instagram using the authenticated session.""" if not self._is_authenticated: if not self.validate_credentials(): diff --git a/adapters/twitter.py b/adapters/twitter.py index 2661183..773396e 100644 --- a/adapters/twitter.py +++ b/adapters/twitter.py @@ -16,8 +16,8 @@ # # We must NOT use asyncio.run() (which creates and *closes* a fresh loop each # call): twikit's underlying httpx.AsyncClient binds its connection pool to the -# loop it first ran on. Closing the loop between calls — e.g. between -# upload_media() and create_tweet() when posting media — invalidates those +# loop it first ran on. Closing the loop between calls - e.g. between +# upload_media() and create_tweet() when posting media - invalidates those # connections and raises "Event loop is closed" on the next call. A persistent # loop keeps the connection pool valid across calls. _EVENT_LOOP: Optional[asyncio.AbstractEventLoop] = None @@ -98,7 +98,7 @@ def _save_cookies(self, client: Client) -> None: def validate_credentials(self) -> bool: """ Authenticate with X (Twitter), preferring cookie-based auth (which - avoids X's Cloudflare-protected login endpoint — important on + avoids X's Cloudflare-protected login endpoint - important on datacenter/VPS IPs). Order: 1. Reuse a previously saved cookies file. 2. Use auth_token + ct0 supplied via env (browser-exported cookies). @@ -120,7 +120,7 @@ def validate_credentials(self) -> bool: except Exception as e: logger.warning(f"Saved X session invalid: {e}") - # 2. Browser-exported cookies (auth_token + ct0) — bypasses login. + # 2. Browser-exported cookies (auth_token + ct0) - bypasses login. if self.auth_token and self.ct0: try: client.set_cookies( @@ -191,7 +191,12 @@ def _create_tweet_raw(self, client: Client, text: str, media_ids: List[str]) -> result = response["data"]["create_tweet"]["tweet_results"]["result"] return result["rest_id"] - def post(self, text: str, media_paths: Optional[List[str]] = None) -> Dict[str, Any]: + def post( + self, + text: str, + media_paths: Optional[List[str]] = None, + media_urls: Optional[List[str]] = None, + ) -> Dict[str, Any]: """ Posts a tweet with optional media. """ diff --git a/adapters/xquik.py b/adapters/xquik.py new file mode 100644 index 0000000..cd0cb53 --- /dev/null +++ b/adapters/xquik.py @@ -0,0 +1,72 @@ +from typing import Optional, List, Dict, Any +import requests + +from adapters.base import SocialAdapter +from utils.logger import logger + + +class XquikAdapter(SocialAdapter): + """Target adapter for posting to X through the Xquik API.""" + + def __init__( + self, + api_key: str, + account: str, + create_tweet_url: str = "https://xquik.com/api/v1/x/tweets", + ): + self.api_key = api_key + self.account = account + self.create_tweet_url = create_tweet_url + + def validate_credentials(self) -> bool: + if not self.api_key: + logger.error("❌ Xquik API key is missing.") + return False + if not self.account: + logger.error("❌ Xquik account is missing.") + return False + return True + + def post( + self, + text: str, + media_paths: Optional[List[str]] = None, + media_urls: Optional[List[str]] = None, + ) -> Dict[str, Any]: + """Post to Xquik using text and public media URLs.""" + if not self.validate_credentials(): + return {"success": False, "url": None, "error": "Xquik credentials missing."} + + body: Dict[str, Any] = { + "account": self.account, + "text": text, + } + if media_urls: + body["media"] = media_urls + + try: + response = requests.post( + self.create_tweet_url, + headers={ + "Content-Type": "application/json", + "X-API-Key": self.api_key, + }, + json=body, + timeout=30, + ) + response.raise_for_status() + data = response.json() + tweet_id = data.get("tweetId") + tweet_url = ( + f"https://x.com/i/web/status/{tweet_id}" + if tweet_id + else None + ) + logger.info(f"✅ Posted to Xquik: {tweet_url or 'tweet created'}") + return {"success": True, "url": tweet_url, "error": None} + except requests.exceptions.RequestException as e: + logger.error(f"❌ Xquik request failed: {e}") + return {"success": False, "url": None, "error": str(e)} + except ValueError as e: + logger.error(f"❌ Xquik returned invalid JSON: {e}") + return {"success": False, "url": None, "error": str(e)} diff --git a/config.py b/config.py index 9eaf94e..b0bcde6 100644 --- a/config.py +++ b/config.py @@ -19,6 +19,14 @@ class Config: X_AUTH_TOKEN: str = os.getenv("X_AUTH_TOKEN", "") X_CT0: str = os.getenv("X_CT0", "") + # Xquik API target + XQUIK_API_KEY: str = os.getenv("XQUIK_API_KEY", "") + XQUIK_ACCOUNT: str = os.getenv("XQUIK_ACCOUNT", "") + XQUIK_CREATE_TWEET_URL: str = os.getenv( + "XQUIK_CREATE_TWEET_URL", + "https://xquik.com/api/v1/x/tweets", + ) + # Instagram IG_USERNAME: str = os.getenv("IG_USERNAME", "") IG_PASSWORD: str = os.getenv("IG_PASSWORD", "") diff --git a/core/orchestrator.py b/core/orchestrator.py index d91bd9e..94b7e4e 100644 --- a/core/orchestrator.py +++ b/core/orchestrator.py @@ -39,6 +39,11 @@ def crosspost(self, post_data: Dict[str, Any]) -> List[Dict[str, Any]]: """ text = post_data.get("text", "") media_items = post_data.get("media", []) + media_urls = [ + item.get("url", "") + for item in media_items + if item.get("url", "").startswith(("http://", "https://")) + ] post_uri = post_data.get("uri", "") # Download media files @@ -54,7 +59,7 @@ def crosspost(self, post_data: Dict[str, Any]) -> List[Dict[str, Any]]: results = [] for adapter in self.adapters: logger.info(f"📤 Posting to {adapter.__class__.__name__}...") - result = adapter.post(text, media_paths) + result = adapter.post(text, media_paths, media_urls) results.append(result) if result.get("success"): diff --git a/main.py b/main.py index b8161c8..21c5903 100644 --- a/main.py +++ b/main.py @@ -8,6 +8,7 @@ from core.state_manager import StateManager from adapters.bluesky_source import BlueskySource from adapters.twitter import TwitterAdapter +from adapters.xquik import XquikAdapter from adapters.instagram import InstagramAdapter @@ -53,6 +54,16 @@ def main(): else: logger.warning("⚠️ X (Twitter) credentials not found. Skipping X adapter.") + if config.XQUIK_API_KEY and config.XQUIK_ACCOUNT: + xquik_adapter = XquikAdapter( + api_key=config.XQUIK_API_KEY, + account=config.XQUIK_ACCOUNT, + create_tweet_url=config.XQUIK_CREATE_TWEET_URL, + ) + orchestrator.register_adapter(xquik_adapter) + else: + logger.warning("⚠️ Xquik credentials not found. Skipping Xquik adapter.") + # Register Instagram adapter if credentials are present if config.IG_USERNAME and config.IG_PASSWORD: instagram_adapter = InstagramAdapter(