import asyncio
from collections.abc import Sequence
from datetime import UTC, date, datetime
import json
import math
from typing import TYPE_CHECKING, Any

from app.calculator import (
    calculate_pp,
    calculate_pp_weight,
    calculate_score_to_level,
    calculate_weighted_acc,
    calculate_weighted_pp,
    clamp,
)
from app.config import settings
from app.database.team import TeamMember
from app.models.model import (
    CurrentUserAttributes,
    PinAttributes,
    RespWithCursor,
    UTCBaseModel,
)
from app.models.mods import APIMod, mods_can_get_pp
from app.models.score import (
    INT_TO_MODE,
    MODE_TO_INT,
    GameMode,
    HitResult,
    LeaderboardType,
    Rank,
    ScoreStatistics,
    SoloScoreSubmissionInfo,
)

from .beatmap import Beatmap, BeatmapResp
from .beatmap_playcounts import process_beatmap_playcount
from .beatmapset import BeatmapsetResp
from .best_score import BestScore
from .lazer_user import User, UserResp
from .monthly_playcounts import MonthlyPlaycounts
from .pp_best_score import PPBestScore
from .relationship import (
    Relationship as DBRelationship,
    RelationshipType,
)
from .score_token import ScoreToken

from redis.asyncio import Redis
from sqlalchemy import Column, ColumnExpressionArgument, DateTime
from sqlalchemy.ext.asyncio import AsyncAttrs
from sqlalchemy.orm import aliased
from sqlalchemy.sql.elements import ColumnElement
from sqlmodel import (
    JSON,
    BigInteger,
    Field,
    ForeignKey,
    Relationship,
    SQLModel,
    col,
    func,
    select,
    text,
    true,
)
from sqlmodel.ext.asyncio.session import AsyncSession
from sqlmodel.sql._expression_select_cls import SelectOfScalar

if TYPE_CHECKING:
    from app.fetcher import Fetcher


class ScoreBase(AsyncAttrs, SQLModel, UTCBaseModel):
    # 基本字段
    accuracy: float
    map_md5: str = Field(max_length=32, index=True)
    build_id: int | None = Field(default=None)
    classic_total_score: int | None = Field(
        default=0, sa_column=Column(BigInteger)
    )  # solo_score
    ended_at: datetime = Field(sa_column=Column(DateTime))
    has_replay: bool
    max_combo: int
    mods: list[APIMod] = Field(sa_column=Column(JSON))
    passed: bool
    playlist_item_id: int | None = Field(default=None)  # multiplayer
    pp: float = Field(default=0.0)
    preserve: bool = Field(default=True)
    rank: Rank
    room_id: int | None = Field(default=None)  # multiplayer
    started_at: datetime = Field(sa_column=Column(DateTime))
    total_score: int = Field(default=0, sa_column=Column(BigInteger))
    total_score_without_mods: int = Field(
        default=0, sa_column=Column(BigInteger), exclude=True
    )
    type: str
    beatmap_id: int = Field(index=True, foreign_key="beatmaps.id")

    # optional
    # TODO: current_user_attributes
    # position: int | None = Field(default=None)  # multiplayer


class Score(ScoreBase, table=True):
    __tablename__ = "scores"  # pyright: ignore[reportAssignmentType]
    id: int | None = Field(
        default=None, sa_column=Column(BigInteger, autoincrement=True, primary_key=True)
    )
    user_id: int = Field(
        default=None,
        sa_column=Column(
            BigInteger,
            ForeignKey("lazer_users.id"),
            index=True,
        ),
    )
    # ScoreStatistics
    n300: int = Field(exclude=True)
    n100: int = Field(exclude=True)
    n50: int = Field(exclude=True)
    nmiss: int = Field(exclude=True)
    ngeki: int = Field(exclude=True)
    nkatu: int = Field(exclude=True)
    nlarge_tick_miss: int | None = Field(default=None, exclude=True)
    nlarge_tick_hit: int | None = Field(default=None, exclude=True)
    nslider_tail_hit: int | None = Field(default=None, exclude=True)
    nsmall_tick_hit: int | None = Field(default=None, exclude=True)
    gamemode: GameMode = Field(index=True)
    pinned_order: int = Field(default=0, exclude=True)

    # optional
    beatmap: Beatmap = Relationship()
    user: User = Relationship(sa_relationship_kwargs={"lazy": "joined"})

    @property
    def is_perfect_combo(self) -> bool:
        return self.max_combo == self.beatmap.max_combo

    @staticmethod
    def select_clause_unique(
        *where_clauses: ColumnExpressionArgument[bool] | bool,
    ) -> SelectOfScalar["Score"]:
        rownum = (
            func.row_number()
            .over(
                partition_by=col(Score.user_id), order_by=col(Score.total_score).desc()
            )
            .label("rn")
        )
        subq = select(Score, rownum).where(*where_clauses).subquery()
        best = aliased(Score, subq, adapt_on_names=True)
        return select(best).where(subq.c.rn == 1)


class ScoreResp(ScoreBase):
    id: int
    user_id: int
    is_perfect_combo: bool = False
    legacy_perfect: bool = False
    legacy_total_score: int = 0  # FIXME
    processed: bool = False  # solo_score
    weight: float = 0.0
    best_id: int | None = None
    ruleset_id: int | None = None
    beatmap: BeatmapResp | None = None
    beatmapset: BeatmapsetResp | None = None
    user: UserResp | None = None
    statistics: ScoreStatistics | None = None
    maximum_statistics: ScoreStatistics | None = None
    rank_global: int | None = None
    rank_country: int | None = None
    position: int | None = None
    scores_around: "ScoreAround | None" = None
    current_user_attributes: CurrentUserAttributes | None = None

    @classmethod
    async def from_db(cls, session: AsyncSession, score: Score) -> "ScoreResp":
        s = cls.model_validate(score.model_dump())
        assert score.id
        await score.awaitable_attrs.beatmap
        s.beatmap = await BeatmapResp.from_db(score.beatmap)
        s.beatmapset = await BeatmapsetResp.from_db(
            score.beatmap.beatmapset, session=session, user=score.user
        )
        s.is_perfect_combo = s.max_combo == s.beatmap.max_combo
        s.legacy_perfect = s.max_combo == s.beatmap.max_combo
        s.ruleset_id = MODE_TO_INT[score.gamemode]
        best_id = await get_best_id(session, score.id)
        if best_id:
            s.best_id = best_id
            s.weight = calculate_pp_weight(best_id - 1)
        s.statistics = {
            HitResult.MISS: score.nmiss,
            HitResult.MEH: score.n50,
            HitResult.OK: score.n100,
            HitResult.GREAT: score.n300,
            HitResult.PERFECT: score.ngeki,
            HitResult.GOOD: score.nkatu,
        }
        if score.nlarge_tick_miss is not None:
            s.statistics[HitResult.LARGE_TICK_MISS] = score.nlarge_tick_miss
        if score.nslider_tail_hit is not None:
            s.statistics[HitResult.SLIDER_TAIL_HIT] = score.nslider_tail_hit
        if score.nsmall_tick_hit is not None:
            s.statistics[HitResult.SMALL_TICK_HIT] = score.nsmall_tick_hit
        if score.nlarge_tick_hit is not None:
            s.statistics[HitResult.LARGE_TICK_HIT] = score.nlarge_tick_hit
        if score.gamemode == GameMode.MANIA:
            s.maximum_statistics = {
                HitResult.PERFECT: score.beatmap.max_combo,
            }
        else:
            s.maximum_statistics = {
                HitResult.GREAT: score.beatmap.max_combo,
            }
        s.user = await UserResp.from_db(
            score.user,
            session,
            include=["statistics", "team", "daily_challenge_user_stats"],
            ruleset=score.gamemode,
        )
        s.rank_global = (
            await get_score_position_by_id(
                session,
                score.beatmap_id,
                score.id,
                mode=score.gamemode,
                user=score.user,
            )
            or None
        )
        s.rank_country = (
            await get_score_position_by_id(
                session,
                score.beatmap_id,
                score.id,
                score.gamemode,
                score.user,
                type=LeaderboardType.COUNTRY,
            )
            or None
        )
        s.current_user_attributes = CurrentUserAttributes(
            pin=PinAttributes(is_pinned=bool(score.pinned_order), score_id=score.id)
        )
        return s


class MultiplayerScores(RespWithCursor):
    scores: list[ScoreResp] = Field(default_factory=list)
    params: dict[str, Any] = Field(default_factory=dict)


class ScoreAround(SQLModel):
    higher: MultiplayerScores | None = None
    lower: MultiplayerScores | None = None


async def get_best_id(session: AsyncSession, score_id: int) -> None:
    rownum = (
        func.row_number()
        .over(
            partition_by=col(PPBestScore.user_id), order_by=col(PPBestScore.pp).desc()
        )
        .label("rn")
    )
    subq = select(PPBestScore, rownum).subquery()
    stmt = select(subq.c.rn).where(subq.c.score_id == score_id)
    result = await session.exec(stmt)
    return result.one_or_none()


async def _score_where(
    type: LeaderboardType,
    beatmap: int,
    mode: GameMode,
    mods: list[str] | None = None,
    user: User | None = None,
) -> list[ColumnElement[bool]] | None:
    wheres = [
        col(BestScore.beatmap_id) == beatmap,
        col(BestScore.gamemode) == mode,
    ]

    if type == LeaderboardType.FRIENDS:
        if user and user.is_supporter:
            subq = (
                select(DBRelationship.target_id)
                .where(
                    DBRelationship.type == RelationshipType.FOLLOW,
                    DBRelationship.user_id == user.id,
                )
                .subquery()
            )
            wheres.append(col(BestScore.user_id).in_(select(subq.c.target_id)))
        else:
            return None
    elif type == LeaderboardType.COUNTRY:
        if user and user.is_supporter:
            wheres.append(
                col(BestScore.user).has(col(User.country_code) == user.country_code)
            )
        else:
            return None
    elif type == LeaderboardType.TEAM:
        if user:
            team_membership = await user.awaitable_attrs.team_membership
            if team_membership:
                team_id = team_membership.team_id
                wheres.append(
                    col(BestScore.user).has(
                        col(User.team_membership).has(TeamMember.team_id == team_id)
                    )
                )
    if mods:
        if user and user.is_supporter:
            wheres.append(
                text(
                    "JSON_CONTAINS(total_score_best_scores.mods, :w)"
                    " AND JSON_CONTAINS(:w, total_score_best_scores.mods)"
                )  # pyright: ignore[reportArgumentType]
            )
        else:
            return None
    return wheres


async def get_leaderboard(
    session: AsyncSession,
    beatmap: int,
    mode: GameMode,
    type: LeaderboardType = LeaderboardType.GLOBAL,
    mods: list[str] | None = None,
    user: User | None = None,
    limit: int = 50,
) -> tuple[list[Score], Score | None]:
    is_rx = "RX" in (mods or [])
    is_ap = "AP" in (mods or [])
    if settings.enable_osu_rx and is_rx:
        mode = GameMode.OSURX
    elif settings.enable_osu_ap and is_ap:
        mode = GameMode.OSUAP

    wheres = await _score_where(type, beatmap, mode, mods, user)
    if wheres is None:
        return [], None
    query = (
        select(BestScore)
        .where(*wheres)
        .limit(limit)
        .order_by(col(BestScore.total_score).desc())
    )
    if mods:
        query = query.params(w=json.dumps(mods))
    scores = [s.score for s in await session.exec(query)]
    user_score = None
    if user:
        self_query = (
            select(BestScore)
            .where(BestScore.user_id == user.id)
            .where(
                col(BestScore.beatmap_id) == beatmap,
                col(BestScore.gamemode) == mode,
            )
            .order_by(col(BestScore.total_score).desc())
            .limit(1)
        )
        if mods:
            self_query = self_query.where(
                text(
                    "JSON_CONTAINS(total_score_best_scores.mods, :w)"
                    " AND JSON_CONTAINS(:w, total_score_best_scores.mods)"
                )
            ).params(w=json.dumps(mods))
        user_bs = (await session.exec(self_query)).first()
        if user_bs:
            user_score = user_bs.score
        if user_score and user_score not in scores:
            scores.append(user_score)
    return scores, user_score


async def get_score_position_by_user(
    session: AsyncSession,
    beatmap: int,
    user: User,
    mode: GameMode,
    type: LeaderboardType = LeaderboardType.GLOBAL,
    mods: list[str] | None = None,
) -> int:
    wheres = await _score_where(type, beatmap, mode, mods, user=user)
    if wheres is None:
        return 0
    rownum = (
        func.row_number()
        .over(
            partition_by=col(BestScore.beatmap_id),
            order_by=col(BestScore.total_score).desc(),
        )
        .label("row_number")
    )
    subq = select(BestScore, rownum).join(Beatmap).where(*wheres).subquery()
    stmt = select(subq.c.row_number).where(subq.c.user_id == user.id)
    result = await session.exec(stmt)
    s = result.one_or_none()
    return s if s else 0


async def get_score_position_by_id(
    session: AsyncSession,
    beatmap: int,
    score_id: int,
    mode: GameMode,
    user: User | None = None,
    type: LeaderboardType = LeaderboardType.GLOBAL,
    mods: list[str] | None = None,
) -> int:
    wheres = await _score_where(type, beatmap, mode, mods, user=user)
    if wheres is None:
        return 0
    rownum = (
        func.row_number()
        .over(
            partition_by=col(BestScore.beatmap_id),
            order_by=col(BestScore.total_score).desc(),
        )
        .label("row_number")
    )
    subq = select(BestScore, rownum).join(Beatmap).where(*wheres).subquery()
    stmt = select(subq.c.row_number).where(subq.c.score_id == score_id)
    result = await session.exec(stmt)
    s = result.one_or_none()
    return s if s else 0


async def get_user_best_score_in_beatmap(
    session: AsyncSession,
    beatmap: int,
    user: int,
    mode: GameMode | None = None,
) -> BestScore | None:
    return (
        await session.exec(
            select(BestScore)
            .where(
                BestScore.gamemode == mode if mode is not None else true(),
                BestScore.beatmap_id == beatmap,
                BestScore.user_id == user,
            )
            .order_by(col(BestScore.total_score).desc())
        )
    ).first()


# FIXME
async def get_user_best_score_with_mod_in_beatmap(
    session: AsyncSession,
    beatmap: int,
    user: int,
    mod: list[str],
    mode: GameMode | None = None,
) -> BestScore | None:
    return (
        await session.exec(
            select(BestScore)
            .where(
                BestScore.gamemode == mode if mode is not None else True,
                BestScore.beatmap_id == beatmap,
                BestScore.user_id == user,
                # BestScore.mods == mod,
            )
            .order_by(col(BestScore.total_score).desc())
        )
    ).first()


async def get_user_best_pp_in_beatmap(
    session: AsyncSession,
    beatmap: int,
    user: int,
    mode: GameMode,
) -> PPBestScore | None:
    return (
        await session.exec(
            select(PPBestScore).where(
                PPBestScore.beatmap_id == beatmap,
                PPBestScore.user_id == user,
                PPBestScore.gamemode == mode,
            )
        )
    ).first()


async def get_user_best_pp(
    session: AsyncSession,
    user: int,
    mode: GameMode,
    limit: int = 200,
) -> Sequence[PPBestScore]:
    return (
        await session.exec(
            select(PPBestScore)
            .where(PPBestScore.user_id == user, PPBestScore.gamemode == mode)
            .order_by(col(PPBestScore.pp).desc())
            .limit(limit)
        )
    ).all()


async def process_user(
    session: AsyncSession, user: User, score: Score, length: int, ranked: bool = False
):
    assert user.id
    assert score.id
    mod_for_save = list({mod["acronym"] for mod in score.mods})
    previous_score_best = await get_user_best_score_in_beatmap(
        session, score.beatmap_id, user.id, score.gamemode
    )
    previous_score_best_mod = await get_user_best_score_with_mod_in_beatmap(
        session, score.beatmap_id, user.id, mod_for_save, score.gamemode
    )
    add_to_db = False
    mouthly_playcount = (
        await session.exec(
            select(MonthlyPlaycounts).where(
                MonthlyPlaycounts.user_id == user.id,
                MonthlyPlaycounts.year == date.today().year,
                MonthlyPlaycounts.month == date.today().month,
            )
        )
    ).first()
    if mouthly_playcount is None:
        mouthly_playcount = MonthlyPlaycounts(
            user_id=user.id, year=date.today().year, month=date.today().month
        )
        add_to_db = True
    statistics = None
    for i in await user.awaitable_attrs.statistics:
        if i.mode == score.gamemode.value:
            statistics = i
            break
    if statistics is None:
        raise ValueError(
            f"User {user.id} does not have statistics for mode {score.gamemode.value}"
        )

    # pc, pt, tth, tts
    statistics.total_score += score.total_score
    difference = (
        score.total_score - previous_score_best.total_score
        if previous_score_best
        else score.total_score
    )
    if difference > 0 and score.passed and ranked:
        match score.rank:
            case Rank.X:
                statistics.grade_ss += 1
            case Rank.XH:
                statistics.grade_ssh += 1
            case Rank.S:
                statistics.grade_s += 1
            case Rank.SH:
                statistics.grade_sh += 1
            case Rank.A:
                statistics.grade_a += 1
        if previous_score_best is not None:
            match previous_score_best.rank:
                case Rank.X:
                    statistics.grade_ss -= 1
                case Rank.XH:
                    statistics.grade_ssh -= 1
                case Rank.S:
                    statistics.grade_s -= 1
                case Rank.SH:
                    statistics.grade_sh -= 1
                case Rank.A:
                    statistics.grade_a -= 1
        else:
            previous_score_best = BestScore(
                user_id=user.id,
                beatmap_id=score.beatmap_id,
                gamemode=score.gamemode,
                score_id=score.id,
                total_score=score.total_score,
                rank=score.rank,
                mods=mod_for_save,
            )
            session.add(previous_score_best)

        statistics.ranked_score += difference
        statistics.level_current = calculate_score_to_level(statistics.ranked_score)
        statistics.maximum_combo = max(statistics.maximum_combo, score.max_combo)
    if score.passed and ranked:
        if previous_score_best_mod is not None:
            previous_score_best_mod.mods = mod_for_save
            previous_score_best_mod.score_id = score.id
            previous_score_best_mod.rank = score.rank
            previous_score_best_mod.total_score = score.total_score
        elif (
            previous_score_best is not None and previous_score_best.score_id != score.id
        ):
            session.add(
                BestScore(
                    user_id=user.id,
                    beatmap_id=score.beatmap_id,
                    gamemode=score.gamemode,
                    score_id=score.id,
                    total_score=score.total_score,
                    rank=score.rank,
                    mods=mod_for_save,
                )
            )
    statistics.play_count += 1
    mouthly_playcount.playcount += 1
    statistics.play_time += length
    statistics.count_100 += score.n100 + score.nkatu
    statistics.count_300 += score.n300 + score.ngeki
    statistics.count_50 += score.n50
    statistics.count_miss += score.nmiss
    statistics.total_hits += (
        score.n300 + score.n100 + score.n50 + score.ngeki + score.nkatu
    )

    if score.passed and ranked:
        best_pp_scores = await get_user_best_pp(session, user.id, score.gamemode)
        pp_sum = 0.0
        acc_sum = 0.0
        for i, bp in enumerate(best_pp_scores):
            pp_sum += calculate_weighted_pp(bp.pp, i)
            acc_sum += calculate_weighted_acc(bp.acc, i)
        if len(best_pp_scores):
            # https://github.com/ppy/osu-queue-score-statistics/blob/c538ae/osu.Server.Queues.ScoreStatisticsProcessor/Helpers/UserTotalPerformanceAggregateHelper.cs#L41-L45
            acc_sum *= 100 / (20 * (1 - math.pow(0.95, len(best_pp_scores))))
        acc_sum = clamp(acc_sum, 0.0, 100.0)
        statistics.pp = pp_sum
        statistics.hit_accuracy = acc_sum
    if add_to_db:
        session.add(mouthly_playcount)
    await process_beatmap_playcount(session, user.id, score.beatmap_id)
    await session.commit()
    await session.refresh(user)


async def process_score(
    user: User,
    beatmap_id: int,
    ranked: bool,
    score_token: ScoreToken,
    info: SoloScoreSubmissionInfo,
    fetcher: "Fetcher",
    session: AsyncSession,
    redis: Redis,
    item_id: int | None = None,
    room_id: int | None = None,
) -> Score:
    assert user.id
    can_get_pp = info.passed and ranked and mods_can_get_pp(info.ruleset_id, info.mods)
    acronyms = [mod["acronym"] for mod in info.mods]
    is_rx = "RX" in acronyms
    is_ap = "AP" in acronyms
    gamemode = INT_TO_MODE[info.ruleset_id]
    if settings.enable_osu_rx and is_rx and gamemode == GameMode.OSU:
        gamemode = GameMode.OSURX
    elif settings.enable_osu_ap and is_ap and gamemode == GameMode.OSU:
        gamemode = GameMode.OSUAP
    score = Score(
        accuracy=info.accuracy,
        max_combo=info.max_combo,
        # maximum_statistics=info.maximum_statistics,
        mods=info.mods,
        passed=info.passed,
        rank=info.rank,
        total_score=info.total_score,
        total_score_without_mods=info.total_score_without_mods,
        beatmap_id=beatmap_id,
        ended_at=datetime.now(UTC),
        gamemode=gamemode,
        started_at=score_token.created_at,
        user_id=user.id,
        preserve=info.passed,
        map_md5=score_token.beatmap.checksum,
        has_replay=False,
        type="solo",
        n300=info.statistics.get(HitResult.GREAT, 0),
        n100=info.statistics.get(HitResult.OK, 0),
        n50=info.statistics.get(HitResult.MEH, 0),
        nmiss=info.statistics.get(HitResult.MISS, 0),
        ngeki=info.statistics.get(HitResult.PERFECT, 0),
        nkatu=info.statistics.get(HitResult.GOOD, 0),
        nlarge_tick_miss=info.statistics.get(HitResult.LARGE_TICK_MISS, 0),
        nsmall_tick_hit=info.statistics.get(HitResult.SMALL_TICK_HIT, 0),
        nlarge_tick_hit=info.statistics.get(HitResult.LARGE_TICK_HIT, 0),
        nslider_tail_hit=info.statistics.get(HitResult.SLIDER_TAIL_HIT, 0),
        playlist_item_id=item_id,
        room_id=room_id,
    )
    if can_get_pp:
        beatmap_raw = await fetcher.get_or_fetch_beatmap_raw(redis, beatmap_id)
        pp = await asyncio.get_event_loop().run_in_executor(
            None, calculate_pp, score, beatmap_raw
        )
        score.pp = pp
    session.add(score)
    user_id = user.id
    await session.commit()
    await session.refresh(score)
    if can_get_pp:
        previous_pp_best = await get_user_best_pp_in_beatmap(
            session, beatmap_id, user_id, score.gamemode
        )
        if previous_pp_best is None or score.pp > previous_pp_best.pp:
            assert score.id
            best_score = PPBestScore(
                user_id=user_id,
                score_id=score.id,
                beatmap_id=beatmap_id,
                gamemode=score.gamemode,
                pp=score.pp,
                acc=score.accuracy,
            )
            session.add(best_score)
            await session.delete(previous_pp_best) if previous_pp_best else None
            await session.commit()
            await session.refresh(score)
    await session.refresh(score_token)
    await session.refresh(user)
    await redis.publish("score:processed", score.id)
    return score