Add Xquik export helpers

PyTweetToolkit/__init__.py

@@ -1 +1,4 @@
 from .main import PyTweetClient
+from .utils.xquik_export import parse_xquik_export
+
+__all__ = ["PyTweetClient", "parse_xquik_export"]

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)

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

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

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