Add TweetClaw content converter

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).
 
-<p align="center">
-  <img src="docs/images/demo.gif" alt="SocialSimBench Demo" width="800"/>
-</p>
-
----
-
 ## 🚀 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

scripts/__init__.py

@@ -0,0 +1 @@
+"""Repository utility scripts."""

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

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