diff --git a/README.md b/README.md index a091c98..d061589 100644 --- a/README.md +++ b/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`. diff --git a/tweetclaw_to_project_data.py b/tweetclaw_to_project_data.py new file mode 100644 index 0000000..5eb5e99 --- /dev/null +++ b/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()