from __future__ import annotations

from datetime import datetime
from decimal import Decimal
from collections import defaultdict
import hashlib
import re
from typing import Any, Dict, List, Optional

from flask import Flask, current_app
from sqlalchemy import case

from flaskr.dao import db
from flaskr.service.common.dtos import PageNationDTO
from flaskr.service.common.models import (
    AppException,
    raise_error,
    raise_error_with_args,
    raise_param_error,
)
from flaskr.i18n import _
from flaskr.service.order.admin_dtos import (
    OrderAdminActivityDTO,
    OrderAdminCouponDTO,
    OrderAdminDetailDTO,
    OrderAdminOverviewDTO,
    OrderAdminPaymentDTO,
    OrderAdminSummaryDTO,
)
from flaskr.service.order.funs import init_buy_record, success_buy_record
from flaskr.service.order.consts import (
    ORDER_STATUS_INIT,
    ORDER_STATUS_REFUND,
    ORDER_STATUS_SUCCESS,
    ORDER_STATUS_TIMEOUT,
    ORDER_STATUS_TO_BE_PAID,
)
from flaskr.service.order.models import (
    Order,
    PingxxOrder,
    StripeOrder,
)
from flaskr.service.order.raw_snapshots import (
    legacy_native_snapshot_query,
    legacy_pingxx_snapshot_query,
    legacy_stripe_snapshot_query,
    native_snapshot_model,
)
from flaskr.service.promo.consts import (
    COUPON_STATUS_ACTIVE,
    COUPON_STATUS_INACTIVE,
    COUPON_STATUS_TIMEOUT,
    COUPON_STATUS_USED,
    COUPON_TYPE_FIXED,
    COUPON_TYPE_PERCENT,
    PROMO_CAMPAIGN_APPLICATION_STATUS_APPLIED,
    PROMO_CAMPAIGN_APPLICATION_STATUS_VOIDED,
)
from flaskr.service.promo.models import CouponUsage, PromoRedemption
from flaskr.service.shifu.models import DraftShifu, PublishedShifu
from flaskr.service.shifu.shifu_draft_funcs import get_user_created_shifu_bids
from flaskr.service.shifu.utils import get_shifu_creator_bid
from flaskr.service.user.models import AuthCredential, UserInfo as UserEntity
from flaskr.service.user.repository import (
    ensure_user_for_identifier,
    get_user_entity_by_bid,
    load_user_aggregate_by_identifier,
    update_user_entity_fields,
    upsert_credential,
)
from flaskr.service.common.phone_numbers import normalize_phone_identifier
from flaskr.service.user.utils import ensure_demo_course_permissions
from flaskr.service.user.consts import USER_STATE_REGISTERED, USER_STATE_UNREGISTERED
from flaskr.util.timezone import serialize_with_app_timezone


ORDER_STATUS_KEY_MAP = {
    ORDER_STATUS_INIT: "server.order.orderStatusInit",
    ORDER_STATUS_SUCCESS: "server.order.orderStatusSuccess",
    ORDER_STATUS_REFUND: "server.order.orderStatusRefund",
    ORDER_STATUS_TO_BE_PAID: "server.order.orderStatusToBePaid",
    ORDER_STATUS_TIMEOUT: "server.order.orderStatusTimeout",
}

PAYMENT_STATUS_KEY_MAP = {
    0: "module.order.paymentStatus.pending",
    1: "module.order.paymentStatus.paid",
    2: "module.order.paymentStatus.refunded",
    3: "module.order.paymentStatus.closed",
    4: "module.order.paymentStatus.failed",
}

IMPORT_ACTIVATION_MOBILE_PATTERN = re.compile(r"(?<!\d)\d{11}(?!\d)")
IMPORT_ACTIVATION_TEXT_PATTERN = re.compile(r"[A-Za-z\u4E00-\u9FFF]")
IMPORT_ACTIVATION_MAX_TEXT_LENGTH = 10_000

ACTIVE_STATUS_KEY_MAP = {
    PROMO_CAMPAIGN_APPLICATION_STATUS_APPLIED: "module.order.activeStatus.active",
    PROMO_CAMPAIGN_APPLICATION_STATUS_VOIDED: "module.order.activeStatus.failed",
}

COUPON_STATUS_KEY_MAP = {
    COUPON_STATUS_INACTIVE: "module.order.couponStatus.inactive",
    COUPON_STATUS_ACTIVE: "module.order.couponStatus.active",
    COUPON_STATUS_USED: "module.order.couponStatus.used",
    COUPON_STATUS_TIMEOUT: "module.order.couponStatus.timeout",
}

COUPON_TYPE_KEY_MAP = {
    COUPON_TYPE_FIXED: "module.order.couponType.fixed",
    COUPON_TYPE_PERCENT: "module.order.couponType.percent",
}

PAYMENT_CHANNEL_KEY_MAP = {
    "pingxx": "module.order.paymentChannel.pingxx",
    "stripe": "module.order.paymentChannel.stripe",
    "alipay": "module.order.paymentChannel.alipay",
    "wechatpay": "module.order.paymentChannel.wechatpay",
    "manual": "module.order.paymentChannel.manual",
    "open_api": "module.order.paymentChannel.open_api",
}

ORDER_SOURCE_USER_PURCHASE = "user_purchase"
ORDER_SOURCE_COUPON_REDEEM = "coupon_redeem"
ORDER_SOURCE_IMPORT_ACTIVATION = "import_activation"
ORDER_SOURCE_OPEN_API = "open_api"

ORDER_SOURCE_KEY_MAP = {
    ORDER_SOURCE_USER_PURCHASE: "module.operationsOrder.source.userPurchase",
    ORDER_SOURCE_COUPON_REDEEM: "module.operationsOrder.source.couponRedeem",
    ORDER_SOURCE_IMPORT_ACTIVATION: "module.operationsOrder.source.importActivation",
    ORDER_SOURCE_OPEN_API: "module.operationsOrder.source.openApi",
}

MOBILE_PATTERN = re.compile(r"^\d{11}$")
EMAIL_PATTERN = re.compile(r"^[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}$")


def normalize_mobile(mobile: str) -> str:
    """Normalize and validate a mobile number (11 digits)."""
    normalized_mobile = normalize_phone_identifier(mobile)
    if not normalized_mobile:
        raise_param_error("mobile")
    if not MOBILE_PATTERN.fullmatch(normalized_mobile):
        raise_param_error(f"mobile format invalid: {normalized_mobile}")
    return normalized_mobile


def normalize_email(email: str) -> str:
    """Normalize and validate an email address."""
    normalized_email = str(email or "").strip().lower()
    if not normalized_email:
        raise_param_error("email")
    if not EMAIL_PATTERN.fullmatch(normalized_email):
        raise_param_error("email format invalid")
    return normalized_email


def _mask_contact_identifier(identifier: str) -> str:
    """Return a non-reversible hash label for contact identifiers."""
    if not identifier:
        return ""
    digest = hashlib.sha256(identifier.encode("utf-8")).hexdigest()[:8]
    return f"hash:{digest}"


def _log_error_code(exc: AppException) -> str:
    """Return a safe error identifier for logs without leaking PII."""
    if getattr(exc, "code", None) is not None:
        return str(exc.code)
    return exc.__class__.__name__


def normalize_contact_identifier(identifier: str, contact_type: str) -> str:
    """Normalize and validate phone/email identifiers."""
    if contact_type == "email":
        return normalize_email(identifier)
    if contact_type == "phone":
        return normalize_mobile(identifier)
    raise_param_error("contact_type")


def _format_decimal(value: Optional[Decimal]) -> str:
    """Format a Decimal or numeric string to trimmed two-decimal string."""
    if value is None:
        return "0"
    if isinstance(value, str):
        normalized = value
    else:
        normalized = "{0:.2f}".format(value)
    if normalized.endswith(".00"):
        return normalized[:-3]
    return normalized


def _format_cents(value: Optional[int]) -> str:
    """Convert cents integer to string representation in units."""
    if value is None:
        return "0"
    try:
        return _format_decimal(Decimal(value) / Decimal(100))
    except (ArithmeticError, ValueError):
        return "0"


def _format_admin_datetime(value: Optional[datetime]) -> str:
    """Serialize admin/operator datetimes as UTC ISO strings."""
    if not value:
        return ""
    serialized_value = serialize_with_app_timezone(
        current_app._get_current_object(),
        value,
        tz_name="UTC",
    )
    return str(serialized_value or "").replace("+00:00", "Z")


def _parse_datetime(value: str, is_end: bool = False) -> Optional[datetime]:
    """Parse date/time string with multiple formats; auto fill day bounds."""
    if not value:
        return None
    normalized = str(value).strip()
    if not normalized:
        return None
    for fmt in ("%Y-%m-%d", "%Y-%m-%d %H:%M:%S", "%Y-%m-%dT%H:%M", "%Y-%m-%dT%H:%M:%S"):
        try:
            parsed = datetime.strptime(normalized, fmt)
            if fmt == "%Y-%m-%d":
                if is_end:
                    parsed = parsed.replace(hour=23, minute=59, second=59)
                else:
                    parsed = parsed.replace(hour=0, minute=0, second=0)
            return parsed
        except ValueError:
            continue
    return None


def _trim_import_activation_nickname(value: str) -> str:
    """Trim nickname content around non-text separators."""
    text = str(value or "").strip()
    if not text:
        return ""
    start = 0
    end = len(text)
    while start < end and not IMPORT_ACTIVATION_TEXT_PATTERN.match(text[start]):
        start += 1
    while end > start and not IMPORT_ACTIVATION_TEXT_PATTERN.match(text[end - 1]):
        end -= 1
    return text[start:end].strip()


def parse_import_activation_entries(
    text: str, contact_type: str = "phone"
) -> List[Dict[str, str]]:
    """Parse contact identifiers and optional nicknames from raw text."""
    if not text:
        return []
    safe_text = str(text)
    if len(safe_text) > IMPORT_ACTIVATION_MAX_TEXT_LENGTH:
        safe_text = safe_text[:IMPORT_ACTIVATION_MAX_TEXT_LENGTH]
    if contact_type == "email":
        return _parse_import_activation_emails(safe_text)
    return _parse_import_activation_mobiles(safe_text)


def _parse_import_activation_mobiles(text: str) -> List[Dict[str, str]]:
    """Parse phone identifiers and optional nicknames from raw text."""
    matches = list(IMPORT_ACTIVATION_MOBILE_PATTERN.finditer(text))
    if not matches:
        return []
    entries: List[Dict[str, str]] = []
    for index, match in enumerate(matches):
        start = match.start()
        end = matches[index + 1].start() if index + 1 < len(matches) else len(text)
        segment = text[start:end]
        identifier = match.group(0)
        nickname_source = segment.replace(identifier, "", 1)
        nickname = _trim_import_activation_nickname(nickname_source)
        entries.append({"mobile": identifier, "nickname": nickname})
    return entries


def _parse_import_activation_emails(text: str) -> List[Dict[str, str]]:
    """Parse email identifiers and optional nicknames from raw text."""
    entries: List[Dict[str, str]] = []
    for raw_line in text.splitlines():
        line = raw_line.strip()
        if not line:
            continue
        matches = _find_email_matches(line)
        if not matches:
            continue

        for index, (start, _end, identifier) in enumerate(matches):
            next_start = (
                matches[index + 1][0] if index + 1 < len(matches) else len(line)
            )
            segment = line[start:next_start]
            nickname_source = segment.replace(identifier, "", 1)
            nickname = _trim_import_activation_nickname(nickname_source)
            entries.append({"mobile": identifier, "nickname": nickname})
    return entries


def _find_email_matches(line: str) -> List[tuple[int, int, str]]:
    """Find email candidates in a line using linear scanning."""
    if "@" not in line:
        return []
    allowed = set("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789._%+-")
    matches: List[tuple[int, int, str]] = []
    seen: set[tuple[int, int]] = set()
    length = len(line)
    for index, char in enumerate(line):
        if char != "@":
            continue
        left = index - 1
        while left >= 0 and line[left] in allowed:
            left -= 1
        right = index + 1
        while right < length and line[right] in allowed:
            right += 1
        start = left + 1
        end = right
        if end - start <= 3:
            continue
        if (start, end) in seen:
            continue
        candidate = line[start:end]
        if EMAIL_PATTERN.fullmatch(candidate):
            matches.append((start, end, candidate))
            seen.add((start, end))
    matches.sort(key=lambda item: item[0])
    return matches


def _load_shifu_map(
    shifu_bids: list[str],
) -> Dict[str, DraftShifu | PublishedShifu]:
    """Load shifu records for given bids, preferring published with draft fallback."""
    if not shifu_bids:
        return {}
    shifu_map: Dict[str, DraftShifu | PublishedShifu] = {}

    published_shifus = (
        PublishedShifu.query.filter(
            PublishedShifu.shifu_bid.in_(shifu_bids),
            PublishedShifu.deleted == 0,
        )
        .order_by(PublishedShifu.id.desc())
        .all()
    )
    for shifu in published_shifus:
        if shifu.shifu_bid and shifu.shifu_bid not in shifu_map:
            shifu_map[shifu.shifu_bid] = shifu

    missing_bids = [bid for bid in shifu_bids if bid not in shifu_map]
    if not missing_bids:
        return shifu_map

    shifu_drafts = (
        DraftShifu.query.filter(
            DraftShifu.shifu_bid.in_(missing_bids),
            DraftShifu.deleted == 0,
        )
        .order_by(DraftShifu.id.desc())
        .all()
    )
    for shifu in shifu_drafts:
        if shifu.shifu_bid and shifu.shifu_bid not in shifu_map:
            shifu_map[shifu.shifu_bid] = shifu

    return shifu_map


def _load_user_map(user_bids: list[str]) -> Dict[str, Dict[str, str]]:
    """Load user mobile/nickname info for given user bids."""
    if not user_bids:
        return {}
    credentials = (
        AuthCredential.query.filter(
            AuthCredential.user_bid.in_(user_bids),
            AuthCredential.provider_name.in_(["phone", "email"]),
        )
        .order_by(AuthCredential.id.desc())
        .all()
    )
    phone_map: Dict[str, str] = {}
    email_map: Dict[str, str] = {}
    for credential in credentials:
        if not credential.user_bid:
            continue
        if credential.provider_name == "phone":
            if credential.user_bid not in phone_map:
                phone_map[credential.user_bid] = credential.identifier or ""
        if credential.provider_name == "email":
            if credential.user_bid not in email_map:
                email_map[credential.user_bid] = credential.identifier or ""

    users = UserEntity.query.filter(UserEntity.user_bid.in_(user_bids)).all()
    user_map: Dict[str, Dict[str, str]] = {}
    for user in users:
        mobile = phone_map.get(user.user_bid, "")
        email = email_map.get(user.user_bid, "")
        if not mobile and (user.user_identify or "").isdigit():
            mobile = user.user_identify
        if not email and "@" in (user.user_identify or ""):
            email = user.user_identify
        user_map[user.user_bid] = {
            "mobile": mobile or "",
            "email": email or "",
            "nickname": user.nickname or "",
            "identify": user.user_identify or "",
        }
    return user_map


def _load_coupon_code_map(order_bids: list[str]) -> Dict[str, List[str]]:
    """Load coupon codes for given orders and map by order bid."""
    if not order_bids:
        return {}
    records = (
        CouponUsage.query.filter(
            CouponUsage.order_bid.in_(order_bids),
            CouponUsage.deleted == 0,
        )
        .order_by(CouponUsage.id.desc())
        .all()
    )
    coupon_map: Dict[str, List[str]] = defaultdict(list)
    for record in records:
        order_bid = record.order_bid or ""
        code = record.code or ""
        if not order_bid or not code:
            continue
        if code in coupon_map[order_bid]:
            continue
        coupon_map[order_bid].append(code)
    return dict(coupon_map)


def _build_coupon_usage_order_bid_subquery():
    return (
        db.session.query(CouponUsage.order_bid.label("order_bid"))
        .filter(
            CouponUsage.deleted == 0,
            CouponUsage.order_bid != "",
        )
        .distinct()
        .subquery()
    )


def _resolve_order_source(
    *,
    payment_channel: str,
    coupon_codes: List[str],
    paid_price: Decimal | str | None,
) -> tuple[str, str]:
    normalized_payment_channel = str(payment_channel or "").strip()
    if normalized_payment_channel == "manual":
        return ORDER_SOURCE_IMPORT_ACTIVATION, ORDER_SOURCE_KEY_MAP[
            ORDER_SOURCE_IMPORT_ACTIVATION
        ]
    if normalized_payment_channel == "open_api":
        return ORDER_SOURCE_OPEN_API, ORDER_SOURCE_KEY_MAP[ORDER_SOURCE_OPEN_API]

    normalized_paid_price = Decimal(str(paid_price or 0))
    if coupon_codes and normalized_paid_price == Decimal("0"):
        return ORDER_SOURCE_COUPON_REDEEM, ORDER_SOURCE_KEY_MAP[
            ORDER_SOURCE_COUPON_REDEEM
        ]

    return ORDER_SOURCE_USER_PURCHASE, ORDER_SOURCE_KEY_MAP[ORDER_SOURCE_USER_PURCHASE]


def _load_matching_user_bids_for_keyword(keyword: str) -> List[str]:
    normalized_keyword = str(keyword or "").strip()
    if not normalized_keyword:
        return []

    matched_user_bids = set()
    for row in (
        UserEntity.query.filter(
            db.or_(
                UserEntity.user_bid == normalized_keyword,
                UserEntity.user_identify == normalized_keyword,
            )
        )
        .yield_per(200)
        .enable_eagerloads(False)
    ):
        user_bid = str(row.user_bid or "").strip()
        if user_bid:
            matched_user_bids.add(user_bid)

    normalized_credential_identifier = (
        normalized_keyword.lower()
        if EMAIL_PATTERN.fullmatch(normalized_keyword)
        else normalized_keyword
    )
    for row in (
        AuthCredential.query.filter(
            AuthCredential.identifier == normalized_credential_identifier,
            AuthCredential.provider_name.in_(["phone", "email", "google"]),
            AuthCredential.deleted == 0,
        )
        .yield_per(200)
        .enable_eagerloads(False)
    ):
        user_bid = str(row.user_bid or "").strip()
        if user_bid:
            matched_user_bids.add(user_bid)

    return sorted(matched_user_bids)


def _load_matching_shifu_bids_for_course_name(course_name: str) -> List[str]:
    normalized_course_name = str(course_name or "").strip()
    if not normalized_course_name:
        return []

    like_value = f"%{normalized_course_name}%"
    matched_shifu_bids = set()
    for row in (
        DraftShifu.query.filter(
            DraftShifu.deleted == 0,
            DraftShifu.title.like(like_value),
        )
        .yield_per(200)
        .enable_eagerloads(False)
    ):
        shifu_bid = str(row.shifu_bid or "").strip()
        if shifu_bid:
            matched_shifu_bids.add(shifu_bid)

    for row in (
        PublishedShifu.query.filter(
            PublishedShifu.deleted == 0,
            PublishedShifu.title.like(like_value),
        )
        .yield_per(200)
        .enable_eagerloads(False)
    ):
        shifu_bid = str(row.shifu_bid or "").strip()
        if shifu_bid:
            matched_shifu_bids.add(shifu_bid)

    return sorted(matched_shifu_bids)


def _apply_order_source_filter(query, order_source: str):
    normalized_order_source = str(order_source or "").strip()
    if not normalized_order_source:
        return query

    coupon_usage_order_bid_subquery = _build_coupon_usage_order_bid_subquery()
    coupon_order_bid_query = db.session.query(
        coupon_usage_order_bid_subquery.c.order_bid
    )
    non_special_payment_channel = db.or_(
        Order.payment_channel.is_(None),
        Order.payment_channel.notin_(["manual", "open_api"]),
    )

    if normalized_order_source == ORDER_SOURCE_IMPORT_ACTIVATION:
        return query.filter(Order.payment_channel == "manual")

    if normalized_order_source == ORDER_SOURCE_OPEN_API:
        return query.filter(Order.payment_channel == "open_api")

    if normalized_order_source == ORDER_SOURCE_COUPON_REDEEM:
        return query.filter(
            non_special_payment_channel,
            Order.order_bid.in_(coupon_order_bid_query),
            Order.paid_price == Decimal("0"),
        )

    if normalized_order_source == ORDER_SOURCE_USER_PURCHASE:
        return query.filter(
            non_special_payment_channel,
        ).filter(
            db.not_(
                db.and_(
                    Order.order_bid.in_(coupon_order_bid_query),
                    Order.paid_price == Decimal("0"),
                )
            )
        )

    return query.filter(db.literal(False))


def _build_order_item(
    order: Order,
    shifu_map: Dict[str, DraftShifu | PublishedShifu],
    user_map: Dict[str, Dict[str, str]],
    coupon_map: Optional[Dict[str, List[str]]] = None,
) -> OrderAdminSummaryDTO:
    """Build admin order summary DTO from order plus shifu/user lookups."""
    shifu = shifu_map.get(order.shifu_bid)
    user = user_map.get(order.user_bid, {})
    payment_channel = order.payment_channel or ""
    status_key = ORDER_STATUS_KEY_MAP.get(order.status, "server.order.orderStatusInit")
    coupon_codes = []
    if coupon_map is not None:
        coupon_codes = coupon_map.get(order.order_bid, []) or []
    order_source, order_source_key = _resolve_order_source(
        payment_channel=payment_channel,
        coupon_codes=coupon_codes,
        paid_price=order.paid_price,
    )
    return OrderAdminSummaryDTO(
        order_bid=order.order_bid,
        shifu_bid=order.shifu_bid,
        shifu_name=shifu.title if shifu else "",
        user_bid=order.user_bid,
        user_mobile=user.get("mobile", ""),
        user_email=user.get("email", ""),
        user_nickname=user.get("nickname", ""),
        payable_price=_format_decimal(order.payable_price),
        paid_price=_format_decimal(order.paid_price),
        discount_amount=_format_decimal(
            Decimal(order.payable_price or 0) - Decimal(order.paid_price or 0)
        ),
        status=order.status,
        status_key=status_key,
        payment_channel=payment_channel,
        payment_channel_key=PAYMENT_CHANNEL_KEY_MAP.get(
            payment_channel, "module.order.paymentChannel.unknown"
        ),
        order_source=order_source,
        order_source_key=order_source_key,
        coupon_codes=coupon_codes,
        created_at=_format_admin_datetime(order.created_at),
        updated_at=_format_admin_datetime(order.updated_at),
    )


def import_activation_order(
    app: Flask,
    mobile: str,
    course_id: str,
    user_nick_name: Optional[str] = None,
    contact_type: str = "phone",
    allow_empty_nickname: bool = False,
    payment_channel: str = "manual",
) -> Dict[str, str]:
    """Create activation order for a user identified by phone or email."""
    normalized_identifier = normalize_contact_identifier(mobile, contact_type)

    normalized_course_id = str(course_id or "").strip()
    if not normalized_course_id:
        raise_param_error("course_id")

    normalized_nickname = str(user_nick_name or "").strip()
    nickname_value = (
        normalized_nickname
        if allow_empty_nickname
        else normalized_nickname or normalized_identifier
    )
    defaults = {
        "identify": normalized_identifier,
        "nickname": nickname_value,
        "state": USER_STATE_REGISTERED,
    }
    existing_aggregate = load_user_aggregate_by_identifier(
        normalized_identifier, providers=[contact_type]
    )
    aggregate, created_new_user = ensure_user_for_identifier(
        app,
        provider=contact_type,
        identifier=normalized_identifier,
        defaults=defaults,
    )

    if not aggregate:
        raise_error("server.user.userNotFound")

    user_id = aggregate.user_bid

    existing_success_order = (
        Order.query.filter(
            Order.user_bid == user_id,
            Order.shifu_bid == normalized_course_id,
            Order.status == ORDER_STATUS_SUCCESS,
            Order.deleted == 0,
        )
        .order_by(Order.id.desc())
        .first()
    )
    if existing_success_order:
        if contact_type == "email":
            raise_error_with_args(
                "server.order.emailAlreadyActivated", email=normalized_identifier
            )
        else:
            raise_error_with_args(
                "server.order.mobileAlreadyActivated",
                mobile=normalized_identifier,
            )

    entity = get_user_entity_by_bid(user_id, include_deleted=True)
    if entity:
        updates = {"identify": normalized_identifier}
        if normalized_nickname:
            updates["nickname"] = normalized_nickname
        if aggregate.state == USER_STATE_UNREGISTERED:
            updates["state"] = USER_STATE_REGISTERED
        update_user_entity_fields(entity, **updates)
    if created_new_user or (
        existing_aggregate and existing_aggregate.state == USER_STATE_UNREGISTERED
    ):
        ensure_demo_course_permissions(app, user_id)

    upsert_credential(
        app,
        user_bid=user_id,
        provider_name=contact_type,
        subject_id=normalized_identifier,
        subject_format=contact_type,
        identifier=normalized_identifier,
        metadata={"course_id": normalized_course_id},
        verified=True,
    )
    db.session.commit()

    buy_record = init_buy_record(app, user_id, normalized_course_id)
    order = Order.query.filter(Order.order_bid == buy_record.order_id).first()
    if not order:
        raise_error("server.order.orderNotFound")

    order.payable_price = Decimal("0")
    order.paid_price = Decimal("0")
    order.payment_channel = payment_channel
    db.session.commit()

    success_buy_record(app, order.order_bid)

    return {"order_bid": order.order_bid}


def import_activation_orders(
    app: Flask,
    mobiles: List[str],
    course_id: str,
    user_nick_name: Optional[str] = None,
    contact_type: str = "phone",
) -> Dict[str, Any]:
    """Bulk import activation orders from a list of phone/email identifiers."""
    results: Dict[str, Any] = {"success": [], "failed": []}
    for mobile in mobiles:
        normalized_mobile = str(mobile or "").strip()
        try:
            order = import_activation_order(
                app,
                normalized_mobile,
                course_id,
                user_nick_name,
                contact_type=contact_type,
            )
            results["success"].append({"mobile": normalized_mobile, **order})
        except AppException as exc:
            if hasattr(app, "logger"):
                masked_identifier = _mask_contact_identifier(normalized_mobile)
                app.logger.warning(
                    "import activation failed for %s (code=%s)",
                    masked_identifier,
                    _log_error_code(exc),
                )
            results["failed"].append(
                {"mobile": normalized_mobile, "message": exc.message}
            )
        except Exception:  # noqa: BLE001
            if hasattr(app, "logger"):
                masked_identifier = _mask_contact_identifier(normalized_mobile)
                app.logger.exception(
                    "import activation unexpected failure for %s",
                    masked_identifier,
                )
            results["failed"].append(
                {
                    "mobile": normalized_mobile,
                    "message": _("server.common.unknownError"),
                }
            )
    return results


def import_activation_orders_from_entries(
    app: Flask,
    entries: List[Dict[str, str]],
    course_id: str,
    contact_type: str = "phone",
) -> Dict[str, Any]:
    """Bulk import activation orders from parsed phone/email+nickname entries."""
    results: Dict[str, Any] = {"success": [], "failed": []}
    for entry in entries:
        normalized_mobile = str(entry.get("mobile", "")).strip()
        if not normalized_mobile:
            continue
        nickname = entry.get("nickname") or ""
        try:
            order = import_activation_order(
                app,
                normalized_mobile,
                course_id,
                nickname,
                contact_type=contact_type,
                allow_empty_nickname=True,
            )
            results["success"].append({"mobile": normalized_mobile, **order})
        except AppException as exc:
            if hasattr(app, "logger"):
                masked_identifier = _mask_contact_identifier(normalized_mobile)
                app.logger.warning(
                    "import activation failed for %s (code=%s)",
                    masked_identifier,
                    _log_error_code(exc),
                )
            results["failed"].append(
                {"mobile": normalized_mobile, "message": exc.message}
            )
        except Exception:  # noqa: BLE001
            if hasattr(app, "logger"):
                masked_identifier = _mask_contact_identifier(normalized_mobile)
                app.logger.exception(
                    "import activation unexpected failure for %s",
                    masked_identifier,
                )
            results["failed"].append(
                {
                    "mobile": normalized_mobile,
                    "message": _("server.common.unknownError"),
                }
            )
    return results


def list_orders(
    app: Flask,
    user_id: str,
    page_index: int,
    page_size: int,
    filters: Optional[Dict[str, Any]] = None,
) -> PageNationDTO:
    """List orders visible to the current operator with optional filters."""
    with app.app_context():
        page_index = max(page_index, 1)
        page_size = max(page_size, 1)
        filters = filters or {}

        shifu_bids = get_user_created_shifu_bids(app, user_id)
        if not shifu_bids:
            return PageNationDTO(page_index, page_size, 0, [])

        query = Order.query.filter(
            Order.deleted == 0,
            Order.shifu_bid.in_(shifu_bids),
        )

        order_bid = filters.get("order_bid")
        if order_bid:
            query = query.filter(Order.order_bid == order_bid)

        user_bid = filters.get("user_bid")
        if user_bid:
            identify_value = str(user_bid).strip()
            if identify_value:
                matched_user_bids = [
                    user.user_bid
                    for user in UserEntity.query.filter(
                        UserEntity.user_identify == identify_value
                    ).all()
                ]
                if matched_user_bids:
                    query = query.filter(Order.user_bid.in_(matched_user_bids))
                else:
                    query = query.filter(Order.user_bid == identify_value)

        shifu_bid = filters.get("shifu_bid")
        if shifu_bid:
            if isinstance(shifu_bid, list):
                shifu_bid_list = [
                    str(bid).strip() for bid in shifu_bid if str(bid).strip()
                ]
            else:
                shifu_bid_list = [
                    bid.strip() for bid in str(shifu_bid).split(",") if bid.strip()
                ]
            if shifu_bid_list:
                allowed_bids = [bid for bid in shifu_bid_list if bid in shifu_bids]
                if not allowed_bids:
                    return PageNationDTO(page_index, page_size, 0, [])
                query = query.filter(Order.shifu_bid.in_(allowed_bids))

        status = filters.get("status")
        if status is not None and str(status).isdigit():
            query = query.filter(Order.status == int(status))

        payment_channel = filters.get("payment_channel")
        if payment_channel:
            query = query.filter(Order.payment_channel == payment_channel)

        start_time = _parse_datetime(filters.get("start_time", ""))
        if start_time:
            query = query.filter(Order.created_at >= start_time)

        end_time = _parse_datetime(filters.get("end_time", ""), is_end=True)
        if end_time:
            query = query.filter(Order.created_at <= end_time)

        total = query.count()
        orders = (
            query.order_by(Order.created_at.desc())
            .offset((page_index - 1) * page_size)
            .limit(page_size)
            .all()
        )

        shifu_map = _load_shifu_map([order.shifu_bid for order in orders])
        user_map = _load_user_map([order.user_bid for order in orders])
        coupon_map = _load_coupon_code_map([order.order_bid for order in orders])

        items = [
            _build_order_item(order, shifu_map, user_map, coupon_map)
            for order in orders
        ]
        return PageNationDTO(page_index, page_size, total, items)


def list_operator_orders(
    app: Flask,
    page_index: int,
    page_size: int,
    filters: Optional[Dict[str, Any]] = None,
) -> PageNationDTO:
    """List global orders for operator views with cross-course filters."""
    with app.app_context():
        page_index = max(page_index, 1)
        page_size = max(page_size, 1)
        filters = filters or {}

        query = Order.query.filter(Order.deleted == 0)

        order_bid = str(filters.get("order_bid", "") or "").strip()
        if order_bid:
            query = query.filter(Order.order_bid == order_bid)

        user_keyword = str(filters.get("user_keyword", "") or "").strip()
        if user_keyword:
            matched_user_bids = _load_matching_user_bids_for_keyword(user_keyword)
            if matched_user_bids:
                query = query.filter(Order.user_bid.in_(matched_user_bids))
            else:
                query = query.filter(Order.user_bid == user_keyword)

        shifu_bid = str(filters.get("shifu_bid", "") or "").strip()
        if shifu_bid:
            query = query.filter(Order.shifu_bid == shifu_bid)

        course_name = str(filters.get("course_name", "") or "").strip()
        if course_name:
            matched_shifu_bids = _load_matching_shifu_bids_for_course_name(course_name)
            if not matched_shifu_bids:
                return PageNationDTO(page_index, page_size, 0, [])
            query = query.filter(Order.shifu_bid.in_(matched_shifu_bids))

        status = filters.get("status")
        if status is not None and str(status).isdigit():
            query = query.filter(Order.status == int(status))

        payment_channel = str(filters.get("payment_channel", "") or "").strip()
        if payment_channel:
            query = query.filter(Order.payment_channel == payment_channel)

        order_source = str(filters.get("order_source", "") or "").strip()
        if order_source:
            query = _apply_order_source_filter(query, order_source)

        start_time = _parse_datetime(str(filters.get("start_time", "") or ""))
        if start_time:
            query = query.filter(Order.created_at >= start_time)

        end_time = _parse_datetime(
            str(filters.get("end_time", "") or ""),
            is_end=True,
        )
        if end_time:
            query = query.filter(Order.created_at <= end_time)

        total = query.count()
        orders = (
            query.order_by(Order.created_at.desc(), Order.id.desc())
            .offset((page_index - 1) * page_size)
            .limit(page_size)
            .all()
        )

        shifu_map = _load_shifu_map([order.shifu_bid for order in orders])
        user_map = _load_user_map([order.user_bid for order in orders])
        coupon_map = _load_coupon_code_map([order.order_bid for order in orders])

        items = [
            _build_order_item(order, shifu_map, user_map, coupon_map)
            for order in orders
        ]
        return PageNationDTO(page_index, page_size, total, items)


def get_operator_order_overview(app: Flask) -> OrderAdminOverviewDTO:
    """Return aggregate metrics for operator learning orders."""
    with app.app_context():
        summary = (
            db.session.query(
                db.func.count(Order.id).label("total_order_count"),
                db.func.coalesce(
                    db.func.sum(
                        case((Order.status == ORDER_STATUS_SUCCESS, 1), else_=0)
                    ),
                    0,
                ).label("paid_order_count"),
                db.func.coalesce(
                    db.func.sum(
                        case((Order.status == ORDER_STATUS_TO_BE_PAID, 1), else_=0)
                    ),
                    0,
                ).label("pending_order_count"),
                db.func.coalesce(
                    db.func.sum(
                        case((Order.status == ORDER_STATUS_REFUND, 1), else_=0)
                    ),
                    0,
                ).label("refunded_order_count"),
                db.func.coalesce(
                    db.func.sum(
                        case((Order.status == ORDER_STATUS_TIMEOUT, 1), else_=0)
                    ),
                    0,
                ).label("closed_order_count"),
                db.func.coalesce(
                    db.func.sum(
                        case(
                            (Order.status == ORDER_STATUS_SUCCESS, Order.paid_price),
                            else_=0,
                        )
                    ),
                    0,
                ).label("paid_amount_total"),
            )
            .filter(Order.deleted == 0)
            .one()
        )
        return OrderAdminOverviewDTO(
            total_order_count=int(summary.total_order_count or 0),
            paid_order_count=int(summary.paid_order_count or 0),
            pending_order_count=int(summary.pending_order_count or 0),
            refunded_order_count=int(summary.refunded_order_count or 0),
            closed_order_count=int(summary.closed_order_count or 0),
            paid_amount_total=_format_decimal(summary.paid_amount_total),
        )


def _load_order_activities(order_bid: str) -> List[OrderAdminActivityDTO]:
    """Load activity records tied to an order and format as DTOs."""
    records = PromoRedemption.query.filter(
        PromoRedemption.order_bid == order_bid,
        PromoRedemption.deleted == 0,
    ).all()
    activities: List[OrderAdminActivityDTO] = []
    for record in records:
        activities.append(
            OrderAdminActivityDTO(
                active_id=record.promo_bid,
                active_name=record.promo_name,
                price=_format_decimal(record.discount_amount),
                status=record.status,
                status_key=ACTIVE_STATUS_KEY_MAP.get(
                    record.status, "module.order.activeStatus.unknown"
                ),
                created_at=_format_admin_datetime(record.created_at),
                updated_at=_format_admin_datetime(record.updated_at),
            )
        )
    return activities


def _load_order_coupons(order_bid: str) -> List[OrderAdminCouponDTO]:
    """Load coupon usage records tied to an order and format as DTOs."""
    records = CouponUsage.query.filter(
        CouponUsage.order_bid == order_bid,
        CouponUsage.deleted == 0,
    ).all()
    coupons: List[OrderAdminCouponDTO] = []
    for record in records:
        coupons.append(
            OrderAdminCouponDTO(
                coupon_bid=record.coupon_bid,
                code=record.code,
                name=record.name,
                discount_type=record.discount_type,
                discount_type_key=COUPON_TYPE_KEY_MAP.get(
                    record.discount_type, "module.order.couponType.unknown"
                ),
                value=_format_decimal(record.value),
                status=record.status,
                status_key=COUPON_STATUS_KEY_MAP.get(
                    record.status, "module.order.couponStatus.unknown"
                ),
                created_at=_format_admin_datetime(record.created_at),
                updated_at=_format_admin_datetime(record.updated_at),
            )
        )
    return coupons


def _load_payment_detail(order: Order) -> Optional[OrderAdminPaymentDTO]:
    """Build payment detail DTO from channel-specific order records."""
    payment_channel = order.payment_channel or ""
    if payment_channel == "stripe":
        stripe_order = (
            legacy_stripe_snapshot_query()
            .filter(
                StripeOrder.order_bid == order.order_bid,
            )
            .order_by(StripeOrder.id.desc())
            .first()
        )
        if not stripe_order:
            return None
        return OrderAdminPaymentDTO(
            payment_channel="stripe",
            payment_channel_key=PAYMENT_CHANNEL_KEY_MAP.get(
                "stripe", "module.order.paymentChannel.unknown"
            ),
            status=stripe_order.status,
            status_key=PAYMENT_STATUS_KEY_MAP.get(
                stripe_order.status, "module.order.paymentStatus.unknown"
            ),
            amount=_format_cents(stripe_order.amount),
            currency=stripe_order.currency,
            payment_intent_id=stripe_order.payment_intent_id or "",
            checkout_session_id=stripe_order.checkout_session_id or "",
            latest_charge_id=stripe_order.latest_charge_id or "",
            receipt_url=stripe_order.receipt_url or "",
            payment_method=stripe_order.payment_method or "",
            created_at=_format_admin_datetime(stripe_order.created_at),
            updated_at=_format_admin_datetime(stripe_order.updated_at),
        )

    if payment_channel == "pingxx":
        pingxx_order = (
            legacy_pingxx_snapshot_query()
            .filter(
                PingxxOrder.order_bid == order.order_bid,
            )
            .order_by(PingxxOrder.id.desc())
            .first()
        )
        if not pingxx_order:
            return None
        return OrderAdminPaymentDTO(
            payment_channel="pingxx",
            payment_channel_key=PAYMENT_CHANNEL_KEY_MAP.get(
                "pingxx", "module.order.paymentChannel.unknown"
            ),
            status=pingxx_order.status,
            status_key=PAYMENT_STATUS_KEY_MAP.get(
                pingxx_order.status, "module.order.paymentStatus.unknown"
            ),
            amount=_format_cents(pingxx_order.amount),
            currency=pingxx_order.currency,
            transaction_no=pingxx_order.transaction_no or "",
            charge_id=pingxx_order.charge_id or "",
            channel=pingxx_order.channel or "",
            created_at=_format_admin_datetime(pingxx_order.created_at),
            updated_at=_format_admin_datetime(pingxx_order.updated_at),
        )

    if payment_channel in {"alipay", "wechatpay"}:
        native_model = native_snapshot_model(payment_channel)
        native_order = (
            legacy_native_snapshot_query(payment_channel)
            .filter(
                native_model.order_bid == order.order_bid,
            )
            .order_by(native_model.id.desc())
            .first()
        )
        if not native_order:
            return None
        return OrderAdminPaymentDTO(
            payment_channel=payment_channel,
            payment_channel_key=PAYMENT_CHANNEL_KEY_MAP.get(
                payment_channel, "module.order.paymentChannel.unknown"
            ),
            status=native_order.status,
            status_key=PAYMENT_STATUS_KEY_MAP.get(
                native_order.status, "module.order.paymentStatus.unknown"
            ),
            amount=_format_cents(native_order.amount),
            currency=native_order.currency,
            transaction_no=native_order.provider_attempt_id or "",
            charge_id=native_order.transaction_id or "",
            channel=native_order.channel or "",
            created_at=_format_admin_datetime(native_order.created_at),
            updated_at=_format_admin_datetime(native_order.updated_at),
        )

    return None


def get_order_detail(app: Flask, user_id: str, order_bid: str) -> OrderAdminDetailDTO:
    """Return admin order detail after permission check for the operator."""
    with app.app_context():
        order = Order.query.filter(
            Order.order_bid == order_bid,
            Order.deleted == 0,
        ).first()
        if not order:
            raise_error("server.order.orderNotFound")
        creator_bid = get_shifu_creator_bid(app, order.shifu_bid)
        if creator_bid != user_id:
            raise_error("server.shifu.noPermission")

        shifu_map = _load_shifu_map([order.shifu_bid])
        user_map = _load_user_map([order.user_bid])
        coupon_map = _load_coupon_code_map([order.order_bid])
        summary = _build_order_item(order, shifu_map, user_map, coupon_map)
        payment_detail = _load_payment_detail(order)
        if not payment_detail:
            payment_channel = order.payment_channel or ""
            payment_detail = OrderAdminPaymentDTO(
                payment_channel=payment_channel,
                payment_channel_key=PAYMENT_CHANNEL_KEY_MAP.get(
                    payment_channel, "module.order.paymentChannel.unknown"
                ),
                status=0,
                status_key="module.order.paymentStatus.unknown",
                amount="0",
                currency="",
            )

        return OrderAdminDetailDTO(
            order=summary,
            activities=_load_order_activities(order.order_bid),
            coupons=_load_order_coupons(order.order_bid),
            payment=payment_detail,
        )


def get_operator_order_detail(app: Flask, order_bid: str) -> OrderAdminDetailDTO:
    """Return global operator order detail without creator scope restrictions."""
    with app.app_context():
        order = Order.query.filter(
            Order.order_bid == order_bid,
            Order.deleted == 0,
        ).first()
        if not order:
            raise_error("server.order.orderNotFound")

        shifu_map = _load_shifu_map([order.shifu_bid])
        user_map = _load_user_map([order.user_bid])
        coupon_map = _load_coupon_code_map([order.order_bid])
        summary = _build_order_item(order, shifu_map, user_map, coupon_map)
        payment_detail = _load_payment_detail(order)
        if not payment_detail:
            payment_channel = order.payment_channel or ""
            payment_detail = OrderAdminPaymentDTO(
                payment_channel=payment_channel,
                payment_channel_key=PAYMENT_CHANNEL_KEY_MAP.get(
                    payment_channel, "module.order.paymentChannel.unknown"
                ),
                status=0,
                status_key="module.order.paymentStatus.unknown",
                amount="0",
                currency="",
            )

        return OrderAdminDetailDTO(
            order=summary,
            activities=_load_order_activities(order.order_bid),
            coupons=_load_order_coupons(order.order_bid),
            payment=payment_detail,
        )
