diff --git a/README.md b/README.md index 63218d2..b3b2090 100644 --- a/README.md +++ b/README.md @@ -8,12 +8,6 @@ SocialSimBench is a platform for simulating social media dynamics using **LLM-powered agents**. It supports multiple platform modes (Twitter, Reddit, Weibo), multiple network topologies, and a **benchmark task pool** for studying information diffusion, misinformation/rumor correction, stance evolution, collaboration, and polarization. It also supports **dataset-grounded runs** via **Kaggle import** and local file upload, and exports **reproducible episode artifacts** (config + logs + metrics + plots). -

- SocialSimBench Demo -

- ---- - ## 🚀 2–4 Minute Demo Flow (what reviewers/attendees do) 1. **Select data source**: Generated / Kaggle / Upload @@ -229,6 +223,17 @@ Upload CSV files with: - **Network**: `source,target` columns (or `from,to`, `node1,node2`) - **Content**: `text` column (or `content`, `tweet`, `message`) +TweetClaw tweet exports can be converted into content rows for Twitter-mode +simulations: + +```bash +python -m scripts.tweetclaw_export tweetclaw-export.json > datalake/content/tweetclaw_content.json +``` + +The converter accepts TweetClaw JSON, JSONL, NDJSON, and CSV exports with text +fields such as `text`, `full_text`, `tweetText`, `tweet_text`, `content`, or +`body`. + --- ## ⚙️ Configuration diff --git a/scripts/__init__.py b/scripts/__init__.py new file mode 100644 index 0000000..b4c0756 --- /dev/null +++ b/scripts/__init__.py @@ -0,0 +1 @@ +"""Repository utility scripts.""" diff --git a/scripts/tweetclaw_export.py b/scripts/tweetclaw_export.py new file mode 100644 index 0000000..73de0d4 --- /dev/null +++ b/scripts/tweetclaw_export.py @@ -0,0 +1,109 @@ +"""Convert local TweetClaw exports into SocialSimBench content rows.""" + +from __future__ import annotations + +import csv +import json +import sys +from io import StringIO +from pathlib import Path +from typing import Any + + +TEXT_FIELDS = ("text", "full_text", "tweetText", "tweet_text", "content", "body") +USER_FIELDS = ("username", "screen_name", "author", "authorUsername", "user") +TIME_FIELDS = ("created_at", "createdAt", "timestamp", "time") +ID_FIELDS = ("id", "id_str", "tweet_id", "tweetId") +LIST_FIELDS = ("tweets", "items", "data", "results") + + +def convert_tweetclaw_export(file_bytes: bytes, filename: str = "") -> list[dict[str, str]]: + """Return SocialSimBench content rows from a TweetClaw JSON, JSONL, or CSV export.""" + raw_text = file_bytes.decode("utf-8-sig").strip() + if not raw_text: + raise ValueError("TweetClaw export is empty.") + + records = _load_records(raw_text, filename) + rows = [_to_content_row(record) for record in records] + rows = [row for row in rows if row["text"]] + if not rows: + raise ValueError("No tweet text fields were found in the TweetClaw export.") + return rows + + +def _load_records(raw_text: str, filename: str) -> list[dict[str, Any]]: + if Path(filename).suffix.lower() == ".csv": + return [dict(row) for row in csv.DictReader(StringIO(raw_text))] + + try: + payload = json.loads(raw_text) + except json.JSONDecodeError: + return _load_jsonl_records(raw_text) + + if isinstance(payload, list): + return [item for item in payload if isinstance(item, dict)] + + if isinstance(payload, dict): + for field in LIST_FIELDS: + value = payload.get(field) + if isinstance(value, list): + return [item for item in value if isinstance(item, dict)] + return [payload] + + return [] + + +def _load_jsonl_records(raw_text: str) -> list[dict[str, Any]]: + records: list[dict[str, Any]] = [] + for line in raw_text.splitlines(): + if line.strip(): + item = json.loads(line) + if isinstance(item, dict): + records.append(item) + return records + + +def _to_content_row(record: dict[str, Any]) -> dict[str, str]: + text = _first_string(record, TEXT_FIELDS) + row = { + "text": text, + "source": "tweetclaw", + } + + row_id = _first_string(record, ID_FIELDS) + user = _first_string(record, USER_FIELDS) + created_at = _first_string(record, TIME_FIELDS) + + if row_id: + row["id"] = row_id + if user: + row["user"] = user + if created_at: + row["created_at"] = created_at + + return row + + +def _first_string(record: dict[str, Any], fields: tuple[str, ...]) -> str: + for field in fields: + value = record.get(field) + if isinstance(value, str) and value.strip(): + return value.strip() + if value is not None and not isinstance(value, (dict, list)): + return str(value).strip() + return "" + + +def main() -> None: + """Convert one export file to JSON content rows on stdout.""" + if len(sys.argv) != 2: + raise SystemExit("Usage: python -m scripts.tweetclaw_export ") + + path = Path(sys.argv[1]) + rows = convert_tweetclaw_export(path.read_bytes(), path.name) + json.dump(rows, sys.stdout, indent=2) + sys.stdout.write("\n") + + +if __name__ == "__main__": + main() diff --git a/tests/test_tweetclaw_export.py b/tests/test_tweetclaw_export.py new file mode 100644 index 0000000..a94c167 --- /dev/null +++ b/tests/test_tweetclaw_export.py @@ -0,0 +1,60 @@ +import json +import unittest + +from scripts.tweetclaw_export import convert_tweetclaw_export + + +class TweetClawExportTest(unittest.TestCase): + def test_converts_json_array_to_content_rows(self): + payload = b'[{"id":"1","text":"first tweet","username":"alice"},{"id":"2","full_text":"second tweet"}]' + + self.assertEqual( + convert_tweetclaw_export(payload, "tweets.json"), + [ + {"text": "first tweet", "source": "tweetclaw", "id": "1", "user": "alice"}, + {"text": "second tweet", "source": "tweetclaw", "id": "2"}, + ], + ) + + def test_converts_wrapped_export_records(self): + payload = b'{"data":[{"tweetText":"wrapped tweet","createdAt":"2026-06-29"}]}' + + self.assertEqual( + convert_tweetclaw_export(payload, "export.json"), + [{"text": "wrapped tweet", "source": "tweetclaw", "created_at": "2026-06-29"}], + ) + + def test_converts_jsonl_records(self): + payload = b'{"content":"one"}\n{"body":"two"}\n' + + self.assertEqual( + convert_tweetclaw_export(payload, "tweets.jsonl"), + [ + {"text": "one", "source": "tweetclaw"}, + {"text": "two", "source": "tweetclaw"}, + ], + ) + + def test_converts_csv_records(self): + payload = b"id,text,screen_name\n1,hello from csv,bob\n2,second row,\n" + + self.assertEqual( + convert_tweetclaw_export(payload, "tweets.csv"), + [ + {"text": "hello from csv", "source": "tweetclaw", "id": "1", "user": "bob"}, + {"text": "second row", "source": "tweetclaw", "id": "2"}, + ], + ) + + def test_cli_shape_matches_json_content_dataset(self): + rows = convert_tweetclaw_export(b'[{"text":"hello"}]', "tweets.json") + + self.assertEqual(json.loads(json.dumps(rows)), [{"text": "hello", "source": "tweetclaw"}]) + + def test_rejects_exports_without_text(self): + with self.assertRaisesRegex(ValueError, "No tweet text fields"): + convert_tweetclaw_export(b'[{"id":"1"}]', "tweets.json") + + +if __name__ == "__main__": + unittest.main()