Add optional Xquik target adapter

.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)
 # ================================

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.
 

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",
+]

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:

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

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

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

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

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

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(