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/api/upload.py b/PyTweetToolkit/api/upload.py index bb965e6..a2959ce 100644 --- a/PyTweetToolkit/api/upload.py +++ b/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): """ 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..b4feb3b 100644 --- a/README.md +++ b/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). diff --git a/tests/test_upload.py b/tests/test_upload.py new file mode 100644 index 0000000..001c973 --- /dev/null +++ b/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" 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")