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/api/upload.py

@@ -1,6 +1,7 @@
 import os
 import time
 import mimetypes
+from urllib.parse import urlparse
 
 from . import auth
 
@@ -61,7 +62,11 @@ def upload(self, source: str, media_category: str = None) -> str:
             ValueError: If the media category is invalid or the file does not meet Twitter's requirements.
         """
 
-        if not os.path.exists(source):
+        is_remote_gif = (
+            source.startswith(('http://', 'https://'))
+            and urlparse(source).path.lower().endswith('.gif')
+        )
+        if not is_remote_gif and not os.path.exists(source):
             raise FileNotFoundError(f"The path '{source}' does not exist.")
 
         # List of valid media categories
@@ -85,7 +90,7 @@ def upload(self, source: str, media_category: str = None) -> str:
         self.media_category = media_category
         self.total_bytes = 0
 
-        if (source.startswith('http://') or source.startswith('https://')) and source.endswith('.gif'):
+        if is_remote_gif:
             self.is_gif = True
             self.media_type = "image/gif"
 
@@ -168,34 +173,33 @@ def _upload_append(self):
 
         segment_id = 0
         bytes_sent = 0
-        file = open(self.source, 'rb')
-
-        while bytes_sent < self.total_bytes:
-            chunk = file.read(4*1024*1024)
-
-            request_data = {
-                'command': 'APPEND',
-                'media_id': self.media_id,
-                'segment_index': segment_id
-            }
-
-            files = {
-                'media': chunk
-            }
-
-            response = self.request_handler.post(
-                self.media_endpoint_url,
-                headers=headers,
-                cookies=cookies,
-                data=request_data,
-                files=files
-            )
-
-            if response.status_code < 200 or response.status_code > 299:
-                raise RuntimeError(f"Error while uploading: HTTP status code {response.status_code} indicates failure.")
-
-            segment_id = segment_id + 1
-            bytes_sent = file.tell()
+        with open(self.source, 'rb') as file:
+            while bytes_sent < self.total_bytes:
+                chunk = file.read(4*1024*1024)
+
+                request_data = {
+                    'command': 'APPEND',
+                    'media_id': self.media_id,
+                    'segment_index': segment_id
+                }
+
+                files = {
+                    'media': chunk
+                }
+
+                response = self.request_handler.post(
+                    self.media_endpoint_url,
+                    headers=headers,
+                    cookies=cookies,
+                    data=request_data,
+                    files=files
+                )
+
+                if response.status_code < 200 or response.status_code > 299:
+                    raise RuntimeError(f"Error while uploading: HTTP status code {response.status_code} indicates failure.")
+
+                segment_id = segment_id + 1
+                bytes_sent = file.tell()
 
     def _upload_finalize(self):
         """

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,22 @@ 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.
+
+Learn more at [xquik.com](https://xquik.com) and the [Xquik API documentation](https://docs.xquik.com).
+
+Xquik is an independent third-party service. Not affiliated with X Corp. "Twitter" and "X" are trademarks of X Corp.
+
 ## 📚 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_upload.py

@@ -0,0 +1,13 @@
+from PyTweetToolkit.api.upload import UploadActions
+
+
+def test_upload_accepts_remote_gif_with_query_string():
+    actions = UploadActions.__new__(UploadActions)
+    actions._upload_init = lambda: setattr(actions, "media_id", "123")
+    actions._check_status = lambda: None
+
+    media_id = actions.upload("https://example.com/animation.GIF?download=1")
+
+    assert media_id == "123"
+    assert actions.is_gif is True
+    assert actions.media_type == "image/gif"

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