feat(docloaders): add Xquik tweet loader

README.md

@@ -142,6 +142,45 @@ The output will be a dictionary like the following:
     }
 }
 ```
+
+### Scrape X posts with Xquik
+
+Public X status pages can return login or client-rendered shells instead of the
+post content. Install the optional Xquik integration to retrieve the published
+tweet data before ScrapeGraphAI extracts the requested fields:
+
+```bash
+pip install scrapegraphai x_twitter_scraper
+```
+
+```python
+import os
+
+from scrapegraphai.graphs import SmartScraperGraph
+
+graph = SmartScraperGraph(
+    prompt="Extract the post text, author username, creation time, and metrics.",
+    source="https://x.com/example/status/1893456789012345678",
+    config={
+        "llm": {
+            "api_key": os.environ["OPENAI_API_KEY"],
+            "model": "openai/gpt-4o-mini",
+        },
+        "xquik": {
+            "api_key": os.environ["X_TWITTER_SCRAPER_API_KEY"],
+            "timeout": 30,
+        },
+    },
+)
+
+result = graph.run()
+```
+
+The source may also be a numeric tweet ID. See the
+[complete Xquik example](examples/xquik_tweet/README.md).
+
+Xquik is an independent third-party service. Not affiliated with X Corp.
+
 There are other pipelines that can be used to extract information from multiple pages, generate Python scripts, or even generate audio files.
 
 | Pipeline Name           | Description                                                                                                      |

examples/xquik_tweet/.env.example

@@ -0,0 +1,2 @@
+OPENAI_API_KEY=
+X_TWITTER_SCRAPER_API_KEY=

examples/xquik_tweet/README.md

@@ -0,0 +1,22 @@
+# Extract structured data from an X post
+
+Use Xquik when a public X status page does not expose stable HTML to the
+browser loader. The integration retrieves the post through the published tweet
+lookup API, then runs the normal SmartScraperGraph extraction pipeline.
+
+Install the optional dependency:
+
+```bash
+pip install scrapegraphai x_twitter_scraper
+```
+
+Copy `.env.example` to `.env` and set both API keys. Then run:
+
+```bash
+python examples/xquik_tweet/xquik_tweet.py
+```
+
+The source can be a public `x.com` or `twitter.com` status URL. It can also be
+a numeric tweet ID.
+
+Xquik is an independent third-party service. Not affiliated with X Corp.

examples/xquik_tweet/xquik_tweet.py

@@ -0,0 +1,26 @@
+"""Extract structured fields from a public X post through Xquik."""
+
+import os
+
+from dotenv import load_dotenv
+
+from scrapegraphai.graphs import SmartScraperGraph
+
+load_dotenv()
+
+graph = SmartScraperGraph(
+    prompt="Extract the post text, author username, creation time, and metrics.",
+    source="https://x.com/example/status/1893456789012345678",
+    config={
+        "llm": {
+            "api_key": os.environ["OPENAI_API_KEY"],
+            "model": "openai/gpt-4o-mini",
+        },
+        "xquik": {
+            "api_key": os.environ["X_TWITTER_SCRAPER_API_KEY"],
+            "timeout": 30,
+        },
+    },
+)
+
+print(graph.run())

scrapegraphai/docloaders/__init__.py

@@ -7,6 +7,7 @@
 
 from .browser_base import browser_base_fetch
 from .scrape_do import scrape_do_fetch
+from .xquik import XquikLoader
 
 _LAZY_MODULES = {
     "ChromiumLoader": ".chromium",
@@ -17,6 +18,7 @@
 def __getattr__(name):
     if name in _LAZY_MODULES:
         import importlib
+
         module = importlib.import_module(_LAZY_MODULES[name], __package__)
         return getattr(module, name)
     raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
@@ -27,4 +29,5 @@ def __getattr__(name):
     "ChromiumLoader",
     "PlasmateLoader",
     "scrape_do_fetch",
+    "XquikLoader",
 ]

scrapegraphai/docloaders/xquik.py

@@ -0,0 +1,123 @@
+"""Load public X posts through the Xquik API."""
+
+from __future__ import annotations
+
+import json
+import re
+from collections.abc import Iterator, Sequence
+from typing import Any, Protocol, cast
+
+from langchain_community.document_loaders.base import BaseLoader
+from langchain_core.documents import Document
+
+_TWEET_ID = re.compile(r"^\d{1,20}$")
+_TWEET_URL = re.compile(
+    r"^https?://(?:(?:www|mobile)\.)?(?:x|twitter)\.com/[^/]+/status/"
+    r"(\d{1,20})(?:[/?#].*)?$",
+    re.IGNORECASE,
+)
+
+
+class _TweetResponse(Protocol):
+    def model_dump(self, *, by_alias: bool, exclude_none: bool) -> dict[str, Any]: ...
+
+
+class _TweetsResource(Protocol):
+    def retrieve(self, tweet_id: str) -> _TweetResponse: ...
+
+
+class _XResource(Protocol):
+    @property
+    def tweets(self) -> _TweetsResource: ...
+
+
+class _XquikClient(Protocol):
+    @property
+    def x(self) -> _XResource: ...
+
+    def close(self) -> None: ...
+
+
+def extract_tweet_id(source: str) -> str:
+    """Return the tweet ID from a numeric ID or public X status URL."""
+    value = source.strip()
+    if _TWEET_ID.fullmatch(value):
+        return value
+
+    match = _TWEET_URL.fullmatch(value)
+    if match:
+        return match.group(1)
+
+    raise ValueError(
+        "XquikLoader requires a numeric tweet ID or an x.com/twitter.com status URL."
+    )
+
+
+def _create_client(api_key: str | None, timeout: float) -> _XquikClient:
+    try:
+        from x_twitter_scraper import XTwitterScraper
+    except ImportError as exc:
+        raise ImportError(
+            "XquikLoader requires x_twitter_scraper. "
+            "Install it with `pip install x_twitter_scraper`."
+        ) from exc
+
+    return cast(
+        _XquikClient,
+        XTwitterScraper(api_key=api_key, timeout=timeout),
+    )
+
+
+class XquikLoader(BaseLoader):
+    """Fetch public X posts as LangChain documents.
+
+    Each source must be a numeric tweet ID or a public ``x.com`` or
+    ``twitter.com`` status URL. The loader uses the published Xquik tweet lookup
+    API and stores the complete response as JSON for downstream extraction.
+
+    Args:
+        sources: Tweet IDs or public status URLs.
+        api_key: Xquik API key. The SDK reads ``X_TWITTER_SCRAPER_API_KEY`` when
+            omitted.
+        timeout: Request timeout in seconds.
+        client: Optional configured Xquik client.
+    """
+
+    def __init__(
+        self,
+        sources: Sequence[str],
+        *,
+        api_key: str | None = None,
+        timeout: float = 30.0,
+        client: _XquikClient | None = None,
+    ) -> None:
+        self.sources = list(sources)
+        self.api_key = api_key
+        self.timeout = timeout
+        self._client = client
+
+    def lazy_load(self) -> Iterator[Document]:
+        """Yield one document for each X post."""
+        owns_client = self._client is None
+        client = self._client or _create_client(self.api_key, self.timeout)
+
+        try:
+            for source in self.sources:
+                tweet_id = extract_tweet_id(source)
+                response = client.x.tweets.retrieve(tweet_id)
+                payload = response.model_dump(by_alias=True, exclude_none=True)
+                yield Document(
+                    page_content=json.dumps(
+                        payload,
+                        ensure_ascii=False,
+                        sort_keys=True,
+                    ),
+                    metadata={
+                        "source": source,
+                        "loader": "xquik",
+                        "tweet_id": tweet_id,
+                    },
+                )
+        finally:
+            if owns_client:
+                client.close()

scrapegraphai/graphs/smart_scraper_graph.py

@@ -64,7 +64,11 @@ def __init__(
     ):
         super().__init__(prompt, config, source, schema)
 
-        self.input_key = "url" if source.startswith("http") else "local_dir"
+        self.input_key = (
+            "url"
+            if source.startswith("http") or config.get("xquik") is not None
+            else "local_dir"
+        )
 
         # for detailed logging of the SmartScraper API set it to True
         self.verbose = config.get("verbose", False)
@@ -104,6 +108,7 @@ def _create_graph(self) -> BaseGraph:
                 "loader_kwargs": self.config.get("loader_kwargs", {}),
                 "browser_base": self.config.get("browser_base"),
                 "scrape_do": self.config.get("scrape_do"),
+                "xquik": self.config.get("xquik"),
                 "storage_state": self.config.get("storage_state"),
             },
         )

scrapegraphai/nodes/fetch_node.py

@@ -2,9 +2,9 @@
 FetchNode Module
 """
 
+import concurrent.futures
 import json
 from typing import List, Optional
-import concurrent.futures
 
 import requests
 from langchain_core.documents import Document
@@ -86,6 +86,8 @@ def __init__(
             None if node_config is None else node_config.get("plasmate", None)
         )
 
+        self.xquik = None if node_config is None else node_config.get("xquik", None)
+
         self.storage_state = (
             None if node_config is None else node_config.get("storage_state", None)
         )
@@ -182,6 +184,7 @@ def load_file_content(self, source, input_type):
 
         if input_type == "pdf":
             from langchain_community.document_loaders import PyPDFLoader
+
             loader = PyPDFLoader(source)
             # PyPDFLoader.load() can be blocking for large PDFs. Run it in a thread and
             # enforce the configured timeout if provided.
@@ -317,7 +320,17 @@ def handle_web_source(self, state, source):
             if "timeout" not in loader_kwargs and self.timeout is not None:
                 loader_kwargs["timeout"] = self.timeout
 
-            if self.browser_base:
+            if self.xquik is not None:
+                from ..docloaders.xquik import XquikLoader
+
+                xquik_config = self.xquik if isinstance(self.xquik, dict) else {}
+                xquik_loader = XquikLoader(
+                    [source],
+                    api_key=xquik_config.get("api_key"),
+                    timeout=xquik_config.get("timeout", self.timeout or 30),
+                )
+                document = xquik_loader.load()
+            elif self.browser_base:
                 try:
                     from ..docloaders.browser_base import browser_base_fetch
                 except ImportError:
@@ -385,7 +398,7 @@ def handle_web_source(self, state, source):
 
             parsed_content = document[0].page_content
 
-            if (
+            if self.xquik is None and (
                 (
                     isinstance(self.llm_model, ChatOpenAI)
                     or isinstance(self.llm_model, AzureChatOpenAI)
@@ -397,8 +410,13 @@ def handle_web_source(self, state, source):
             ):
                 parsed_content = convert_to_md(document[0].page_content, parsed_content)
 
+            metadata = (
+                dict(document[0].metadata)
+                if self.xquik is not None
+                else {"source": "html file"}
+            )
             compressed_document = [
-                Document(page_content=parsed_content, metadata={"source": "html file"})
+                Document(page_content=parsed_content, metadata=metadata)
             ]
         state["doc"] = document
         state.update(

tests/nodes/fetch_node_test.py

@@ -1,3 +1,5 @@
+from typing import Any
+
 from langchain_core.documents import Document
 
 from scrapegraphai.nodes import FetchNode
@@ -39,6 +41,35 @@ def test_fetch_html(mocker):
     )
 
 
+def test_fetch_x_status_with_xquik(mocker: Any) -> None:
+    source = "https://x.com/example/status/1893456789012345678"
+    document = Document(
+        page_content='{"tweet":{"id":"1893456789012345678"}}',
+        metadata={"source": source, "loader": "xquik"},
+    )
+    loader_class = mocker.patch("scrapegraphai.docloaders.xquik.XquikLoader")
+    loader_class.return_value.load.return_value = [document]
+    convert_to_md = mocker.patch("scrapegraphai.nodes.fetch_node.convert_to_md")
+    node = FetchNode(
+        input="url",
+        output=["doc"],
+        node_config={
+            "force": True,
+            "xquik": {"api_key": "test-key", "timeout": 12},
+        },
+    )
+
+    result = node.execute({"url": source})
+
+    loader_class.assert_called_once_with(
+        [source],
+        api_key="test-key",
+        timeout=12,
+    )
+    convert_to_md.assert_not_called()
+    assert result["doc"] == [document]
+
+
 def test_fetch_json():
     node = FetchNode(
         input="json",