"""Phone-first authentication service."""

from __future__ import annotations

from datetime import timedelta
import hashlib
import os
import random
import re
from typing import Any

from fastapi import HTTPException
from sqlalchemy import and_, desc, func, or_, select
from sqlalchemy.ext.asyncio import AsyncSession

from ..models import PhoneVerificationCode, User, UserStatus
from ..services.auth import AuthService
from ..services.auth.service import SECRET_KEY
from ..services.platform_access_service import get_platform_context
from ..services.workspace_service import (
    accept_pending_invitations_for_phone,
    ensure_personal_workspace,
    list_user_workspaces,
)
from ..utils.timezone import now_cn_naive

PHONE_CODE_TTL_MINUTES = 10
PHONE_CODE_MAX_ATTEMPTS = 5
PHONE_AUTH_PURPOSES = {"login", "register", "accept_invitation", "bind_phone"}


def normalize_phone(country_code: str | None, phone_number: str | None) -> dict[str, str]:
    code = str(country_code or "+86").strip()
    if not code.startswith("+"):
        code = f"+{code}"
    code_digits = re.sub(r"\D", "", code)
    code = f"+{code_digits}"
    number = re.sub(r"\D", "", str(phone_number or ""))
    if not code or code == "+":
        raise HTTPException(status_code=400, detail="手机号国家区号无效")
    if len(number) < 6 or len(number) > 15:
        raise HTTPException(status_code=400, detail="手机号格式无效")
    if code == "+86" and len(number) != 11:
        raise HTTPException(status_code=400, detail="中国大陆手机号需要 11 位")
    return {
        "phone_country_code": code,
        "phone_number": number,
        "phone_e164": f"{code}{number}",
    }


def mask_phone(phone_e164: str) -> str:
    raw = str(phone_e164 or "")
    if len(raw) <= 7:
        return raw
    return f"{raw[:3]}****{raw[-4:]}"


def _normalize_purpose(purpose: str | None) -> str:
    value = str(purpose or "login").strip().lower()
    return value if value in PHONE_AUTH_PURPOSES else "login"


def _hash_code(phone_e164: str, purpose: str, code: str) -> str:
    payload = f"{phone_e164}:{purpose}:{code}:{SECRET_KEY}"
    return hashlib.sha256(payload.encode("utf-8")).hexdigest()


def _generate_code() -> str:
    configured = str(os.getenv("PMFLOW_PHONE_OTP_DEV_CODE", "")).strip()
    if configured and configured.isdigit() and len(configured) == 6:
        return configured
    return f"{random.randint(0, 999999):06d}"


def _should_return_dev_code() -> bool:
    env = str(os.getenv("APP_ENV") or os.getenv("ENV") or os.getenv("NODE_ENV") or "development").lower()
    return env not in {"production", "prod"}


async def send_phone_verification_code(
    db: AsyncSession,
    *,
    country_code: str | None,
    phone_number: str | None,
    purpose: str | None = None,
    request_ip: str | None = None,
) -> dict[str, Any]:
    normalized = normalize_phone(country_code, phone_number)
    normalized_purpose = _normalize_purpose(purpose)
    code = _generate_code()
    verification = PhoneVerificationCode(
        phone_e164=normalized["phone_e164"],
        purpose=normalized_purpose,
        code_hash=_hash_code(normalized["phone_e164"], normalized_purpose, code),
        expires_at=now_cn_naive() + timedelta(minutes=PHONE_CODE_TTL_MINUTES),
        request_ip=request_ip,
    )
    db.add(verification)
    await db.commit()
    payload: dict[str, Any] = {
        "phone_e164": normalized["phone_e164"],
        "masked_phone": mask_phone(normalized["phone_e164"]),
        "purpose": normalized_purpose,
        "expires_in_seconds": PHONE_CODE_TTL_MINUTES * 60,
    }
    if _should_return_dev_code():
        payload["dev_code"] = code
    return payload


async def verify_phone_code(
    db: AsyncSession,
    *,
    country_code: str | None,
    phone_number: str | None,
    code: str,
    purpose: str | None = None,
) -> dict[str, str]:
    normalized = normalize_phone(country_code, phone_number)
    normalized_purpose = _normalize_purpose(purpose)
    clean_code = str(code or "").strip()
    if not re.fullmatch(r"\d{6}", clean_code):
        raise HTTPException(status_code=400, detail="验证码格式无效")
    verification = (
        await db.execute(
            select(PhoneVerificationCode)
            .where(
                and_(
                    PhoneVerificationCode.phone_e164 == normalized["phone_e164"],
                    PhoneVerificationCode.purpose == normalized_purpose,
                    PhoneVerificationCode.consumed_at.is_(None),
                    PhoneVerificationCode.expires_at >= now_cn_naive(),
                )
            )
            .order_by(desc(PhoneVerificationCode.created_at), desc(PhoneVerificationCode.id))
            .limit(1)
        )
    ).scalar_one_or_none()
    if not verification:
        raise HTTPException(status_code=400, detail="验证码不存在或已过期")
    if int(verification.attempt_count or 0) >= PHONE_CODE_MAX_ATTEMPTS:
        raise HTTPException(status_code=429, detail="验证码尝试次数过多，请重新获取")
    verification.attempt_count = int(verification.attempt_count or 0) + 1
    if verification.code_hash != _hash_code(normalized["phone_e164"], normalized_purpose, clean_code):
        await db.commit()
        raise HTTPException(status_code=400, detail="验证码不正确")
    verification.consumed_at = now_cn_naive()
    await db.flush()
    return normalized


def _phone_placeholder_email(phone_e164: str) -> str:
    token = re.sub(r"\D", "", phone_e164)
    return f"{token}@phone.pmflow.local"


def _phone_username(phone_e164: str) -> str:
    token = re.sub(r"\D", "", phone_e164)
    suffix = token[-4:] if token else "user"
    digest = hashlib.sha1(phone_e164.encode("utf-8")).hexdigest()[:8]
    return f"user_{suffix}_{digest}"


async def get_or_create_phone_user(
    db: AsyncSession,
    *,
    phone_country_code: str,
    phone_number: str,
    phone_e164: str,
    name: str | None = None,
) -> User:
    user = (await db.execute(select(User).where(User.phone_e164 == phone_e164))).scalar_one_or_none()
    if user:
        user.phone_country_code = phone_country_code
        user.phone_number = phone_number
        user.phone_verified_at = user.phone_verified_at or now_cn_naive()
        user.last_login_at = now_cn_naive()
        user.login_method = "phone_otp"
        await ensure_personal_workspace(db, user=user, commit=False)
        return user

    username = _phone_username(phone_e164)
    existing_username = (await db.execute(select(User.id).where(User.username == username))).scalar_one_or_none()
    if existing_username:
        username = f"{username}_{random.randint(1000, 9999)}"
    display_name = str(name or f"用户{phone_number[-4:]}").strip()
    from ..models import Tenant, TenantMembership  # avoid circular import in module load
    from .workspace_service import ACTIVE_MEMBERSHIP_STATUS, PERSONAL_WORKSPACE_TYPE

    tenant = Tenant(
        name=f"{display_name} 的个人空间",
        workspace_type=PERSONAL_WORKSPACE_TYPE,
        slug=None,
        owner_user_id=None,
        status="active",
    )
    db.add(tenant)
    await db.flush()
    user = User(
        tenant_id=int(tenant.id),
        username=username,
        password_hash=AuthService.hash_password(hashlib.sha256(os.urandom(32)).hexdigest()[:16]),
        name=display_name,
        email=_phone_placeholder_email(phone_e164),
        phone_country_code=phone_country_code,
        phone_number=phone_number,
        phone_e164=phone_e164,
        phone_verified_at=now_cn_naive(),
        last_login_at=now_cn_naive(),
        login_method="phone_otp",
        role="OWNER",
        assigned_domains=[],
        workload=0,
        status=UserStatus.ACTIVE.value,
        is_active=True,
    )
    db.add(user)
    await db.flush()
    tenant.owner_user_id = int(user.id)
    tenant.slug = f"personal-{int(user.id)}"
    db.add(
        TenantMembership(
            tenant_id=int(tenant.id),
            user_id=int(user.id),
            role="OWNER",
            status=ACTIVE_MEMBERSHIP_STATUS,
            is_personal_owner=True,
        )
    )
    return user


async def verify_phone_and_login(
    db: AsyncSession,
    *,
    country_code: str | None,
    phone_number: str | None,
    code: str,
    purpose: str | None = "login",
    invite_token: str | None = None,
    name: str | None = None,
) -> dict[str, Any]:
    normalized = await verify_phone_code(
        db,
        country_code=country_code,
        phone_number=phone_number,
        code=code,
        purpose=purpose,
    )
    user = await get_or_create_phone_user(db, name=name, **normalized)
    accepted = []
    if invite_token:
        from .workspace_service import accept_invitation_by_verified_phone

        accepted.append(
            await accept_invitation_by_verified_phone(
                db,
                token=invite_token,
                user=user,
                phone_e164=normalized["phone_e164"],
                switch_to_workspace=True,
            )
        )
    else:
        accepted = await accept_pending_invitations_for_phone(
            db,
            user=user,
            phone_e164=normalized["phone_e164"],
            auto_accept=False,
        )
    await db.commit()
    await db.refresh(user)
    access_token = AuthService.create_access_token(data={"sub": user.username})
    workspaces = await list_user_workspaces(db, user=user)
    platform_context = await get_platform_context(db, user=user)
    return {
        "access_token": access_token,
        "token_type": "bearer",
        "user": {**user.to_dict(), "workspace_context": workspaces, "platform_context": platform_context},
        "pending_invitations": accepted,
    }