from flask import Flask


from .constants import SYS_USER_LANGUAGE, SYS_USER_NICKNAME
from .models import VariableValue
from ...dao import db
from typing import Optional

import logging
from flaskr.service.user.repository import (
    UserAggregate,
    _ensure_user_entity as ensure_user_entity,
    load_user_aggregate,
    update_user_entity_fields,
)
from ...i18n import _
import datetime
from ..check_risk.funcs import add_risk_control_result
from flaskr.api.check import (
    check_text,
    CHECK_RESULT_PASS,
    CHECK_RESULT_REJECT,
)
from flaskr.util.uuid import generate_id
from flaskr.service.common import raise_error
from flaskr.service.profile.profile_manage import get_profile_item_definition_list
from flaskr.service.profile.dtos import ProfileToSave
from flaskr.service.user.dtos import UserProfileLabelDTO, UserProfileLabelItemDTO

logger = logging.getLogger(__name__)


def _get_latest_variable_value(
    values: list[VariableValue],
    variable_key: str,
    shifu_bid: str,
) -> Optional[VariableValue]:
    """
    Return the newest variable value row from a pre-fetched, id-desc sorted
    collection.

    Matching is by key only (not variable_bid) so the newest row for the
    logical profile field wins even if the underlying Variable definition was
    recreated and now has a different variable_bid.

    Precedence:
    1) shifu scope (shifu_bid) - newest record matching key
    2) global/system scope (empty shifu_bid) - newest record matching key
    """
    target_shifu = shifu_bid or ""

    def _pick(scope_shifu_bid: str) -> Optional[VariableValue]:
        return next(
            (
                item
                for item in values
                if item.shifu_bid == scope_shifu_bid and item.key == variable_key
            ),
            None,
        )

    scoped = _pick(target_shifu)
    if scoped:
        return scoped

    if target_shifu:
        return _pick("")

    return None


def _ensure_user_aggregate(user_id: str) -> Optional[UserAggregate]:
    aggregate = load_user_aggregate(user_id)
    if aggregate:
        return aggregate
    ensure_user_entity(user_id)
    return load_user_aggregate(user_id)


def _update_aggregate_field(
    aggregate: Optional[UserAggregate], mapping: str, value
) -> None:
    if not aggregate:
        return
    if mapping == "name":
        aggregate.nickname = value or ""
    elif mapping == "user_avatar":
        aggregate.avatar = value or ""
    elif mapping == "user_language":
        aggregate.language = value or ""
    elif mapping == "user_birth":
        aggregate.birthday = value


def _normalize_core_value(mapping: str, value):
    if mapping == "user_birth":
        if isinstance(value, datetime.date):
            return value
        if isinstance(value, str) and value:
            try:
                return datetime.date.fromisoformat(value)
            except ValueError:
                return None
        return None
    return value or ""


def _apply_core_mapping(user_id: str, mapping: str, value):
    entity = ensure_user_entity(user_id)
    normalized = _normalize_core_value(mapping, value)
    if mapping == "name":
        update_user_entity_fields(entity, nickname=normalized)
    elif mapping == "user_avatar":
        update_user_entity_fields(entity, avatar=normalized)
    elif mapping == "user_language":
        update_user_entity_fields(entity, language=normalized)
    elif mapping == "user_birth":
        update_user_entity_fields(entity, birthday=normalized)
    return normalized


def _current_core_value(aggregate: Optional[UserAggregate], mapping: str):
    if not aggregate:
        return None
    if mapping == "name":
        return aggregate.nickname
    if mapping == "user_avatar":
        return aggregate.avatar
    if mapping == "user_language":
        return aggregate.language
    if mapping == "user_birth":
        return aggregate.birthday
    return None


def check_text_content(
    app: Flask,
    user_id: str,
    input: str,
):
    check_id = generate_id(app)
    res = check_text(app, check_id, input, user_id)
    add_risk_control_result(
        app,
        check_id,
        user_id,
        input,
        res.provider,
        res.check_result,
        str(res.raw_data),
        1 if res.check_result == CHECK_RESULT_PASS else 0,
        "check_text",
    )
    if res.check_result == CHECK_RESULT_REJECT:
        return False
    return True


def get_profile_labels():
    # language = get_current_language()
    return {
        "sys_user_nickname": {
            "label": _("server.profile.nickname"),
            "mapping": "name",
            "default": "",
        },
        "sys_user_background": {"label": _("server.profile.userBackground")},
        "sex": {
            "label": _("server.profile.sex"),
            "mapping": "user_sex",
            "items": [
                _("server.profile.sexMale"),
                _("server.profile.sexFemale"),
                _("server.profile.sexSecret"),
            ],
            "items_mapping": {
                0: _("server.profile.sexSecret"),
                1: _("server.profile.sexMale"),
                2: _("server.profile.sexFemale"),
            },
            "default": 0,
        },
        "birth": {
            "label": _("server.profile.birth"),
            "mapping": "user_birth",
            "type": "date",
            "default": datetime.date(2003, 1, 1),
        },
        "avatar": {
            "label": _("server.profile.avatar"),
            "mapping": "user_avatar",
            "type": "image",
            "default": "",
        },
        "language": {
            "label": _("server.profile.language"),
            "items": ["中文", "English", "Français"],
            "mapping": "user_language",
            "items_mapping": {
                "zh-CN": "中文",
                "en-US": "English",
                "fr-FR": "Français",
            },
            "default": "zh-CN",
        },
        "sys_user_style": {
            "label": _("server.profile.style"),
        },
    }


def save_user_profiles(
    app: Flask, user_id: str, course_id: str, profiles: list[ProfileToSave]
) -> bool:
    PROFILES_LABLES = get_profile_labels()
    app.logger.info("save user profiles:%s", profiles)
    aggregate = _ensure_user_aggregate(user_id)
    profiles_items = get_profile_item_definition_list(app, course_id)

    candidate_shifus = [course_id or ""]
    if course_id:
        candidate_shifus.append("")

    try:
        user_values: list[VariableValue] = (
            VariableValue.query.filter(
                VariableValue.user_bid == user_id,
                VariableValue.deleted == 0,
                VariableValue.shifu_bid.in_(candidate_shifus),
            )
            .order_by(VariableValue.id.desc())
            .all()
        )
    except Exception as exc:  # pragma: no cover - defensive fallback
        app.logger.warning("Failed to load var_variable_values: %s", exc)
        user_values = []

    for profile in profiles:
        profile_item = next(
            (item for item in profiles_items if item.profile_key == profile.key), None
        )
        variable_bid = (profile.bid or "").strip() or (
            profile_item.profile_id if profile_item else ""
        )
        target_shifu = "" if profile.key in PROFILES_LABLES else (course_id or "")

        latest_value = _get_latest_variable_value(
            user_values,
            variable_key=profile.key,
            shifu_bid=target_shifu,
        )
        if not latest_value or latest_value.value != profile.value:
            user_value = VariableValue(
                variable_value_bid=generate_id(app),
                user_bid=user_id,
                shifu_bid=target_shifu,
                variable_bid=variable_bid,
                key=profile.key,
                value=profile.value or "",
                deleted=0,
            )
            db.session.add(user_value)
            user_values.insert(0, user_value)

        if profile.key in PROFILES_LABLES:
            profile_lable = PROFILES_LABLES[profile.key]
            if profile_lable.get("mapping"):
                if profile_lable.get("items_mapping"):
                    profile.value = profile_lable["items_mapping"].get(
                        profile.value, profile.value
                    )
                normalized = _apply_core_mapping(
                    user_id, profile_lable["mapping"], profile.value
                )
                _update_aggregate_field(aggregate, profile_lable["mapping"], normalized)

    db.session.flush()
    return True


def get_user_profiles(app: Flask, user_id: str, course_id: str) -> dict:
    """Get user profiles for Mdflow run.

    NOTE:
    - Some profile keys ("labels") are stored globally with ``shifu_bid=''``.
    - Other profile keys are stored per-course with ``shifu_bid=course_id``.

    This function must follow the same shifu_bid routing rules as
    :func:`save_user_profiles`, otherwise the run context may see values different
    from what the user sees in "个人设置".
    """

    PROFILES_LABLES = get_profile_labels()
    profiles_items = get_profile_item_definition_list(app, course_id)

    candidate_shifus = [course_id or ""]
    if course_id:
        candidate_shifus.append("")

    try:
        user_values: list[VariableValue] = (
            VariableValue.query.filter(
                VariableValue.user_bid == user_id,
                VariableValue.deleted == 0,
                VariableValue.shifu_bid.in_(candidate_shifus),
            )
            .order_by(VariableValue.id.desc())
            .all()
        )
    except Exception as exc:  # pragma: no cover - defensive fallback
        app.logger.warning("Failed to load var_variable_values: %s", exc)
        user_values = []

    aggregate = load_user_aggregate(user_id, with_credentials=False)

    result: dict[str, str] = {}
    for profile_item in profiles_items:
        # Follow save_user_profiles routing: label keys are global, others per-course.
        target_shifu = (
            "" if profile_item.profile_key in PROFILES_LABLES else (course_id or "")
        )

        user_value = (
            _get_latest_variable_value(
                user_values,
                variable_key=profile_item.profile_key,
                shifu_bid=target_shifu,
            )
            if user_values
            else None
        )
        if user_value:
            result[profile_item.profile_key] = user_value.value

    # Keep runtime variable resolution aligned with /api/user/get_profile:
    # mapped system fields should use the latest canonical user entity values.
    if aggregate:
        for key, profile_label in PROFILES_LABLES.items():
            mapping = profile_label.get("mapping")
            if not mapping:
                continue
            raw_value = _current_core_value(aggregate, mapping)
            if raw_value is None:
                continue
            if isinstance(raw_value, datetime.date):
                result[key] = raw_value.isoformat()
            else:
                result[key] = str(raw_value)
        # Runtime prompt variables must stay consistent with canonical user fields.
        result[SYS_USER_LANGUAGE] = aggregate.user_language
        result[SYS_USER_NICKNAME] = aggregate.nickname or ""

    # Ensure system variables are always available.
    if result.get(SYS_USER_LANGUAGE) is None:
        result[SYS_USER_LANGUAGE] = aggregate.user_language if aggregate else "en-US"

    if not result.get(SYS_USER_NICKNAME):
        result[SYS_USER_NICKNAME] = aggregate.nickname if aggregate else ""

    return result


def get_user_profile_labels(
    app: Flask, user_id: str, course_id: str
) -> UserProfileLabelDTO:
    """
    Get user profile labels
    Args:
        app: Flask application instance
        user_id: User id
        course_id: Course id
    Returns:
        list: User profile labels
    """
    app.logger.info("get user profile labels:{}".format(course_id))
    candidate_shifus = [course_id or ""]
    if course_id:
        candidate_shifus.append("")

    try:
        user_values: list[VariableValue] = (
            VariableValue.query.filter(
                VariableValue.user_bid == user_id,
                VariableValue.deleted == 0,
                VariableValue.shifu_bid.in_(candidate_shifus),
            )
            .order_by(VariableValue.id.desc())
            .all()
        )
    except Exception as exc:  # pragma: no cover - defensive fallback
        app.logger.warning("Failed to load var_variable_values: %s", exc)
        user_values = []
    profiles_items = get_profile_item_definition_list(app, course_id)
    PROFILES_LABLES = get_profile_labels()
    aggregate = load_user_aggregate(user_id)
    language_value = aggregate.user_language if aggregate else "en-US"
    result = UserProfileLabelDTO(profiles=[], language=language_value)
    mapping_keys = []
    if aggregate:
        for key, meta in PROFILES_LABLES.items():
            mapping = meta.get("mapping")
            if not mapping:
                continue
            mapping_keys.append(key)
            raw_value = _current_core_value(aggregate, mapping)
            if raw_value is None:
                value_entry = (
                    _get_latest_variable_value(
                        user_values,
                        variable_key=key,
                        shifu_bid="",
                    )
                    if user_values
                    else None
                )
                if value_entry:
                    raw_value = value_entry.value
            display_value = raw_value
            if meta.get("items_mapping"):
                mapping_items = meta.get("items", [])
                default_value = mapping_items[0] if mapping_items else ""
                display_value = meta["items_mapping"].get(raw_value, default_value)
            result.profiles.append(
                UserProfileLabelItemDTO(
                    key=key,
                    label=meta["label"],
                    type=meta.get("type", "select" if "items" in meta else "text"),
                    value=display_value,
                    items=meta.get("items"),
                )
            )
    for key in PROFILES_LABLES.keys():
        if key in mapping_keys:
            continue
        profile_key = key
        item = {
            "key": profile_key,
            "label": PROFILES_LABLES[profile_key]["label"],
            "type": PROFILES_LABLES[profile_key].get(
                "type",
                ("select" if "items" in PROFILES_LABLES[profile_key] else "text"),
            ),
            "value": "",
            "items": (
                PROFILES_LABLES[profile_key]["items"]
                if "items" in PROFILES_LABLES[profile_key]
                else None
            ),
        }
        user_value = None
        profile_item = next(
            (item for item in profiles_items if item.profile_key == profile_key), None
        )
        if profile_item:
            if user_values:
                user_value = _get_latest_variable_value(
                    user_values,
                    variable_key=profile_key,
                    shifu_bid="",
                )
        else:
            app.logger.info("profile_item not found:{}".format(profile_key))
        if user_value is None and user_values:
            user_value = _get_latest_variable_value(
                user_values,
                variable_key=profile_key,
                shifu_bid="",
            )
        if user_value:
            item["value"] = user_value.value
        result.profiles.append(
            UserProfileLabelItemDTO(
                key=item["key"],
                label=item["label"],
                type=item["type"],
                value=item["value"],
                items=item["items"],
            )
        )

    return result


def update_user_profile_with_lable(
    app: Flask,
    user_id: str,
    profiles: list,
    update_all: bool = False,
    course_id: str = None,
):
    app.logger.info("update user profile with lable:{}".format(course_id))
    PROFILES_LABLES = get_profile_labels()
    if isinstance(profiles, UserProfileLabelDTO):
        profiles = profiles.profiles or []
    elif isinstance(profiles, UserProfileLabelItemDTO):
        profiles = [profiles]

    if profiles and isinstance(profiles[0], UserProfileLabelItemDTO):
        profiles = [item.__json__() for item in profiles]

    aggregate = _ensure_user_aggregate(user_id)
    profile_items = get_profile_item_definition_list(app, course_id)

    if not profiles:
        db.session.flush()
        return True

    nickname = next((p for p in profiles if p.get("key") == "sys_user_nickname"), None)
    if nickname and not check_text_content(app, user_id, nickname.get("value")):
        raise_error("server.common.nicknameNotAllowed")

    background = next(
        (p for p in profiles if p.get("key") == "sys_user_background"), None
    )
    if background and not check_text_content(app, user_id, background.get("value")):
        raise_error("server.common.backgroundNotAllowed")

    candidate_shifus = [course_id or ""]
    if course_id:
        candidate_shifus.append("")

    try:
        user_values: list[VariableValue] = (
            VariableValue.query.filter(
                VariableValue.user_bid == user_id,
                VariableValue.deleted == 0,
                VariableValue.shifu_bid.in_(candidate_shifus),
            )
            .order_by(VariableValue.id.desc())
            .all()
        )
    except Exception as exc:  # pragma: no cover - defensive fallback
        app.logger.warning("Failed to load var_variable_values: %s", exc)
        user_values = []

    for profile in profiles:
        key = profile.get("key")
        if not key:
            continue
        profile_value = profile.get("value")
        profile_item = next(
            (
                item
                for item in profile_items
                if item.profile_key == key or item.profile_id == profile.get("id")
            ),
            None,
        )

        app.logger.info("update user profile:%s-%s", key, profile_value)

        profile_lable = PROFILES_LABLES.get(key, None)
        default_value = profile_lable.get("default", None) if profile_lable else None

        if profile_lable and profile_lable.get("items_mapping"):
            for source_value, mapped in profile_lable["items_mapping"].items():
                if mapped == profile_value:
                    profile_value = source_value
                    break

        app.logger.info("profile_value:%s", profile_value)
        mapping = profile_lable.get("mapping") if profile_lable else None
        if mapping and (
            update_all
            or (
                profile_value != default_value
                and _current_core_value(aggregate, mapping) != profile_value
            )
        ):
            app.logger.info(
                "update user info: %s - %s",
                key,
                profile_value,
            )
            normalized = _apply_core_mapping(user_id, mapping, profile_value)
            _update_aggregate_field(aggregate, mapping, normalized)
        elif not profile_lable:
            app.logger.info("profile_lable not found:%s", key)

        # System variables (in PROFILES_LABLES) are global; custom variables
        # are scoped to the course.  This must match save_user_profiles() so
        # that the run interface reads the same value the settings page wrote.
        target_shifu = "" if key in PROFILES_LABLES else (course_id or "")

        should_persist_value = (
            profile_value not in (None, "") and profile_value != default_value
        )
        if should_persist_value:
            latest_value = _get_latest_variable_value(
                user_values,
                variable_key=key,
                shifu_bid=target_shifu,
            )
            if latest_value is None or latest_value.value != profile_value:
                variable_bid = profile_item.profile_id if profile_item else ""
                new_value = VariableValue(
                    variable_value_bid=generate_id(app),
                    user_bid=user_id,
                    shifu_bid=target_shifu,
                    variable_bid=variable_bid,
                    key=key,
                    value=str(profile_value) if profile_value is not None else "",
                    deleted=0,
                )
                db.session.add(new_value)
                user_values.insert(0, new_value)
    db.session.flush()
    return True
