Xの全履歴アーカイブを自分で集計する手順（スクリプト付き）

window.YTD.tweets.part0 = [
  {
    "tweet" : {
      "edit_info" : { ... },
      "id_str" : "...",
      "created_at" : "Wed Sep 09 12:00:00 +0000 2026",
      "full_text" : "...",
      "favorite_count" : "0",
      "retweet_count" : "0",

#!/usr/bin/env python3
"""X の「データのアーカイブ」から自分の投稿を年別・月別に数える。

使い方:
    python3 x_archive_stats.py twitter-archive.zip --by year
    python3 x_archive_stats.py ./展開したフォルダ --by month --out stats.json

標準ライブラリだけで動く（Python 3.9 以降）。読むのは data/tweets.js と
tweets-part1.js などの続きのファイルだけで、DM や like.js には触らない。
"""
import argparse
import json
import os
import re
import sys
import zipfile
from collections import Counter, defaultdict
from datetime import datetime, timedelta, timezone

JST = timezone(timedelta(hours=9))
TWEETS_FILE = re.compile(r"(^|/)tweets(-part\d+)?\.js$")
PREFIX = re.compile(r"^\s*window\.YTD\.\w+\.part\d+\s*=\s*")
WORD = re.compile(r"[ァ-ヴー]{3,}|[一-龥]{2,}")  # カタカナ3字以上か漢字2字以上の連続
STOPWORDS = set()  # 上位に出てきて邪魔な語があれば、ここに足す


def read_tweet_files(path):
    """ZIP か展開済みフォルダから、tweets.js と tweets-partN.js の中身を返す。"""
    if os.path.isdir(path):
        for root, _dirs, files in os.walk(path):
            for name in sorted(files):
                full = os.path.join(root, name)
                if TWEETS_FILE.search(full.replace(os.sep, "/")):
                    with open(full, encoding="utf-8-sig") as f:
                        yield name, f.read()
    else:
        with zipfile.ZipFile(path) as z:
            for name in sorted(z.namelist()):
                if TWEETS_FILE.search(name):
                    yield name, z.read(name).decode("utf-8-sig")


def edit_ids(node):
    """edit_info の中にある editTweetIds（同じ投稿の版の ID 一覧）を探す。"""
    if isinstance(node, dict):
        if "editTweetIds" in node:
            return node["editTweetIds"]
        for value in node.values():
            found = edit_ids(value)
            if found:
                return found
    return []


def load(path):
    """全ファイルを読み、ID の重複と編集前の版を除いた投稿のリストを返す。"""
    by_id = {}
    for name, text in read_tweet_files(path):
        body = PREFIX.sub("", text, count=1).strip().rstrip(";")
        for item in json.loads(body):
            tweet = item.get("tweet", item)
            if tweet.get("id_str"):
                by_id[tweet["id_str"]] = tweet
    if not by_id:
        sys.exit("tweets.js が見つからない: %s" % path)
    latest = {}  # 編集した投稿は版ごとに入っているので、最後の版だけ残す
    for tid, tweet in by_id.items():
        group = min(edit_ids(tweet.get("edit_info")) or [tid], key=int)
        if group not in latest or int(tid) > int(latest[group]["id_str"]):
            latest[group] = tweet
    return list(latest.values())


def normalize(tweet):
    """集計に使う項目だけ取り出す。日時は日本時間にする。"""
    try:
        dt = datetime.strptime(tweet["created_at"], "%a %b %d %H:%M:%S %z %Y")
    except (KeyError, ValueError):
        return None
    dt = dt.astimezone(JST)
    if dt.year < 2006:  # サービス開始より前の日付は壊れたデータとして捨てる
        return None
    text = tweet.get("full_text") or tweet.get("text") or ""
    return {
        "dt": dt,
        "text": text,
        "is_rt": text.startswith("RT @"),
        "is_reply": bool(tweet.get("in_reply_to_status_id_str")),
        "fav": int(tweet.get("favorite_count") or 0),
        "rt": int(tweet.get("retweet_count") or 0),
    }


def pct(part, whole):
    return round(part * 100.0 / whole, 1) if whole else 0.0


def summarize(rows):
    own = [r for r in rows if not r["is_rt"]]  # 本人の投稿＝リツイート以外すべて
    words = Counter()
    for r in own:
        for w in WORD.findall(re.sub(r"https?://\S+", "", r["text"])):
            if w not in STOPWORDS:
                words[w] += 1
    top = sorted(own, key=lambda r: (r["fav"], r["rt"]), reverse=True)[:5]
    return {
        "total": len(rows),
        "own": len(own),
        "retweet": len(rows) - len(own),
        "reply": sum(1 for r in own if r["is_reply"]),  # 本人の投稿の内数
        "retweet_ratio": pct(len(rows) - len(own), len(rows)),
        "zero_fav_ratio": pct(sum(1 for r in own if r["fav"] == 0), len(own)),
        "fav_total": sum(r["fav"] for r in own),
        "by_month": dict(sorted(Counter(r["dt"].strftime("%Y-%m") for r in rows).items())),
        "by_hour": dict((h, sum(1 for r in rows if r["dt"].hour == h)) for h in range(24)),
        "top_words": words.most_common(20),
        "top_posts": [{"date": r["dt"].strftime("%Y-%m-%d %H:%M"), "fav": r["fav"],
                       "rt": r["rt"], "text": r["text"][:80]} for r in top],
    }


def main():
    ap = argparse.ArgumentParser(description="X のアーカイブを年別・月別に集計する")
    ap.add_argument("archive", help="アーカイブの ZIP か、展開したフォルダ")
    ap.add_argument("--by", choices=["year", "month"], default="year")
    ap.add_argument("--out", help="集計結果を JSON で保存する先")
    args = ap.parse_args()

    rows = [r for r in map(normalize, load(args.archive)) if r]
    groups = defaultdict(list)
    for r in rows:
        groups[r["dt"].strftime("%Y" if args.by == "year" else "%Y-%m")].append(r)
    result = dict((key, summarize(groups[key])) for key in sorted(groups))

    print("期間      全体   本人   RT   返信  RT率  反応ゼロ率")
    for key, s in result.items():
        print("%-8s %6d %6d %5d %5d %5.1f%% %6.1f%%" % (
            key, s["total"], s["own"], s["retweet"], s["reply"],
            s["retweet_ratio"], s["zero_fav_ratio"]))
    if args.out:
        with open(args.out, "w", encoding="utf-8") as f:
            json.dump(result, f, ensure_ascii=False, indent=1)


if __name__ == "__main__":
    main()