diff --git a/README.md b/README.md index 989f2674..54c78678 100644 --- a/README.md +++ b/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 | diff --git a/examples/xquik_tweet/.env.example b/examples/xquik_tweet/.env.example new file mode 100644 index 00000000..bcaa8c5a --- /dev/null +++ b/examples/xquik_tweet/.env.example @@ -0,0 +1,2 @@ +OPENAI_API_KEY= +X_TWITTER_SCRAPER_API_KEY= diff --git a/examples/xquik_tweet/README.md b/examples/xquik_tweet/README.md new file mode 100644 index 00000000..e6a9cd5e --- /dev/null +++ b/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. diff --git a/examples/xquik_tweet/xquik_tweet.py b/examples/xquik_tweet/xquik_tweet.py new file mode 100644 index 00000000..6aac84e0 --- /dev/null +++ b/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()) diff --git a/scrapegraphai/docloaders/__init__.py b/scrapegraphai/docloaders/__init__.py index a4e8e383..40f17505 100644 --- a/scrapegraphai/docloaders/__init__.py +++ b/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", ] diff --git a/scrapegraphai/docloaders/xquik.py b/scrapegraphai/docloaders/xquik.py new file mode 100644 index 00000000..82250d56 --- /dev/null +++ b/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() diff --git a/scrapegraphai/graphs/smart_scraper_graph.py b/scrapegraphai/graphs/smart_scraper_graph.py index b29d038a..e5d93a65 100644 --- a/scrapegraphai/graphs/smart_scraper_graph.py +++ b/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"), }, ) diff --git a/scrapegraphai/nodes/fetch_node.py b/scrapegraphai/nodes/fetch_node.py index c55b96f6..0fa9f134 100644 --- a/scrapegraphai/nodes/fetch_node.py +++ b/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( diff --git a/tests/nodes/fetch_node_test.py b/tests/nodes/fetch_node_test.py index 91144daa..f2e69072 100644 --- a/tests/nodes/fetch_node_test.py +++ b/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", diff --git a/tests/test_xquik.py b/tests/test_xquik.py new file mode 100644 index 00000000..cb7f5dba --- /dev/null +++ b/tests/test_xquik.py @@ -0,0 +1,179 @@ +"""Tests for the Xquik document loader.""" + +from __future__ import annotations + +import json +from typing import Any +from unittest.mock import MagicMock, patch + +import pytest + +from scrapegraphai.docloaders.xquik import XquikLoader, extract_tweet_id +from scrapegraphai.graphs.smart_scraper_graph import SmartScraperGraph + + +class _Response: + def __init__(self, payload: dict[str, Any]) -> None: + self.payload = payload + + def model_dump(self, *, by_alias: bool, exclude_none: bool) -> dict[str, Any]: + assert by_alias is True + assert exclude_none is True + return self.payload + + +class _Tweets: + def __init__(self, responses: dict[str, dict[str, Any]]) -> None: + self.responses = responses + self.calls: list[str] = [] + + def retrieve(self, tweet_id: str) -> _Response: + self.calls.append(tweet_id) + return _Response(self.responses[tweet_id]) + + +class _X: + def __init__(self, tweets: _Tweets) -> None: + self.tweets = tweets + + +class _Client: + def __init__(self, responses: dict[str, dict[str, Any]]) -> None: + self.tweets = _Tweets(responses) + self.x = _X(self.tweets) + self.closed = False + + def close(self) -> None: + self.closed = True + + +@pytest.mark.parametrize( + ("source", "expected"), + [ + ("1893456789012345678", "1893456789012345678"), + ( + "https://x.com/example/status/1893456789012345678", + "1893456789012345678", + ), + ( + "https://mobile.twitter.com/example/status/1893456789012345678?s=20", + "1893456789012345678", + ), + ], +) +def test_extract_tweet_id(source: str, expected: str) -> None: + assert extract_tweet_id(source) == expected + + +@pytest.mark.parametrize( + "source", + [ + "https://example.com/status/1893456789012345678", + "https://x.com/example", + "not-a-tweet", + ], +) +def test_extract_tweet_id_rejects_invalid_sources(source: str) -> None: + with pytest.raises(ValueError, match="numeric tweet ID"): + extract_tweet_id(source) + + +def test_loader_returns_complete_tweet_json_and_metadata() -> None: + tweet_id = "1893456789012345678" + payload = { + "tweet": { + "id": tweet_id, + "text": "Scrape structured data from this post.", + "likeCount": 12, + }, + "author": {"username": "example"}, + } + client = _Client({tweet_id: payload}) + + documents = XquikLoader( + [f"https://x.com/example/status/{tweet_id}"], + client=client, + ).load() + + assert len(documents) == 1 + assert json.loads(documents[0].page_content) == payload + assert documents[0].metadata == { + "source": f"https://x.com/example/status/{tweet_id}", + "loader": "xquik", + "tweet_id": tweet_id, + } + assert client.tweets.calls == [tweet_id] + assert client.closed is False + + +def test_loader_fetches_each_source_in_order() -> None: + first = "1893456789012345678" + second = "1893456789012345679" + client = _Client( + { + first: {"tweet": {"id": first, "text": "First"}}, + second: {"tweet": {"id": second, "text": "Second"}}, + } + ) + + documents = XquikLoader([first, second], client=client).load() + + assert [document.metadata["tweet_id"] for document in documents] == [ + first, + second, + ] + assert client.tweets.calls == [first, second] + + +def test_loader_closes_its_client(monkeypatch: pytest.MonkeyPatch) -> None: + tweet_id = "1893456789012345678" + client = _Client({tweet_id: {"tweet": {"id": tweet_id, "text": "Post"}}}) + monkeypatch.setattr( + "scrapegraphai.docloaders.xquik._create_client", + lambda api_key, timeout: client, + ) + + documents = XquikLoader([tweet_id]).load() + + assert len(documents) == 1 + assert client.closed is True + + +def test_smart_scraper_passes_fetch_backend_configuration() -> None: + graph = object.__new__(SmartScraperGraph) + graph.llm_model = object() + graph.config = { + "xquik": {"api_key": "test-key"}, + } + graph.prompt = "Extract the post text." + graph.source = "https://x.com/example/status/1893456789012345678" + graph.schema = None + graph.model_token = 8192 + + with ( + patch("scrapegraphai.graphs.smart_scraper_graph.FetchNode") as fetch_node, + patch("scrapegraphai.graphs.smart_scraper_graph.ParseNode"), + patch("scrapegraphai.graphs.smart_scraper_graph.GenerateAnswerNode"), + patch( + "scrapegraphai.graphs.smart_scraper_graph.BaseGraph", + return_value=MagicMock(), + ), + ): + graph._create_graph() + + node_config = fetch_node.call_args.kwargs["node_config"] + assert node_config["xquik"] == {"api_key": "test-key"} + + +def test_smart_scraper_treats_numeric_tweet_id_as_url_input() -> None: + with patch( + "scrapegraphai.graphs.smart_scraper_graph.AbstractGraph.__init__", + return_value=None, + ): + graph = SmartScraperGraph( + prompt="Extract the post text.", + source="1893456789012345678", + config={"xquik": {"api_key": "test-key"}}, + ) + + assert graph.input_key == "url"