#!/usr/bin/env python3
"""人狼戦績 バッチ集計プログラム（サーバB: 公開API用）

毎日 AM4:00 に実行。ランキングキャッシュと個人戦績キャッシュを事前計算。
初回は全件、以降はウォーターマーク方式で差分処理。

配役分類: 通常 / カオス / 全体 の3カテゴリで戦績を集計。

使い方:
  python batch.py                # 差分
  python batch.py --full         # 強制フルリビルド
  python batch.py --ranks-only   # ランキングのみ
  python batch.py --personal-only # 個人のみ
"""

from __future__ import annotations

import argparse
import asyncio
import json
import logging
import os
import sys
import time
from collections import defaultdict
from datetime import datetime, timezone, timedelta

import asyncpg
from pydantic_settings import BaseSettings

JST = timezone(timedelta(hours=9))

class Settings(BaseSettings):
    database_url: str = "postgresql://zinro:password@localhost:5432/zinro_meta"
    batch_chunk_size: int = 500
    model_config = {"env_file": ".env", "env_file_encoding": "utf-8"}

logging.basicConfig(level=logging.INFO, format="%(asctime)s [BATCH] %(levelname)-8s %(message)s", stream=sys.stdout)
log = logging.getLogger("batch")

# ── 時刻ヘルパ ──
def ts_to_jst_date(ts): return datetime.fromtimestamp(ts, tz=JST).strftime("%Y-%m-%d")
def ts_to_jst_ym(ts): return datetime.fromtimestamp(ts, tz=JST).strftime("%Y-%m")
def ts_to_jst_hour(ts): return datetime.fromtimestamp(ts, tz=JST).hour
def ts_to_jst_weekday(ts): return (datetime.fromtimestamp(ts, tz=JST).weekday() + 1) % 7

# ╔═════════════════════════════════════╗
# ║  DB                                  ║
# ╚═════════════════════════════════════╝

async def create_pool() -> asyncpg.Pool:
    s = Settings()
    return await asyncpg.create_pool(dsn=s.database_url, min_size=2, max_size=5,
        statement_cache_size=128, server_settings={"timezone": "UTC", "statement_timeout": "0"})

BATCH_DDL = """
CREATE TABLE IF NOT EXISTS batch_state (key TEXT PRIMARY KEY, value_int BIGINT NOT NULL DEFAULT 0, value_text TEXT NOT NULL DEFAULT '', updated_at BIGINT NOT NULL DEFAULT 0);
CREATE TABLE IF NOT EXISTS personal_stats_cache (key_type TEXT NOT NULL, key_value TEXT NOT NULL, watermark_game_id BIGINT NOT NULL DEFAULT 0, stats_json TEXT NOT NULL DEFAULT '{}', game_ids_json TEXT NOT NULL DEFAULT '[]', partners_json TEXT NOT NULL DEFAULT '{}', first_ts BIGINT DEFAULT NULL, last_ts BIGINT DEFAULT NULL, updated_at BIGINT NOT NULL DEFAULT 0, PRIMARY KEY (key_type, key_value));
CREATE TABLE IF NOT EXISTS shared_trips (trip TEXT PRIMARY KEY, name_count INT NOT NULL DEFAULT 0, updated_at BIGINT NOT NULL DEFAULT 0);
ALTER TABLE personal_stats_cache ADD COLUMN IF NOT EXISTS partners_json TEXT NOT NULL DEFAULT '{}';
CREATE TABLE IF NOT EXISTS personal_search_priority (key_type TEXT NOT NULL, key_value TEXT NOT NULL, search_count INTEGER NOT NULL DEFAULT 0, last_searched_at BIGINT NOT NULL DEFAULT 0, last_cache_hit SMALLINT NOT NULL DEFAULT 0, PRIMARY KEY (key_type, key_value));
CREATE INDEX IF NOT EXISTS idx_personal_search_priority_hot ON personal_search_priority (last_searched_at DESC, search_count DESC);
"""

async def ensure_tables(pool):
    async with pool.acquire() as c:
        await c.execute(BATCH_DDL)
    log.info("バッチ用テーブル確認完了")

# ── ウォーターマーク ──
WM_KEY = "last_processed_game_id"
async def get_watermark(pool):
    async with pool.acquire() as c:
        r = await c.fetchrow("SELECT value_int FROM batch_state WHERE key=$1", WM_KEY)
    return int(r["value_int"]) if r else 0

async def set_watermark(pool, gid):
    now = int(time.time())
    async with pool.acquire() as c:
        await c.execute("INSERT INTO batch_state (key,value_int,updated_at) VALUES($1,$2,$3) ON CONFLICT(key) DO UPDATE SET value_int=EXCLUDED.value_int, updated_at=EXCLUDED.updated_at", WM_KEY, gid, now)

async def get_max_game_id(pool):
    async with pool.acquire() as c:
        return int(await c.fetchval("SELECT COALESCE(MAX(id),0) FROM games"))

# ╔═════════════════════════════════════╗
# ║  ランキングキャッシュ                  ║
# ╚═════════════════════════════════════╝

async def get_all_months(pool):
    async with pool.acquire() as c:
        rows = await c.fetch("SELECT DISTINCT TO_CHAR(TO_TIMESTAMP(started_at) AT TIME ZONE 'Asia/Tokyo','YYYY-MM') AS ym FROM games ORDER BY ym")
    return [r["ym"] for r in rows]

async def get_all_years(pool):
    async with pool.acquire() as c:
        rows = await c.fetch("SELECT DISTINCT TO_CHAR(TO_TIMESTAMP(started_at) AT TIME ZONE 'Asia/Tokyo','YYYY') AS yr FROM games ORDER BY yr")
    return [r["yr"] for r in rows]

async def get_affected_periods(pool, wm):
    async with pool.acquire() as c:
        rows = await c.fetch("SELECT DISTINCT TO_CHAR(TO_TIMESTAMP(started_at) AT TIME ZONE 'Asia/Tokyo','YYYY-MM') AS ym, TO_CHAR(TO_TIMESTAMP(started_at) AT TIME ZONE 'Asia/Tokyo','YYYY') AS yr FROM games WHERE id>$1", wm)
    return {r["ym"] for r in rows}, {r["yr"] for r in rows}

# ── ランク計算クエリ群 ──

async def _rq_games(c, kt, ymf, pb):
    col = "name" if kt == "name" else "trip"
    rows = await c.fetch(f"SELECT p.{col} AS k,COUNT(*) AS games,SUM(p.win) AS wins,ROUND(SUM(p.win)::numeric/COUNT(*)*100,1) AS wr FROM players p JOIN games g ON g.id=p.game_id WHERE p.is_spectator=0 {ymf} AND p.{col} IS NOT NULL AND p.{col}<>'' GROUP BY p.{col} ORDER BY games DESC", *pb)
    return [{col: r["k"], "games": int(r["games"]), "wins": int(r["wins"]), "win_rate": float(r["wr"])} for r in rows]

async def _rq_win_rate(c, kt, ymf, pb, mg=30):
    col = "name" if kt == "name" else "trip"
    n = len(pb) + 1
    rows = await c.fetch(f"SELECT p.{col} AS k,COUNT(*) AS games,SUM(p.win) AS wins,ROUND(SUM(p.win)::numeric/COUNT(*)*100,1) AS wr FROM players p JOIN games g ON g.id=p.game_id WHERE p.is_spectator=0 {ymf} AND p.{col} IS NOT NULL AND p.{col}<>'' GROUP BY p.{col} HAVING COUNT(*)>=${n} ORDER BY wr DESC,games DESC", *pb, mg)
    return [{col: r["k"], "games": int(r["games"]), "wins": int(r["wins"]), "win_rate": float(r["wr"])} for r in rows]

async def _rq_talk(c, ymf, pb, mld=10):
    n = len(pb) + 1
    rows = await c.fetch(f"SELECT p.name AS k,SUM(p.talk_count) AS ts,SUM(p.live_days) AS ls,ROUND(SUM(p.talk_count)::numeric/NULLIF(SUM(p.live_days),0),2) AS d FROM players p JOIN games g ON g.id=p.game_id WHERE p.is_spectator=0 {ymf} AND p.name IS NOT NULL AND p.name<>'' GROUP BY p.name HAVING SUM(p.live_days)>=${n} ORDER BY d DESC", *pb, mld)
    return [{"name": r["k"], "talk_sum": int(r["ts"]), "live_sum": int(r["ls"]), "density": float(r["d"] or 0)} for r in rows]

async def _rq_days(c, ymf, pb):
    rows = await c.fetch(f"SELECT p.name AS k,COUNT(DISTINCT TO_CHAR(TO_TIMESTAMP(g.started_at) AT TIME ZONE 'Asia/Tokyo','YYYY-MM-DD')) AS ud,COUNT(*) AS games FROM players p JOIN games g ON g.id=p.game_id WHERE p.is_spectator=0 {ymf} AND p.name IS NOT NULL AND p.name<>'' GROUP BY p.name ORDER BY ud DESC,games DESC", *pb)
    return [{"name": r["k"], "unique_days": int(r["ud"]), "games": int(r["games"])} for r in rows]

async def _rq_role_win(c, ymf, pb, mg=30):
    n = len(pb) + 1
    rows = await c.fetch(f"SELECT p.role,COUNT(*) AS games,SUM(p.win) AS wins,ROUND(SUM(p.win)::numeric/COUNT(*)*100,1) AS wr FROM players p JOIN games g ON g.id=p.game_id WHERE p.is_spectator=0 {ymf} GROUP BY p.role HAVING COUNT(*)>=${n} ORDER BY wr DESC", *pb, mg)
    return [{"role": r["role"], "games": int(r["games"]), "wins": int(r["wins"]), "win_rate": float(r["wr"])} for r in rows]

async def _rq_haiyaku_win(c, ymf, pb, mp=100):
    n = len(pb) + 1
    rows = await c.fetch(f"SELECT g.haiyaku,COUNT(*) AS pl,SUM(p.win) AS wins,ROUND(SUM(p.win)::numeric/COUNT(*)*100,1) AS wr FROM players p JOIN games g ON g.id=p.game_id WHERE p.is_spectator=0 {ymf} GROUP BY g.haiyaku HAVING COUNT(*)>=${n} ORDER BY wr DESC", *pb, mp)
    return [{"haiyaku": r["haiyaku"], "players": int(r["pl"]), "wins": int(r["wins"]), "win_rate": float(r["wr"])} for r in rows]

async def _rq_village(c, ymfg, pb):
    rows = await c.fetch(f"SELECT g.village_name,COUNT(*) AS games FROM games g WHERE 1=1 {ymfg} GROUP BY g.village_name ORDER BY games DESC", *pb)
    return [{"village_name": r["village_name"], "games": int(r["games"])} for r in rows]

async def _rq_haiyaku(c, ymfg, pb):
    rows = await c.fetch(f"SELECT g.haiyaku,COUNT(*) AS games FROM games g WHERE 1=1 {ymfg} GROUP BY g.haiyaku ORDER BY games DESC", *pb)
    return [{"haiyaku": r["haiyaku"], "games": int(r["games"])} for r in rows]

async def _rq_rm(c, ymfg, pb):
    rows = await c.fetch(f"SELECT COALESCE(g.rm_name,'不明') AS rn,COUNT(*) AS games FROM games g WHERE 1=1 {ymfg} GROUP BY rn ORDER BY games DESC", *pb)
    return [{"rm_name": r["rn"], "games": int(r["games"])} for r in rows]

async def _rq_tsuiho(c, ymf, pb):
    # 参加者フラグ(tsuiho_flag)ではなくゲームごとの追放記録(tsuiho_json)から集計する。
    # ロビーで追放された荒らしは参加者名簿に載らずフラグが立たないため。
    rows = await c.fetch(f"SELECT k AS name,COUNT(*) AS times FROM games g CROSS JOIN LATERAL jsonb_object_keys(g.tsuiho_json::jsonb) k WHERE g.tsuiho_json IS NOT NULL AND g.tsuiho_json LIKE '{{%' {ymf} AND k<>'' GROUP BY k ORDER BY times DESC", *pb)
    return [{"name": r["name"], "times": int(r["times"])} for r in rows]


async def _rq_career(c, kt, ymf, pb):
    col = "trip" if kt == "trip" else "name"
    rows = await c.fetch(f"SELECT p.{col} AS k,MIN(g.started_at) AS mn,MAX(g.started_at) AS mx,COUNT(*) AS games FROM players p JOIN games g ON g.id=p.game_id WHERE p.is_spectator=0 {ymf} AND p.{col} IS NOT NULL AND p.{col}<>'' GROUP BY p.{col}", *pb)
    jst = timezone(timedelta(hours=9))
    out = []
    for r in rows:
        fd = datetime.fromtimestamp(int(r["mn"]), jst).date()
        ld = datetime.fromtimestamp(int(r["mx"]), jst).date()
        out.append({col: r["k"], "first_day": fd.isoformat(), "last_day": ld.isoformat(), "span_days": (ld - fd).days + 1, "games": int(r["games"])})
    out.sort(key=lambda x: (-x["span_days"], -x["games"]))
    return out


async def compute_rank(pool, kind, key_type, scope, ym):
    if scope == "all":
        ymf = ymfg = ""; pb = []
    else:
        fmt = "YYYY-MM" if scope == "month" else "YYYY"
        ymf = f"AND TO_CHAR(TO_TIMESTAMP(g.started_at) AT TIME ZONE 'Asia/Tokyo','{fmt}')=$1"
        ymfg = ymf; pb = [ym]
    async with pool.acquire() as c:
        if kind == "games" and key_type in ("name","trip"): return await _rq_games(c, key_type, ymf, pb)
        elif kind == "win_rate" and key_type in ("name","trip"): return await _rq_win_rate(c, key_type, ymf, pb)
        elif kind == "talk" and key_type == "name": return await _rq_talk(c, ymf, pb)
        elif kind == "days" and key_type == "name": return await _rq_days(c, ymf, pb)
        elif kind == "role_win": return await _rq_role_win(c, ymf, pb)
        elif kind == "haiyaku_win": return await _rq_haiyaku_win(c, ymf, pb)
        elif kind == "village": return await _rq_village(c, ymfg, pb)
        elif kind == "haiyaku": return await _rq_haiyaku(c, ymfg, pb)
        elif kind == "rm": return await _rq_rm(c, ymfg, pb)
        elif kind == "tsuiho": return await _rq_tsuiho(c, ymf, pb)
        elif kind == "career" and key_type in ("name", "trip"): return await _rq_career(c, key_type, ymf, pb)
    return []


RANK_CACHE_MAX_ROWS = 10000  # 巨大ペイロードはJSONパースだけで数十秒かかるため上位のみ保存する

async def upsert_rank_cache(pool, scope, ym, kind, key_type, rows):
    payload = json.dumps({"rows": rows[:RANK_CACHE_MAX_ROWS], "total": len(rows)}, ensure_ascii=False)
    now = int(time.time())
    cache_key = f"{scope}:{ym}:{kind}:{key_type or '__NULL__'}"
    async with pool.acquire() as c:
        await c.execute("""
            INSERT INTO rank_cache_json (cache_key,scope,ym,kind,key_type,payload,total,built_at,updated_at)
            VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$8)
            ON CONFLICT (scope,ym,kind,COALESCE(key_type,'__NULL__'))
            DO UPDATE SET cache_key=EXCLUDED.cache_key, payload=EXCLUDED.payload, total=EXCLUDED.total, built_at=EXCLUDED.built_at, updated_at=EXCLUDED.updated_at
        """, cache_key, scope, ym, kind, key_type, payload, len(rows), now)


RANK_KINDS = [
    ("games","name"),("games","trip"),("win_rate","name"),("win_rate","trip"),
    ("talk","name"),("days","name"),("role_win",None),("haiyaku_win",None),
    ("village",None),("haiyaku",None),("rm",None),("tsuiho",None),
    ("career","name"),("career","trip"),
]

async def rebuild_ranks(pool, watermark, force_full):
    t0 = time.time()
    all_months = await get_all_months(pool)
    all_years = await get_all_years(pool)
    if force_full or watermark == 0:
        target_months = set(all_months); target_years = set(all_years)
        log.info("ランキング: フルリビルド (%d月, %d年)", len(target_months), len(target_years))
    else:
        target_months, target_years = await get_affected_periods(pool, watermark)
        log.info("ランキング: 差分更新 (影響月=%d, 影響年=%d)", len(target_months), len(target_years))

    sem = asyncio.Semaphore(3)
    count = 0

    async def _proc(kind, kt, scope, ym):
        nonlocal count
        async with sem:
            rows = await compute_rank(pool, kind, kt, scope, ym)
            await upsert_rank_cache(pool, scope, ym, kind, kt, rows)
            count += 1

    tasks = []
    for kind, kt in RANK_KINDS: tasks.append(_proc(kind, kt, "all", "all"))
    for ym in target_months:
        for kind, kt in RANK_KINDS: tasks.append(_proc(kind, kt, "month", ym))
    for yr in target_years:
        for kind, kt in RANK_KINDS: tasks.append(_proc(kind, kt, "year", yr))

    log.info("ランキング: %d件計算中...", len(tasks))
    for i in range(0, len(tasks), 10):
        await asyncio.gather(*tasks[i:i+10])
        if (i+10) % 100 == 0: log.info("  ... %d/%d", min(i+10, len(tasks)), len(tasks))

    log.info("ランキング: %d件完了 (%.1f秒)", count, time.time()-t0)

    # ── 配役分類別統計キャッシュ ──
    log.info("配役分類別統計キャッシュ構築中...")
    await _build_class_stats_cache(pool, target_months, target_years, force_full or watermark == 0)


async def _build_class_stats_cache(pool, target_months, target_years, is_full):
    """stats_by_class 用キャッシュを rank_cache_json に保存"""
    periods = [("all", "all", "")]
    if is_full:
        async with pool.acquire() as c:
            ms = await c.fetch("SELECT DISTINCT TO_CHAR(TO_TIMESTAMP(started_at) AT TIME ZONE 'Asia/Tokyo','YYYY-MM') AS ym FROM games")
            ys = await c.fetch("SELECT DISTINCT TO_CHAR(TO_TIMESTAMP(started_at) AT TIME ZONE 'Asia/Tokyo','YYYY') AS yr FROM games")
        for r in ms: periods.append(("month", r["ym"], f"AND TO_CHAR(TO_TIMESTAMP(g.started_at) AT TIME ZONE 'Asia/Tokyo','YYYY-MM')=$1"))
        for r in ys: periods.append(("year", r["yr"], f"AND TO_CHAR(TO_TIMESTAMP(g.started_at) AT TIME ZONE 'Asia/Tokyo','YYYY')=$1"))
    else:
        for ym in target_months: periods.append(("month", ym, f"AND TO_CHAR(TO_TIMESTAMP(g.started_at) AT TIME ZONE 'Asia/Tokyo','YYYY-MM')=$1"))
        for yr in target_years: periods.append(("year", yr, f"AND TO_CHAR(TO_TIMESTAMP(g.started_at) AT TIME ZONE 'Asia/Tokyo','YYYY')=$1"))

    now = int(time.time())
    for scope, ym, ymf in periods:
        pb = [] if scope == "all" else [ym]
        async with pool.acquire() as c:
            rows = await c.fetch(f"""
                SELECT COALESCE(g.haiyaku_class,'通常') AS cls, COUNT(*) AS games,
                       SUM(p.win) AS wins, ROUND(SUM(p.win)::numeric/NULLIF(COUNT(*),0)*100,1) AS wr
                FROM players p JOIN games g ON g.id=p.game_id
                WHERE p.is_spectator=0 {ymf} GROUP BY cls
            """, *pb)

        result = {"all":{"games":0,"wins":0,"win_rate":0.0},"normal":{"games":0,"wins":0,"win_rate":0.0},"chaos":{"games":0,"wins":0,"win_rate":0.0}}
        tg = tw = 0
        for r in rows:
            g=int(r["games"]); w=int(r["wins"]); wr=float(r["wr"] or 0)
            tg+=g; tw+=w
            if r["cls"]=="通常": result["normal"]={"games":g,"wins":w,"win_rate":wr}
            else: result["chaos"]["games"]+=g; result["chaos"]["wins"]+=w
        if result["chaos"]["games"]>0:
            result["chaos"]["win_rate"]=round(result["chaos"]["wins"]/result["chaos"]["games"]*100,1)
        result["all"]={"games":tg,"wins":tw,"win_rate":round(tw/tg*100,1) if tg else 0.0}
        result["ym"]=ym

        cache_key = f"class_stats:{ym}"
        payload = json.dumps(result, ensure_ascii=False)
        async with pool.acquire() as c:
            await c.execute("""
                INSERT INTO rank_cache_json (cache_key,scope,ym,kind,key_type,payload,total,built_at,updated_at)
                VALUES ($1,$2,$3,'class_stats',NULL,$4,0,$5,$5)
                ON CONFLICT (cache_key) DO UPDATE SET payload=EXCLUDED.payload, built_at=EXCLUDED.built_at, updated_at=EXCLUDED.updated_at
            """, cache_key, scope, ym, payload, now)

    log.info("配役分類別統計: %d件完了", len(periods))


# ╔═════════════════════════════════════╗
# ║  個人戦績キャッシュ（3分類対応）       ║
# ╚═════════════════════════════════════╝

WEEKDAY_LABELS = ["日","月","火","水","木","金","土"]

def _empty_bucket():
    return {"total_games":0,"wins":0,"talk_sum":0,"live_days_sum":0,
        "haiyaku":{},"role":{},"village":{},
        "by_hour":[[0,0] for _ in range(24)],"by_weekday":[[0,0] for _ in range(7)],
        "by_month":{},"calendar":{"start":None,"end":None,"days":{}}}

def _add_row_to_bucket(b, r):
    w=int(r["win"]); ts=int(r["started_at"]); tk=int(r["talk_count"]); lv=int(r["live_days"])
    b["total_games"]+=1; b["wins"]+=w; b["talk_sum"]+=tk; b["live_days_sum"]+=lv
    d=ts_to_jst_date(ts)
    if d not in b["calendar"]["days"]: b["calendar"]["days"][d]={"g":0,"w":0,"l":0}
    b["calendar"]["days"][d]["g"]+=1
    if w==1: b["calendar"]["days"][d]["w"]+=1
    elif w==0: b["calendar"]["days"][d]["l"]+=1
    cs=b["calendar"]["start"]; ce=b["calendar"]["end"]
    if cs is None or d<cs: b["calendar"]["start"]=d
    if ce is None or d>ce: b["calendar"]["end"]=d
    hk=str(r["haiyaku"])
    if hk not in b["haiyaku"]: b["haiyaku"][hk]={"g":0,"w":0}
    b["haiyaku"][hk]["g"]+=1; b["haiyaku"][hk]["w"]+=w
    rk=str(r["role"])
    if rk not in b["role"]: b["role"][rk]={"g":0,"w":0}
    b["role"][rk]["g"]+=1; b["role"][rk]["w"]+=w
    vk=str(r["village_name"])
    b["village"][vk]=b["village"].get(vk,0)+1
    h=ts_to_jst_hour(ts); wd=ts_to_jst_weekday(ts); ym=ts_to_jst_ym(ts)
    b["by_hour"][h][0]+=1; b["by_hour"][h][1]+=w
    b["by_weekday"][wd][0]+=1; b["by_weekday"][wd][1]+=w
    if ym not in b["by_month"]: b["by_month"][ym]=[0,0]
    b["by_month"][ym][0]+=1; b["by_month"][ym][1]+=w

def aggregate_rows_3class(rows):
    """全体/通常/カオス の3分類で集計"""
    buckets = {"all": _empty_bucket(), "normal": _empty_bucket(), "chaos": _empty_bucket()}
    gids = set()
    for r in rows:
        _add_row_to_bucket(buckets["all"], r)
        hc = r.get("haiyaku_class", "通常")
        if hc == "通常": _add_row_to_bucket(buckets["normal"], r)
        else: _add_row_to_bucket(buckets["chaos"], r)
        gids.add(int(r["game_id"]))
    return {"buckets": buckets, "game_ids": sorted(gids)}


async def get_affected_players(pool, wm):
    async with pool.acquire() as c:
        rows = await c.fetch("SELECT DISTINCT trip,name FROM players WHERE game_id IN (SELECT id FROM games WHERE id>$1) AND is_spectator=0", wm)
    trips=set(); names=set()
    for r in rows:
        if r["trip"] and r["trip"].strip(): trips.add(r["trip"])
        if r["name"] and r["name"].strip(): names.add(r["name"])
    return trips, names

async def get_recent_month_players(pool):
    since = int(time.time()) - 31 * 86400
    async with pool.acquire() as c:
        rows = await c.fetch("""
            SELECT DISTINCT p.trip, p.name
            FROM players p JOIN games g ON g.id=p.game_id
            WHERE p.is_spectator=0 AND g.started_at >= $1
        """, since)
    trips=set(); names=set()
    for r in rows:
        if r["trip"] and r["trip"].strip(): trips.add(r["trip"])
        if r["name"] and r["name"].strip(): names.add(r["name"])
    return trips, names

async def get_priority_players(pool):
    async with pool.acquire() as c:
        rows = await c.fetch("""
            SELECT key_type, key_value
            FROM personal_search_priority
            WHERE key_value IS NOT NULL AND key_value <> ''
            ORDER BY last_searched_at DESC, search_count DESC
            LIMIT 20000
        """)
    trips=set(); names=set()
    for r in rows:
        if r["key_type"] == "trip": trips.add(r["key_value"])
        elif r["key_type"] == "name": names.add(r["key_value"])
    return trips, names

async def compute_personal(pool, key_type, key_values, new_wm):
    if not key_values: return 0
    col="trip" if key_type=="trip" else "name"
    count=0; now=int(time.time()); chunk=Settings().batch_chunk_size

    for ci in range(0, len(key_values), chunk):
        ch = key_values[ci:ci+chunk]
        async with pool.acquire() as c:
            rows = await c.fetch(f"""
                SELECT p.game_id, p.win, p.talk_count, p.live_days, p.role, p.{col} AS key_val,
                       g.village_name, g.started_at, g.haiyaku, g.haiyaku_class
                FROM players p JOIN games g ON g.id=p.game_id
                WHERE p.is_spectator=0 AND p.{col}=ANY($1::text[])
                ORDER BY p.{col}, g.started_at ASC
            """, ch)

        groups = defaultdict(list)
        ts_map = {}
        for r in rows:
            kv=r["key_val"]; groups[kv].append(r)
            ts=int(r["started_at"])
            if kv not in ts_map: ts_map[kv]=(ts,ts)
            else: ft,lt=ts_map[kv]; ts_map[kv]=(min(ft,ts),max(lt,ts))

        # このチャンクの全 game_id を一括で同村者取得
        all_gids = list({int(r["game_id"]) for r in rows})
        partner_counts: dict = defaultdict(lambda: defaultdict(int))
        if all_gids:
            async with pool.acquire() as c:
                partner_rows = await c.fetch(f"""
                    SELECT p.{col} AS key_val, p2.{col} AS partner
                    FROM players p
                    JOIN players p2 ON p2.game_id = p.game_id
                    WHERE p.game_id = ANY($1::int[])
                      AND p.is_spectator=0 AND p2.is_spectator=0
                      AND p2.{col} IS NOT NULL AND p2.{col} <> ''
                      AND p.{col} <> p2.{col}
                      AND p.{col} = ANY($2::text[])
                """, all_gids, ch)
            for r in partner_rows:
                partner_counts[r["key_val"]][r["partner"]] += 1

        async with pool.acquire() as c:
            for kv, prows in groups.items():
                agg = aggregate_rows_3class(prows)
                ft, lt = ts_map.get(kv, (None, None))
                stats_json = json.dumps(agg["buckets"], ensure_ascii=False)
                gids_json = json.dumps(agg["game_ids"])

                # 同村者上位30件
                top30 = sorted(partner_counts.get(kv, {}).items(), key=lambda x: -x[1])[:30]
                partners_json = json.dumps(
                    [{"key": k, "games": v} for k, v in top30],
                    ensure_ascii=False
                )

                await c.execute("""
                    INSERT INTO personal_stats_cache
                      (key_type,key_value,watermark_game_id,stats_json,game_ids_json,partners_json,first_ts,last_ts,updated_at)
                    VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9)
                    ON CONFLICT (key_type,key_value) DO UPDATE SET
                        watermark_game_id=EXCLUDED.watermark_game_id,
                        stats_json=EXCLUDED.stats_json,
                        game_ids_json=EXCLUDED.game_ids_json,
                        partners_json=EXCLUDED.partners_json,
                        first_ts=EXCLUDED.first_ts,
                        last_ts=EXCLUDED.last_ts,
                        updated_at=EXCLUDED.updated_at
                """, key_type, kv, new_wm, stats_json, gids_json, partners_json, ft, lt, now)
                count += 1

        if ci+chunk < len(key_values):
            log.info("  ... 個人(%s) %d/%d", key_type, min(ci+chunk, len(key_values)), len(key_values))
    return count

async def rebuild_personal(pool, wm, new_wm, force_full):
    t0 = time.time()
    if force_full or wm == 0:
        log.info("個人戦績: フルリビルド")
        async with pool.acquire() as c:
            tr = await c.fetch("SELECT DISTINCT trip FROM players WHERE is_spectator=0 AND trip IS NOT NULL AND trip<>''")
            nr = await c.fetch("SELECT DISTINCT name FROM players WHERE is_spectator=0 AND name IS NOT NULL AND name<>''")
        trips=[r["trip"] for r in tr]; names=[r["name"] for r in nr]
        log.info("  対象: %d trips, %d names", len(trips), len(names))
    else:
        at, an = await get_affected_players(pool, wm)
        rt, rn = await get_recent_month_players(pool)
        pt, pn = await get_priority_players(pool)
        trips=sorted(at | rt | pt); names=sorted(an | rn | pn)
        log.info("個人戦績: 差分+直近1カ月+検索優先 (trip=%d, name=%d / affected=%d,%d recent=%d,%d priority=%d,%d)",
                 len(trips), len(names), len(at), len(an), len(rt), len(rn), len(pt), len(pn))

    tc = await compute_personal(pool, "trip", trips, new_wm)
    nc = await compute_personal(pool, "name", names, new_wm)
    log.info("個人戦績: 完了 (trip=%d, name=%d, %.1f秒)", tc, nc, time.time()-t0)


# ╔═════════════════════════════════════╗
# ║  メイン                               ║
# ╚═════════════════════════════════════╝

SHARED_TRIP_THRESHOLD = 20  # この名前数以上で割れ酉判定

async def rebuild_shared_trips(pool):
    """同一トリップに対応する名前数が閾値以上のトリップを割れ酉として記録"""
    t0 = time.time()
    log.info("割れ酉判定: 開始")
    now = int(time.time())
    async with pool.acquire() as c:
        rows = await c.fetch("""
            SELECT trip, COUNT(DISTINCT name) AS name_count
            FROM players
            WHERE is_spectator=0 AND trip IS NOT NULL AND trip <> ''
              AND name IS NOT NULL AND name <> ''
            GROUP BY trip
            HAVING COUNT(DISTINCT name) >= $1
        """, SHARED_TRIP_THRESHOLD)

    if not rows:
        log.info("割れ酉判定: 対象なし")
        return

    async with pool.acquire() as c:
        # 既存をクリアして再構築
        await c.execute("DELETE FROM shared_trips")
        await c.executemany(
            "INSERT INTO shared_trips (trip, name_count, updated_at) VALUES ($1, $2, $3)",
            [(r["trip"], int(r["name_count"]), now) for r in rows]
        )
    # RM実績（そのトリップでの村建て回数）を付与。自動追放の免除判定に使う
    async with pool.acquire() as c:
        await c.execute("ALTER TABLE shared_trips ADD COLUMN IF NOT EXISTS rm_games INT NOT NULL DEFAULT 0")
        await c.execute("""
            UPDATE shared_trips s SET rm_games = COALESCE((
              SELECT COUNT(*) FROM players p JOIN games g ON g.id = p.game_id
              WHERE p.trip = s.trip AND p.name = g.rm_name AND p.is_spectator = 0), 0)
        """)
    log.info("割れ酉判定: %d件 (%.1f秒)", len(rows), time.time()-t0)


async def main(args):
    log.info("=" * 60)
    log.info("バッチ集計開始")
    log.info("=" * 60)
    pool = await create_pool()
    try:
        await ensure_tables(pool)
        wm = await get_watermark(pool)
        new_wm = await get_max_game_id(pool)
        force = args.full
        log.info("ウォーターマーク: 前回=%d, 今回=%d%s", wm, new_wm, " (フル)" if force else "")
        if new_wm == 0: log.warning("games空"); return
        if wm == new_wm and not force: log.info("新規なし"); return

        if not args.personal_only: await rebuild_ranks(pool, wm, force)
        if not args.ranks_only: await rebuild_personal(pool, wm, new_wm, force)
        await rebuild_shared_trips(pool)

        await set_watermark(pool, new_wm)
        log.info("ウォーターマーク: %d→%d", wm, new_wm)

        async with pool.acquire() as c:
            for t in ("rank_cache_json","personal_stats_cache","batch_state","shared_trips"):
                await c.execute(f"ANALYZE {t}")

        log.info("=" * 60)
        log.info("バッチ集計完了")
        log.info("=" * 60)
    finally:
        await pool.close()

if __name__ == "__main__":
    p = argparse.ArgumentParser()
    p.add_argument("--full", action="store_true")
    p.add_argument("--ranks-only", action="store_true")
    p.add_argument("--personal-only", action="store_true")
    asyncio.run(main(p.parse_args()))
