"""站内信服务：个人通知 + 公告。"""
import logging
from datetime import datetime, timezone
from typing import Any

from sqlalchemy import and_, delete, func, not_, or_, select, update
from sqlalchemy.ext.asyncio import AsyncSession

from app.models.announcement import Announcement, AnnouncementRead
from app.models.notification import Notification, NotificationPreference
from app.models.user import User, UserStatus
from app.models.user_agent import AgentStatus, UserAgent
from app.models.user_membership import UserMembership

log = logging.getLogger("qycw.notification")

CATEGORY_SOURCE: dict[str, tuple[str, str]] = {
    "order": ("订单中心", "order"),
    "instance": ("云服务器中心", "instance"),
    "finance": ("财务中心", "finance"),
    "ticket": ("工单中心", "ticket"),
    "promotion": ("推广中心", "promotion"),
    "membership": ("会员中心", "membership"),
    "announcement": ("平台公告", "announcement"),
    "system": ("系统通知", "system"),
}

NOTIFICATION_CATEGORIES = ["system", "order", "instance", "finance", "ticket", "promotion", "membership", "announcement"]
VALID_LEVELS = {"info", "success", "warning", "error", "urgent"}
IMPORTANT_LEVELS = {"warning", "error", "urgent"}


def _source_meta(category: str) -> dict[str, str]:
    name, sender_type = CATEGORY_SOURCE.get(category, CATEGORY_SOURCE["system"])
    return {
        "sender_name": name,
        "sender_type": sender_type,
        "source_label": name,
    }


def _normalize_level(level: str | None) -> str:
    value = (level or "info").strip().lower()
    return value if value in VALID_LEVELS else "info"


def _sender_meta(
    category: str,
    *,
    sender_id: int | None = None,
    sender_name: str | None = None,
    sender_type: str | None = None,
) -> dict[str, str | int | None]:
    fallback = _source_meta(category)
    return {
        "sender_id": sender_id,
        "sender_name": sender_name or fallback["sender_name"],
        "sender_type": sender_type or fallback["sender_type"],
        "source_label": fallback["source_label"],
    }


async def _in_app_enabled(db: AsyncSession, user_id: int, category: str) -> bool:
    res = await db.execute(
        select(NotificationPreference.in_app_enabled).where(
            NotificationPreference.user_id == user_id,
            NotificationPreference.category == category,
        )
    )
    value = res.scalar_one_or_none()
    return True if value is None else bool(value)


async def _disabled_in_app_user_ids(db: AsyncSession, user_ids: list[int], category: str) -> set[int]:
    if not user_ids:
        return set()
    res = await db.execute(
        select(NotificationPreference.user_id).where(
            NotificationPreference.user_id.in_(user_ids),
            NotificationPreference.category == category,
            NotificationPreference.in_app_enabled.is_(False),
        )
    )
    return {int(uid) for uid in res.scalars().all()}


# ── 个人通知 ────────────────────────────────────────────────────────

async def notify(
    db: AsyncSession,
    *,
    user_id: int,
    category: str,
    title: str,
    content: str,
    link: str | None = None,
    level: str = "info",
    sender_id: int | None = None,
    sender_name: str | None = None,
    sender_type: str | None = None,
    action_label: str | None = None,
    action_url: str | None = None,
) -> Notification:
    """创建一条个人通知（不 commit，由调用方事务控制）。"""
    if not await _in_app_enabled(db, user_id, category):
        log.info("notification skipped by preference user=%s category=%s", user_id, category)
        raise ValueError("notification disabled by user preference")
    sender = _sender_meta(
        category, sender_id=sender_id, sender_name=sender_name, sender_type=sender_type
    )
    item = Notification(
        user_id=user_id,
        category=category,
        level=_normalize_level(level),
        sender_id=sender["sender_id"],
        sender_name=str(sender["sender_name"]) if sender["sender_name"] else None,
        sender_type=str(sender["sender_type"] or "system"),
        title=title,
        content=content,
        link=link,
        action_label=action_label,
        action_url=action_url or link,
    )
    db.add(item)
    await db.flush()
    return item


async def safe_notify(
    db: AsyncSession,
    *,
    user_id: int,
    category: str,
    title: str,
    content: str,
    link: str | None = None,
    level: str = "info",
    sender_id: int | None = None,
    sender_name: str | None = None,
    sender_type: str | None = None,
    action_label: str | None = None,
    action_url: str | None = None,
) -> Notification | None:
    """业务触发点专用：发通知失败仅记日志，不抛异常、不阻断主流程。"""
    try:
        return await notify(
            db,
            user_id=user_id,
            category=category,
            title=title,
            content=content,
            link=link,
            level=level,
            sender_id=sender_id,
            sender_name=sender_name,
            sender_type=sender_type,
            action_label=action_label,
            action_url=action_url,
        )
    except Exception as exc:  # noqa: BLE001
        if isinstance(exc, ValueError) and "preference" in str(exc):
            return None
        log.warning("safe_notify failed user=%s category=%s: %s", user_id, category, exc)
        return None


async def notify_many(
    db: AsyncSession,
    *,
    user_ids: list[int],
    category: str,
    title: str,
    content: str,
    link: str | None = None,
    level: str = "info",
    sender_id: int | None = None,
    sender_name: str | None = None,
    sender_type: str | None = None,
    action_label: str | None = None,
    action_url: str | None = None,
) -> int:
    """批量给多个用户发个人通知，返回成功条数。"""
    disabled = await _disabled_in_app_user_ids(db, user_ids, category)
    sender = _sender_meta(
        category, sender_id=sender_id, sender_name=sender_name, sender_type=sender_type
    )
    count = 0
    for uid in user_ids:
        if uid in disabled:
            continue
        item = Notification(
            user_id=uid,
            category=category,
            level=_normalize_level(level),
            sender_id=sender["sender_id"],
            sender_name=str(sender["sender_name"]) if sender["sender_name"] else None,
            sender_type=str(sender["sender_type"] or "system"),
            title=title,
            content=content,
            link=link,
            action_label=action_label,
            action_url=action_url or link,
        )
        db.add(item)
        count += 1
    if count:
        await db.flush()
    return count


def notification_to_dict(item: Notification) -> dict[str, Any]:
    source = _source_meta(item.category)
    action_url = item.action_url or item.link
    return {
        "id": item.id,
        "category": item.category,
        "level": item.level,
        "sender_id": item.sender_id,
        "sender_name": item.sender_name or source["sender_name"],
        "sender_type": item.sender_type or source["sender_type"],
        "source_label": source["source_label"],
        "title": item.title,
        "content": item.content,
        "link": item.link,
        "action_label": item.action_label,
        "action_url": action_url,
        "action": {"label": item.action_label or "前往处理", "url": action_url} if action_url else None,
        "is_read": item.is_read,
        "read_at": item.read_at.isoformat() if item.read_at else None,
        "handled_at": item.handled_at.isoformat() if item.handled_at else None,
        "archived_at": item.archived_at.isoformat() if item.archived_at else None,
        "created_at": item.created_at.isoformat() if item.created_at else None,
    }


async def list_notifications(
    db: AsyncSession,
    user_id: int,
    *,
    category: str | None = None,
    level: str | None = None,
    unread_only: bool = False,
    important_only: bool = False,
    archived_only: bool = False,
    handled_state: str | None = None,
    limit: int = 20,
    offset: int = 0,
) -> tuple[list[dict], int]:
    conditions = [Notification.user_id == user_id]
    if category:
        conditions.append(Notification.category == category)
    if level:
        conditions.append(Notification.level == _normalize_level(level))
    if important_only:
        conditions.append(Notification.level.in_(IMPORTANT_LEVELS))
    if unread_only:
        conditions.append(Notification.is_read.is_(False))
    if archived_only:
        conditions.append(Notification.archived_at.is_not(None))
    else:
        conditions.append(Notification.archived_at.is_(None))
    if handled_state == "handled":
        conditions.append(Notification.handled_at.is_not(None))
    elif handled_state == "unhandled":
        conditions.append(Notification.handled_at.is_(None))

    total_res = await db.execute(
        select(func.count(Notification.id)).where(and_(*conditions))
    )
    total = int(total_res.scalar_one() or 0)

    res = await db.execute(
        select(Notification)
        .where(and_(*conditions))
        .order_by(Notification.id.desc())
        .limit(limit)
        .offset(offset)
    )
    items = [notification_to_dict(n) for n in res.scalars().all()]
    return items, total


async def unread_count(db: AsyncSession, user_id: int) -> dict[str, int]:
    """返回个人未读 + 公告未读 + 合计。"""
    notif_res = await db.execute(
        select(func.count(Notification.id)).where(
            Notification.user_id == user_id,
            Notification.is_read.is_(False),
            Notification.archived_at.is_(None),
        )
    )
    notif_unread = int(notif_res.scalar_one() or 0)

    ann_unread = 0
    if await _in_app_enabled(db, user_id, "announcement"):
        visible = _visible_announcements_condition(user_id)
        read_subq = (
            select(AnnouncementRead.announcement_id)
            .where(AnnouncementRead.user_id == user_id)
            .scalar_subquery()
        )
        ann_res = await db.execute(
            select(func.count(Announcement.id)).where(
                visible,
                not_(Announcement.id.in_(read_subq)),
            )
        )
        ann_unread = int(ann_res.scalar_one() or 0)

    return {
        "notifications": notif_unread,
        "announcements": ann_unread,
        "total": notif_unread + ann_unread,
    }


async def mark_read(db: AsyncSession, user_id: int, notification_id: int) -> bool:
    res = await db.execute(
        update(Notification)
        .where(
            Notification.id == notification_id,
            Notification.user_id == user_id,
            Notification.is_read.is_(False),
        )
        .values(is_read=True, read_at=datetime.now(timezone.utc))
    )
    await db.flush()
    return res.rowcount > 0


async def mark_all_read(db: AsyncSession, user_id: int) -> int:
    res = await db.execute(
        update(Notification)
        .where(
            Notification.user_id == user_id,
            Notification.is_read.is_(False),
            Notification.archived_at.is_(None),
        )
        .values(is_read=True, read_at=datetime.now(timezone.utc))
    )
    await db.flush()
    return res.rowcount


async def mark_many_read(db: AsyncSession, user_id: int, notification_ids: list[int]) -> int:
    if not notification_ids:
        return 0
    res = await db.execute(
        update(Notification)
        .where(
            Notification.user_id == user_id,
            Notification.id.in_(notification_ids),
            Notification.is_read.is_(False),
        )
        .values(is_read=True, read_at=datetime.now(timezone.utc))
    )
    await db.flush()
    return res.rowcount


async def set_many_handled(
    db: AsyncSession,
    user_id: int,
    notification_ids: list[int],
    *,
    handled: bool,
) -> int:
    if not notification_ids:
        return 0
    res = await db.execute(
        update(Notification)
        .where(Notification.user_id == user_id, Notification.id.in_(notification_ids))
        .values(handled_at=datetime.now(timezone.utc) if handled else None)
    )
    await db.flush()
    return res.rowcount


async def set_many_archived(
    db: AsyncSession,
    user_id: int,
    notification_ids: list[int],
    *,
    archived: bool,
) -> int:
    if not notification_ids:
        return 0
    res = await db.execute(
        update(Notification)
        .where(Notification.user_id == user_id, Notification.id.in_(notification_ids))
        .values(archived_at=datetime.now(timezone.utc) if archived else None)
    )
    await db.flush()
    return res.rowcount


async def delete_notification(db: AsyncSession, user_id: int, notification_id: int) -> bool:
    res = await db.execute(
        delete(Notification).where(
            Notification.id == notification_id,
            Notification.user_id == user_id,
        )
    )
    await db.flush()
    return res.rowcount > 0


async def delete_many_notifications(db: AsyncSession, user_id: int, notification_ids: list[int]) -> int:
    if not notification_ids:
        return 0
    res = await db.execute(
        delete(Notification).where(
            Notification.user_id == user_id,
            Notification.id.in_(notification_ids),
        )
    )
    await db.flush()
    return res.rowcount


def preference_to_dict(item: NotificationPreference) -> dict[str, Any]:
    return {
        "category": item.category,
        "in_app_enabled": item.in_app_enabled,
        "email_enabled": item.email_enabled,
        "sms_enabled": item.sms_enabled,
    }


async def get_preferences(db: AsyncSession, user_id: int) -> list[dict[str, Any]]:
    res = await db.execute(
        select(NotificationPreference).where(NotificationPreference.user_id == user_id)
    )
    existing = {item.category: item for item in res.scalars().all()}
    items: list[dict[str, Any]] = []
    for category in NOTIFICATION_CATEGORIES:
        item = existing.get(category)
        if item:
            items.append(preference_to_dict(item))
        else:
            items.append(
                {
                    "category": category,
                    "in_app_enabled": True,
                    "email_enabled": False,
                    "sms_enabled": False,
                }
            )
    return items


async def update_preferences(
    db: AsyncSession,
    user_id: int,
    preferences: list[dict[str, Any]],
) -> list[dict[str, Any]]:
    categories = {item["category"] for item in preferences if item.get("category") in NOTIFICATION_CATEGORIES}
    if not categories:
        return await get_preferences(db, user_id)

    res = await db.execute(
        select(NotificationPreference).where(
            NotificationPreference.user_id == user_id,
            NotificationPreference.category.in_(categories),
        )
    )
    existing = {item.category: item for item in res.scalars().all()}
    for data in preferences:
        category = data.get("category")
        if category not in NOTIFICATION_CATEGORIES:
            continue
        item = existing.get(category)
        if not item:
            item = NotificationPreference(user_id=user_id, category=category)
            db.add(item)
        if "in_app_enabled" in data:
            item.in_app_enabled = bool(data["in_app_enabled"])
        if "email_enabled" in data:
            item.email_enabled = bool(data["email_enabled"])
        if "sms_enabled" in data:
            item.sms_enabled = bool(data["sms_enabled"])
    await db.flush()
    return await get_preferences(db, user_id)


async def resolve_broadcast_user_ids(
    db: AsyncSession,
    *,
    target_type: str,
    target_user_ids: list[int] | None = None,
    target_role: str | None = None,
    target_member_level_ids: list[int] | None = None,
    active_agents_only: bool = True,
) -> list[int]:
    base_conditions = [User.status == UserStatus.ACTIVE]

    if target_type == "all":
        stmt = select(User.id).where(*base_conditions)
    elif target_type == "specific":
        ids = [int(uid) for uid in (target_user_ids or []) if int(uid) > 0]
        if not ids:
            return []
        stmt = select(User.id).where(*base_conditions, User.id.in_(ids))
    elif target_type == "role":
        if not target_role:
            return []
        stmt = select(User.id).where(*base_conditions, User.role == target_role)
    elif target_type == "member_level":
        ids = [int(item) for item in (target_member_level_ids or []) if int(item) > 0]
        if not ids:
            return []
        stmt = (
            select(User.id)
            .join(UserMembership, UserMembership.user_id == User.id)
            .where(*base_conditions, UserMembership.member_level_id.in_(ids))
        )
    elif target_type == "agent":
        conditions = [*base_conditions]
        if active_agents_only:
            conditions.append(UserAgent.status.in_([AgentStatus.ACTIVE, AgentStatus.OBSERVING]))
        stmt = select(User.id).join(UserAgent, UserAgent.user_id == User.id).where(*conditions)
    else:
        return []

    res = await db.execute(stmt.distinct())
    return [int(uid) for uid in res.scalars().all()]


async def broadcast_notification(
    db: AsyncSession,
    *,
    target_type: str,
    title: str,
    content: str,
    category: str,
    level: str = "info",
    sender_id: int | None = None,
    sender_name: str | None = None,
    sender_type: str | None = "admin",
    target_user_ids: list[int] | None = None,
    target_role: str | None = None,
    target_member_level_ids: list[int] | None = None,
    active_agents_only: bool = True,
    action_label: str | None = None,
    action_url: str | None = None,
) -> dict[str, Any]:
    user_ids = await resolve_broadcast_user_ids(
        db,
        target_type=target_type,
        target_user_ids=target_user_ids,
        target_role=target_role,
        target_member_level_ids=target_member_level_ids,
        active_agents_only=active_agents_only,
    )
    sent = await notify_many(
        db,
        user_ids=user_ids,
        category=category,
        title=title,
        content=content,
        link=action_url,
        level=level,
        sender_id=sender_id,
        sender_name=sender_name,
        sender_type=sender_type,
        action_label=action_label,
        action_url=action_url,
    )
    return {"matched": len(user_ids), "sent": sent}


# ── 公告 ────────────────────────────────────────────────────────────

def _visible_announcements_condition(user_id: int):
    """用户可见公告条件：已发布 且（全站 或 在指定用户列表内）。"""
    return and_(
        Announcement.is_published.is_(True),
        or_(
            Announcement.target_type == "all",
            and_(
                Announcement.target_type == "specific",
                Announcement.target_user_ids.is_not(None),
                func.json_contains(Announcement.target_user_ids, str(user_id)),
            ),
        ),
    )


def announcement_to_dict(item: Announcement, *, is_read: bool) -> dict[str, Any]:
    return {
        "id": item.id,
        "title": item.title,
        "content": item.content,
        "category": item.category,
        "sender_name": "平台公告",
        "sender_type": "announcement",
        "source_label": "平台公告",
        "is_read": is_read,
        "published_at": item.published_at.isoformat() if item.published_at else None,
        "created_at": item.created_at.isoformat() if item.created_at else None,
    }


async def list_announcements(
    db: AsyncSession,
    user_id: int,
    *,
    limit: int = 20,
    offset: int = 0,
) -> tuple[list[dict], int]:
    """用户可见公告（带 is_read），按发布时间倒序。"""
    if not await _in_app_enabled(db, user_id, "announcement"):
        return [], 0

    visible = _visible_announcements_condition(user_id)

    total_res = await db.execute(select(func.count(Announcement.id)).where(visible))
    total = int(total_res.scalar_one() or 0)

    res = await db.execute(
        select(Announcement)
        .where(visible)
        .order_by(Announcement.id.desc())
        .limit(limit)
        .offset(offset)
    )
    announcements = list(res.scalars().all())

    if not announcements:
        return [], total

    read_res = await db.execute(
        select(AnnouncementRead.announcement_id).where(
            AnnouncementRead.user_id == user_id,
            AnnouncementRead.announcement_id.in_([a.id for a in announcements]),
        )
    )
    read_ids = {int(r) for r in read_res.scalars().all()}

    items = [announcement_to_dict(a, is_read=a.id in read_ids) for a in announcements]
    return items, total


async def mark_announcement_read(db: AsyncSession, user_id: int, announcement_id: int) -> bool:
    existing = await db.execute(
        select(AnnouncementRead).where(
            AnnouncementRead.announcement_id == announcement_id,
            AnnouncementRead.user_id == user_id,
        )
    )
    if existing.scalar_one_or_none():
        return False
    db.add(
        AnnouncementRead(
            announcement_id=announcement_id,
            user_id=user_id,
            read_at=datetime.now(timezone.utc),
        )
    )
    await db.flush()
    return True


# ── 后台公告管理 ────────────────────────────────────────────────────

async def publish_announcement(
    db: AsyncSession,
    *,
    title: str,
    content: str,
    category: str = "announcement",
    target_type: str = "all",
    target_user_ids: list[int] | None = None,
    created_by: int | None = None,
) -> Announcement:
    item = Announcement(
        title=title,
        content=content,
        category=category,
        target_type=target_type,
        target_user_ids=target_user_ids if target_type == "specific" else None,
        is_published=True,
        published_at=datetime.now(timezone.utc),
        created_by=created_by,
    )
    db.add(item)
    await db.flush()
    return item


def admin_announcement_to_dict(item: Announcement) -> dict[str, Any]:
    return {
        "id": item.id,
        "title": item.title,
        "content": item.content,
        "category": item.category,
        "target_type": item.target_type,
        "target_user_ids": item.target_user_ids,
        "is_published": item.is_published,
        "published_at": item.published_at.isoformat() if item.published_at else None,
        "created_by": item.created_by,
        "created_at": item.created_at.isoformat() if item.created_at else None,
    }


async def list_all_announcements(
    db: AsyncSession,
    *,
    limit: int = 20,
    offset: int = 0,
) -> tuple[list[dict], int]:
    total_res = await db.execute(select(func.count(Announcement.id)))
    total = int(total_res.scalar_one() or 0)
    res = await db.execute(
        select(Announcement).order_by(Announcement.id.desc()).limit(limit).offset(offset)
    )
    items = [admin_announcement_to_dict(a) for a in res.scalars().all()]
    return items, total


async def delete_announcement(db: AsyncSession, announcement_id: int) -> bool:
    await db.execute(
        delete(AnnouncementRead).where(AnnouncementRead.announcement_id == announcement_id)
    )
    res = await db.execute(delete(Announcement).where(Announcement.id == announcement_id))
    await db.flush()
    return res.rowcount > 0
