Add TweetClaw export converter

README.md

@@ -16,3 +16,18 @@ intelligence (AI) to mine text for sentiment and subjective information.
 1. [Python](https://www.python.org/downloads/)
 2. [Pycharm](https://www.jetbrains.com/pycharm/download/#section=windows)
 3. [Google Sheets](https://www.google.com/sheets/about/)
+
+## TweetClaw Export Data
+You can replace the sample `project_twitter_data.csv` file with tweets exported
+from [TweetClaw](https://github.com/Xquik-dev/tweetclaw):
+
+```bash
+python tweetclaw_to_project_data.py --input exports/tweetclaw.csv --output project_twitter_data.csv
+python FinalProject.py
+```
+
+The converter accepts TweetClaw CSV, JSON, JSONL, or NDJSON exports. It writes
+the same `tweet_text`, `retweet_count`, and `reply_count` columns used by the
+existing sentiment script, removes commas from tweet text so the original CSV
+splitter keeps three columns, and defaults missing retweet or reply counts to
+`0`.

tweetclaw_to_project_data.py

@@ -0,0 +1,122 @@
+import argparse
+import csv
+import json
+from pathlib import Path
+from typing import Any
+
+
+TEXT_COLUMNS = ("tweet_text", "text", "full_text", "content", "tweet")
+RETWEET_COLUMNS = ("retweet_count", "retweets", "retweetCount", "public_metrics.retweet_count")
+REPLY_COLUMNS = ("reply_count", "replies", "replyCount", "public_metrics.reply_count")
+
+
+def read_export_rows(input_path: Path) -> list[dict[str, Any]]:
+    suffix = input_path.suffix.lower()
+    if suffix in {".jsonl", ".ndjson"}:
+        rows: list[dict[str, Any]] = []
+        with input_path.open(encoding="utf-8") as input_file:
+            for line in input_file:
+                stripped = line.strip()
+                if stripped:
+                    value = json.loads(stripped)
+                    if isinstance(value, dict):
+                        rows.append(value)
+        return rows
+
+    if suffix == ".json":
+        with input_path.open(encoding="utf-8") as input_file:
+            value = json.load(input_file)
+        if isinstance(value, list):
+            return [row for row in value if isinstance(row, dict)]
+        if isinstance(value, dict):
+            for key in ("data", "results", "tweets", "items"):
+                nested = value.get(key)
+                if isinstance(nested, list):
+                    return [row for row in nested if isinstance(row, dict)]
+            return [value]
+        return []
+
+    with input_path.open(newline="", encoding="utf-8-sig") as input_file:
+        return list(csv.DictReader(input_file))
+
+
+def nested_value(row: dict[str, Any], key: str) -> Any:
+    value: Any = row
+    for part in key.split("."):
+        if not isinstance(value, dict) or part not in value:
+            return None
+        value = value[part]
+    return value
+
+
+def pick_value(row: dict[str, Any], candidates: tuple[str, ...], default: Any = "") -> Any:
+    normalized = {str(key).lower(): value for key, value in row.items()}
+    for candidate in candidates:
+        value = nested_value(row, candidate)
+        if value not in (None, ""):
+            return value
+        value = normalized.get(candidate.lower())
+        if value not in (None, ""):
+            return value
+    return default
+
+
+def clean_tweet_text(value: Any) -> str:
+    text = str(value).replace(",", " ")
+    return " ".join(text.split())
+
+
+def parse_count(value: Any) -> int:
+    if value in (None, ""):
+        return 0
+    try:
+        return max(0, int(float(str(value).replace(",", "").strip())))
+    except ValueError:
+        return 0
+
+
+def convert_export(input_path: Path, output_path: Path) -> int:
+    converted_rows: list[tuple[str, int, int]] = []
+    for row in read_export_rows(input_path):
+        tweet_text = clean_tweet_text(pick_value(row, TEXT_COLUMNS))
+        if not tweet_text:
+            continue
+        converted_rows.append(
+            (
+                tweet_text,
+                parse_count(pick_value(row, RETWEET_COLUMNS, 0)),
+                parse_count(pick_value(row, REPLY_COLUMNS, 0)),
+            )
+        )
+
+    if not converted_rows:
+        raise ValueError("No tweet text rows found in the TweetClaw export.")
+
+    with output_path.open("w", newline="", encoding="utf-8") as output_file:
+        writer = csv.writer(output_file)
+        writer.writerow(("tweet_text", "retweet_count", "reply_count"))
+        writer.writerows(converted_rows)
+    return len(converted_rows)
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(
+        description="Convert a TweetClaw CSV, JSON, JSONL, or NDJSON export to project_twitter_data.csv format."
+    )
+    parser.add_argument("--input", required=True, type=Path, help="TweetClaw export file.")
+    parser.add_argument(
+        "--output",
+        default=Path("project_twitter_data.csv"),
+        type=Path,
+        help="Output CSV path for the existing sentiment script.",
+    )
+    args = parser.parse_args()
+    try:
+        row_count = convert_export(args.input, args.output)
+    except ValueError as exc:
+        raise SystemExit(str(exc)) from exc
+    print(f"Converted {row_count} rows to {args.output}")
+
+
+if __name__ == "__main__":
+    main()