From bffc86d451a1cfc8cc436f39222cd56e44c0d263 Mon Sep 17 00:00:00 2001 From: kriptoburak Date: Sat, 4 Jul 2026 18:45:25 +0300 Subject: [PATCH] Add Xquik export helpers --- PyTweetToolkit/__init__.py | 3 + PyTweetToolkit/main.py | 12 ++++ PyTweetToolkit/utils/xquik_export.py | 104 +++++++++++++++++++++++++++ README.md | 12 ++++ tests/test_xquik_export.py | 38 ++++++++++ 5 files changed, 169 insertions(+) create mode 100644 PyTweetToolkit/utils/xquik_export.py create mode 100644 tests/test_xquik_export.py diff --git a/PyTweetToolkit/__init__.py b/PyTweetToolkit/__init__.py index 3e762ef..2b44832 100644 --- a/PyTweetToolkit/__init__.py +++ b/PyTweetToolkit/__init__.py @@ -1 +1,4 @@ from .main import PyTweetClient +from .utils.xquik_export import parse_xquik_export + +__all__ = ["PyTweetClient", "parse_xquik_export"] diff --git a/PyTweetToolkit/main.py b/PyTweetToolkit/main.py index 8c83228..ac5beb5 100644 --- a/PyTweetToolkit/main.py +++ b/PyTweetToolkit/main.py @@ -1,4 +1,5 @@ from .api import bookmark, friendship, interaction, notification, profile, restrictions, search, tweet, upload, user +from .utils.xquik_export import parse_xquik_export class PyTweetClient(bookmark.BookmarkActions, @@ -48,3 +49,14 @@ def __init__(self, auth_token: str, csrf_token: str) -> None: csrf_token (str): The CSRF token for authentication. """ super().__init__(auth_token, csrf_token) + + @staticmethod + def load_xquik_export(raw_export: str, filename: str = "tweets.json"): + """ + Parse a Xquik JSON, JSONL, or CSV export without making API requests. + + Args: + raw_export (str): Export contents. + filename (str): Export filename used to choose the parser. + """ + return parse_xquik_export(raw_export, filename) diff --git a/PyTweetToolkit/utils/xquik_export.py b/PyTweetToolkit/utils/xquik_export.py new file mode 100644 index 0000000..a9492bf --- /dev/null +++ b/PyTweetToolkit/utils/xquik_export.py @@ -0,0 +1,104 @@ +import csv +import json +import re +from typing import Any, Dict, List, Optional + + +TEXT_FIELDS = ("text", "tweet", "full_text", "content", "body") +ID_FIELDS = ("id", "tweet_id", "id_str", "post_id") +URL_FIELDS = ("url", "tweet_url", "permalink") + + +def parse_xquik_export(raw_export: str, filename: str = "tweets.json") -> List[Dict[str, str]]: + """Extract tweet IDs, URLs, and text from a Xquik JSON, JSONL, or CSV export.""" + if not raw_export.strip(): + return [] + + lowered_name = filename.lower() + if lowered_name.endswith(".csv"): + records = _parse_csv(raw_export) + elif lowered_name.endswith(".jsonl"): + records = _parse_jsonl(raw_export) + else: + try: + records = _records_from_json(json.loads(raw_export)) + except json.JSONDecodeError as exc: + if lowered_name.endswith(".json"): + raise ValueError("Xquik JSON export contains invalid JSON.") from exc + records = _parse_jsonl(raw_export) + + tweets = [_normalize_record(record) for record in records] + return [tweet for tweet in tweets if tweet] + + +def _parse_csv(raw_export: str) -> List[Dict[str, Any]]: + reader = csv.DictReader(raw_export.splitlines()) + if reader.fieldnames is None: + return [] + if _find_field(reader.fieldnames, TEXT_FIELDS) is None: + raise ValueError("Xquik CSV export needs a text, tweet, full_text, content, or body column.") + return [dict(row) for row in reader] + + +def _parse_jsonl(raw_export: str) -> List[Dict[str, Any]]: + records: List[Dict[str, Any]] = [] + for line in raw_export.splitlines(): + stripped = line.strip() + if not stripped: + continue + try: + parsed = json.loads(stripped) + except json.JSONDecodeError as exc: + raise ValueError("Xquik JSONL export contains an invalid JSON line.") from exc + if isinstance(parsed, dict): + records.append(parsed) + return records + + +def _records_from_json(parsed: Any) -> List[Dict[str, Any]]: + if isinstance(parsed, list): + return [item for item in parsed if isinstance(item, dict)] + if isinstance(parsed, dict): + for key in ("tweets", "items", "data", "results"): + value = parsed.get(key) + if isinstance(value, list): + return [item for item in value if isinstance(item, dict)] + return [parsed] + return [] + + +def _normalize_record(record: Dict[str, Any]) -> Dict[str, str]: + text = _first_text(record, TEXT_FIELDS) + if not text: + return {} + url = _first_text(record, URL_FIELDS) + tweet_id = _first_text(record, ID_FIELDS) or _status_id_from_url(url) + normalized = {"text": text} + if tweet_id: + normalized["tweet_id"] = tweet_id + if url: + normalized["url"] = url + return normalized + + +def _first_text(record: Dict[str, Any], fields: tuple) -> str: + field = _find_field(record.keys(), fields) + if field is None: + return "" + value = record.get(field) + if value is None: + return "" + return " ".join(str(value).split()) + + +def _find_field(fields: Any, candidates: tuple) -> Optional[str]: + normalized_fields = {str(field).lower(): str(field) for field in fields} + for candidate in candidates: + if candidate in normalized_fields: + return normalized_fields[candidate] + return None + + +def _status_id_from_url(url: str) -> str: + match = re.search(r"/status/(\d+)", url) + return match.group(1) if match else "" diff --git a/README.md b/README.md index 8df3e38..476ac9c 100644 --- a/README.md +++ b/README.md @@ -75,6 +75,18 @@ client.post_tweet("Hello, world! #MyFirstTweet") client.follow("python_community") ``` +## Xquik Export Helpers + +PyTweetToolkit can parse Xquik JSON, JSONL, or CSV exports into tweet dictionaries without making API requests: + +```python +from PyTweetToolkit import parse_xquik_export + +tweets = parse_xquik_export(raw_export, "tweets.jsonl") +``` + +The helper accepts common text fields such as `text`, `tweet`, `full_text`, `content`, and `body`, and derives tweet IDs from ID columns or status URLs when present. + ## 📚 Documentation For detailed documentation, including setup guides, examples, and API references, please visit our [documentation page](https://github.com/DavyJonesCodes/PyTweetToolkit/wiki/1.-Home). diff --git a/tests/test_xquik_export.py b/tests/test_xquik_export.py new file mode 100644 index 0000000..6e647e5 --- /dev/null +++ b/tests/test_xquik_export.py @@ -0,0 +1,38 @@ +from PyTweetToolkit.utils.xquik_export import parse_xquik_export + + +def test_parse_xquik_json_export(): + tweets = parse_xquik_export('{"tweets": [{"id": "1", "text": " Hello "}]}') + + assert tweets == [{"text": "Hello", "tweet_id": "1"}] + + +def test_parse_xquik_jsonl_export(): + tweets = parse_xquik_export( + '{"full_text":"First","url":"https://x.com/acme/status/10"}\n' + '{"tweet":"Second","tweet_url":"https://x.com/acme/status/11"}', + "tweets.jsonl", + ) + + assert tweets == [ + {"text": "First", "tweet_id": "10", "url": "https://x.com/acme/status/10"}, + {"text": "Second", "tweet_id": "11", "url": "https://x.com/acme/status/11"}, + ] + + +def test_parse_xquik_csv_export(): + tweets = parse_xquik_export("id,content\n1,Needs review\n2,Works well\n", "tweets.csv") + + assert tweets == [ + {"text": "Needs review", "tweet_id": "1"}, + {"text": "Works well", "tweet_id": "2"}, + ] + + +def test_reject_csv_without_text_column(): + try: + parse_xquik_export("id,url\n1,https://example.com\n", "tweets.csv") + except ValueError as exc: + assert "CSV export needs" in str(exc) + else: + raise AssertionError("expected missing text column to fail")