"""Application services for the catalog normalization control plane.

Ports (session/ORM) stay in this module so FastAPI and the dashboard share one
use-case surface: overview, review queue, channel policies, projection cache.
"""

from __future__ import annotations

from datetime import datetime
from typing import Any, Dict, List, Optional, Sequence

from sqlalchemy import func, select
from sqlalchemy.orm import Session

from db.channel_exports import resolve_pipeline_channel_code
from db.location_experience import extract_location_settings, merge_location_settings
from db.compat_connections import resolve_native_connection_id
from db.models import (
    CatalogNormalizationProfile,
    CatalogNormalizationRule,
    CatalogNormalizationRuleTarget,
    CatalogReviewQueueItem,
    CatalogValueAlias,
    CollectionCommerceProfile,
    ChannelCatalogPolicy,
    ChannelCatalogPolicyScope,
    ChannelProjectionRow,
    ChannelProjectionRun,
    ChannelShellProduct,
    ChannelVariationPolicy,
    MasterProduct,
)


def _as_bool(value: Any, *, default: bool = False) -> bool:
    if value is None:
        return default
    if isinstance(value, bool):
        return value
    text = str(value).strip().lower()
    if not text:
        return default
    if text in {"1", "true", "yes", "y", "on"}:
        return True
    if text in {"0", "false", "no", "n", "off"}:
        return False
    return default


def _clean_str(value: Any) -> Optional[str]:
    if value is None:
        return None
    text = str(value).strip()
    return text or None


def _normalize_channel_code(value: Any) -> Optional[str]:
    text = _clean_str(value)
    if not text:
        return None
    return resolve_pipeline_channel_code(text)


def build_normalization_overview(session: Session) -> Dict[str, Any]:
    status_rows = session.execute(
        select(MasterProduct.normalization_status, func.count())
        .where(MasterProduct.is_active.is_(True))
        .group_by(MasterProduct.normalization_status)
    ).all()
    normalization_by_status = {str(status or "unknown"): int(count) for status, count in status_rows}

    review_open = session.scalar(
        select(func.count()).select_from(CatalogReviewQueueItem).where(CatalogReviewQueueItem.status == "open")
    ) or 0
    review_by_severity = {
        str(severity or "unknown"): int(count)
        for severity, count in session.execute(
            select(CatalogReviewQueueItem.severity, func.count())
            .where(CatalogReviewQueueItem.status == "open")
            .group_by(CatalogReviewQueueItem.severity)
        ).all()
    }

    projection_by_channel: Dict[str, Any] = {}
    for channel_code, status, count in session.execute(
        select(
            ChannelProjectionRow.channel_code,
            ChannelProjectionRow.projection_status,
            func.count(),
        ).group_by(ChannelProjectionRow.channel_code, ChannelProjectionRow.projection_status)
    ).all():
        bucket = projection_by_channel.setdefault(
            str(channel_code),
            {"ready": 0, "stale": 0, "other": 0, "total": 0},
        )
        key = str(status or "other")
        if key not in {"ready", "stale"}:
            key = "other"
        bucket[key] += int(count)
        bucket["total"] += int(count)

    return {
        "master_active": sum(normalization_by_status.values()),
        "normalization_by_status": normalization_by_status,
        "review_queue": {"open": int(review_open), "by_severity": review_by_severity},
        "projection_by_channel": projection_by_channel,
        "counts": {
            "profiles": session.scalar(select(func.count()).select_from(CatalogNormalizationProfile)) or 0,
            "rules": session.scalar(select(func.count()).select_from(CatalogNormalizationRule)) or 0,
            "value_aliases": session.scalar(select(func.count()).select_from(CatalogValueAlias)) or 0,
            "catalog_policies": session.scalar(select(func.count()).select_from(ChannelCatalogPolicy)) or 0,
            "variation_policies": session.scalar(select(func.count()).select_from(ChannelVariationPolicy)) or 0,
            "shell_products": session.scalar(select(func.count()).select_from(ChannelShellProduct)) or 0,
        },
    }


def list_review_queue(
    session: Session,
    *,
    status: Optional[str] = "open",
    severity: Optional[str] = None,
    sku: Optional[str] = None,
    limit: int = 100,
    offset: int = 0,
) -> Dict[str, Any]:
    filters = []
    if status:
        filters.append(CatalogReviewQueueItem.status == status)
    if severity:
        filters.append(CatalogReviewQueueItem.severity == severity)
    if sku:
        filters.append(CatalogReviewQueueItem.sku == str(sku).strip())

    count_stmt = select(func.count()).select_from(CatalogReviewQueueItem)
    stmt = select(CatalogReviewQueueItem).order_by(CatalogReviewQueueItem.id.desc())
    for clause in filters:
        count_stmt = count_stmt.where(clause)
        stmt = stmt.where(clause)

    total = session.scalar(count_stmt) or 0
    rows = list(session.scalars(stmt.offset(offset).limit(limit)).all())
    return {
        "total": int(total),
        "limit": limit,
        "offset": offset,
        "items": [_review_item_dict(row) for row in rows],
    }


def resolve_review_item(
    session: Session,
    item_id: int,
    *,
    resolution: str = "accepted",
    notes: Optional[str] = None,
    proposed_value: Optional[str] = None,
) -> Dict[str, Any]:
    row = session.get(CatalogReviewQueueItem, item_id)
    if row is None:
        raise KeyError(f"review item {item_id} not found")
    now = datetime.utcnow()
    status = str(resolution or "accepted").strip().lower()
    if status in {"accept", "accepted", "resolved"}:
        row.status = "resolved"
    elif status in {"dismiss", "dismissed"}:
        row.status = "dismissed"
    else:
        row.status = status
    if proposed_value is not None:
        row.proposed_value = proposed_value
    if notes is not None:
        row.notes = notes
    row.resolved_at = now
    row.updated_at = now
    session.flush()
    return _review_item_dict(row)


def list_catalog_policies(session: Session, *, channel_code: Optional[str] = None) -> Dict[str, Any]:
    stmt = select(ChannelCatalogPolicy).order_by(ChannelCatalogPolicy.channel_code, ChannelCatalogPolicy.id)
    if channel_code:
        stmt = stmt.where(ChannelCatalogPolicy.channel_code == resolve_pipeline_channel_code(channel_code))
    policies = list(session.scalars(stmt).all())
    return {"policies": [_policy_dict(session, policy) for policy in policies]}


def upsert_catalog_policy(
    session: Session,
    *,
    channel_code: str,
    connection_id: Optional[int] = None,
    payload: Dict[str, Any],
) -> Dict[str, Any]:
    channel = resolve_pipeline_channel_code(channel_code)
    if connection_id is not None and channel in {"shopify", "magento"}:
        connection_id = resolve_native_connection_id(channel, connection_id)

    stmt = select(ChannelCatalogPolicy).where(ChannelCatalogPolicy.channel_code == channel)
    if connection_id is None:
        stmt = stmt.where(ChannelCatalogPolicy.connection_id.is_(None))
    else:
        stmt = stmt.where(ChannelCatalogPolicy.connection_id == connection_id)
    policy = session.scalar(stmt)
    now = datetime.utcnow()
    if policy is None:
        policy = ChannelCatalogPolicy(channel_code=channel, connection_id=connection_id, created_at=now)
        session.add(policy)

    for field in (
        "policy_name",
        "publish_mode",
        "variation_mode",
        "shell_product_mode",
        "shell_sku_template",
        "notes",
    ):
        if field in payload:
            setattr(policy, field, payload.get(field))
    for field in ("publish_shell_products", "exclude_rta", "include_assembled", "include_accessories_without_brand", "is_active"):
        if field in payload:
            setattr(policy, field, _as_bool(payload.get(field), default=getattr(policy, field)))
    if "max_axes" in payload:
        policy.max_axes = payload.get("max_axes")
    if "default_allowed_axes" in payload:
        policy.default_allowed_axes = payload.get("default_allowed_axes")
    policy.updated_at = now
    session.flush()

    if "scopes" in payload and isinstance(payload.get("scopes"), list):
        existing = list(
            session.scalars(
                select(ChannelCatalogPolicyScope).where(ChannelCatalogPolicyScope.policy_id == policy.id)
            ).all()
        )
        for row in existing:
            session.delete(row)
        session.flush()
        for scope in payload["scopes"]:
            session.add(
                ChannelCatalogPolicyScope(
                    policy_id=policy.id,
                    scope_kind=str(scope.get("scope_kind") or "collection"),
                    scope_code=str(scope.get("scope_code") or "").strip(),
                    action=str(scope.get("action") or "include"),
                    priority=int(scope.get("priority") or 100),
                    notes=scope.get("notes"),
                    created_at=now,
                    updated_at=now,
                )
            )
        session.flush()

    return _policy_dict(session, policy)


def list_normalization_profiles(session: Session) -> Dict[str, Any]:
    profiles = list(
        session.scalars(select(CatalogNormalizationProfile).order_by(CatalogNormalizationProfile.code)).all()
    )
    return {"profiles": [_profile_dict(row) for row in profiles]}


def upsert_normalization_profile(
    session: Session,
    *,
    profile_id: Optional[int] = None,
    payload: Dict[str, Any],
) -> Dict[str, Any]:
    now = datetime.utcnow()
    if profile_id is None:
        row = CatalogNormalizationProfile(created_at=now)
        session.add(row)
    else:
        row = session.get(CatalogNormalizationProfile, int(profile_id))
        if row is None:
            raise KeyError(f"profile {profile_id} not found")

    code = _clean_str(payload.get("code"))
    name = _clean_str(payload.get("name"))
    if not code:
        raise ValueError("Profile code is required")
    if not name:
        raise ValueError("Profile name is required")

    row.code = code
    row.name = name
    row.family_id = payload.get("family_id") or None
    row.config_json = payload.get("config_json")
    row.is_active = _as_bool(payload.get("is_active"), default=getattr(row, "is_active", True))
    row.notes = _clean_str(payload.get("notes"))
    row.updated_at = now
    session.flush()
    return _profile_dict(row)


def delete_normalization_profile(session: Session, profile_id: int) -> Dict[str, Any]:
    row = session.get(CatalogNormalizationProfile, int(profile_id))
    if row is None:
        raise KeyError(f"profile {profile_id} not found")
    session.delete(row)
    session.flush()
    return {"status": "deleted", "id": int(profile_id)}


def list_normalization_rules(
    session: Session,
    *,
    profile_id: Optional[int] = None,
    status: Optional[str] = None,
    limit: int = 200,
) -> Dict[str, Any]:
    stmt = select(CatalogNormalizationRule).order_by(
        CatalogNormalizationRule.priority, CatalogNormalizationRule.id
    )
    if profile_id is not None:
        stmt = stmt.where(CatalogNormalizationRule.profile_id == profile_id)
    if status:
        stmt = stmt.where(CatalogNormalizationRule.rule_status == status)
    rows = list(session.scalars(stmt.limit(limit)).all())
    return {"rules": [_rule_dict(session, row) for row in rows]}


def upsert_normalization_rule(
    session: Session,
    *,
    rule_id: Optional[int] = None,
    payload: Dict[str, Any],
) -> Dict[str, Any]:
    now = datetime.utcnow()
    if rule_id is None:
        row = CatalogNormalizationRule(created_at=now)
        session.add(row)
    else:
        row = session.get(CatalogNormalizationRule, int(rule_id))
        if row is None:
            raise KeyError(f"rule {rule_id} not found")

    rule_code = _clean_str(payload.get("rule_code"))
    if not rule_code:
        raise ValueError("Rule code is required")

    row.profile_id = payload.get("profile_id") or None
    row.rule_code = rule_code
    row.scope_kind = _clean_str(payload.get("scope_kind")) or "sku_prefix"
    row.scope_value = _clean_str(payload.get("scope_value"))
    row.field_name = _clean_str(payload.get("field_name"))
    row.match_operator = _clean_str(payload.get("match_operator")) or "equals"
    row.match_value = _clean_str(payload.get("match_value"))
    row.priority = int(payload.get("priority") or 100)
    row.stop_processing = _as_bool(payload.get("stop_processing"), default=getattr(row, "stop_processing", True))
    row.rule_status = _clean_str(payload.get("rule_status")) or "active"
    row.notes = _clean_str(payload.get("notes"))
    row.updated_at = now
    session.flush()

    if "targets" in payload and isinstance(payload.get("targets"), list):
        existing = list(
            session.scalars(
                select(CatalogNormalizationRuleTarget).where(CatalogNormalizationRuleTarget.rule_id == row.id)
            ).all()
        )
        for target in existing:
            session.delete(target)
        session.flush()
        for item in payload.get("targets") or []:
            target_field = _clean_str(item.get("target_field"))
            if not target_field:
                continue
            session.add(
                CatalogNormalizationRuleTarget(
                    rule_id=row.id,
                    target_field=target_field,
                    target_value=_clean_str(item.get("target_value")),
                    target_code=_clean_str(item.get("target_code")),
                    target_payload=item.get("target_payload"),
                    created_at=now,
                    updated_at=now,
                )
            )
        session.flush()

    return _rule_dict(session, row)


def delete_normalization_rule(session: Session, rule_id: int) -> Dict[str, Any]:
    row = session.get(CatalogNormalizationRule, int(rule_id))
    if row is None:
        raise KeyError(f"rule {rule_id} not found")
    session.delete(row)
    session.flush()
    return {"status": "deleted", "id": int(rule_id)}


def list_value_aliases(
    session: Session,
    *,
    entity_kind: Optional[str] = None,
    field_name: Optional[str] = None,
    limit: int = 200,
) -> Dict[str, Any]:
    stmt = select(CatalogValueAlias).order_by(
        CatalogValueAlias.entity_kind, CatalogValueAlias.field_name, CatalogValueAlias.sort_order
    )
    if entity_kind:
        stmt = stmt.where(CatalogValueAlias.entity_kind == entity_kind)
    if field_name:
        stmt = stmt.where(CatalogValueAlias.field_name == field_name)
    rows = list(session.scalars(stmt.limit(limit)).all())
    return {"aliases": [_alias_dict(row) for row in rows]}


def upsert_value_alias(
    session: Session,
    *,
    alias_id: Optional[int] = None,
    payload: Dict[str, Any],
) -> Dict[str, Any]:
    now = datetime.utcnow()
    if alias_id is None:
        row = CatalogValueAlias(created_at=now)
        session.add(row)
    else:
        row = session.get(CatalogValueAlias, int(alias_id))
        if row is None:
            raise KeyError(f"alias {alias_id} not found")

    entity_kind = _clean_str(payload.get("entity_kind"))
    field_name = _clean_str(payload.get("field_name"))
    alias_value = _clean_str(payload.get("alias_value"))
    canonical_code = _clean_str(payload.get("canonical_code"))
    if not entity_kind:
        raise ValueError("Entity kind is required")
    if not field_name:
        raise ValueError("Field name is required")
    if not alias_value:
        raise ValueError("Alias value is required")
    if not canonical_code:
        raise ValueError("Canonical code is required")

    row.entity_kind = entity_kind
    row.field_name = field_name
    row.alias_value = alias_value
    row.canonical_code = canonical_code
    row.canonical_label = _clean_str(payload.get("canonical_label"))
    row.match_operator = _clean_str(payload.get("match_operator")) or "equals"
    row.sort_order = int(payload.get("sort_order") or 100)
    row.is_active = _as_bool(payload.get("is_active"), default=getattr(row, "is_active", True))
    row.notes = _clean_str(payload.get("notes"))
    row.updated_at = now
    session.flush()
    return _alias_dict(row)


def delete_value_alias(session: Session, alias_id: int) -> Dict[str, Any]:
    row = session.get(CatalogValueAlias, int(alias_id))
    if row is None:
        raise KeyError(f"alias {alias_id} not found")
    session.delete(row)
    session.flush()
    return {"status": "deleted", "id": int(alias_id)}


def list_collection_commerce_profiles(
    session: Session,
    *,
    channel_code: Optional[str] = None,
    connection_id: Optional[int] = None,
    collection_name: Optional[str] = None,
    limit: int = 200,
) -> Dict[str, Any]:
    stmt = select(CollectionCommerceProfile).order_by(
        CollectionCommerceProfile.collection_name,
        CollectionCommerceProfile.channel_code,
        CollectionCommerceProfile.connection_id,
        CollectionCommerceProfile.id,
    )
    if channel_code:
        stmt = stmt.where(CollectionCommerceProfile.channel_code == _normalize_channel_code(channel_code))
    if connection_id is not None:
        stmt = stmt.where(CollectionCommerceProfile.connection_id == int(connection_id))
    if collection_name:
        stmt = stmt.where(CollectionCommerceProfile.collection_name.ilike(f"%{str(collection_name).strip()}%"))
    rows = list(session.scalars(stmt.limit(limit)).all())
    return {"profiles": [_collection_commerce_dict(row) for row in rows]}


def upsert_collection_commerce_profile(
    session: Session,
    *,
    profile_id: Optional[int] = None,
    payload: Dict[str, Any],
) -> Dict[str, Any]:
    now = datetime.utcnow()
    if profile_id is None:
        row = CollectionCommerceProfile(created_at=now)
        session.add(row)
    else:
        row = session.get(CollectionCommerceProfile, int(profile_id))
        if row is None:
            raise KeyError(f"collection commerce profile {profile_id} not found")

    channel_code = _normalize_channel_code(payload.get("channel_code"))
    connection_id = payload.get("connection_id")
    if connection_id is not None and str(connection_id).strip() != "" and channel_code in {"magento", "shopify"}:
        connection_id = resolve_native_connection_id(channel_code, int(connection_id))
    else:
        connection_id = int(connection_id) if connection_id not in (None, "") else None

    availability_modes = payload.get("availability_modes_json")
    if availability_modes is None and isinstance(payload.get("availability_modes"), list):
        availability_modes = {"modes": [str(item).strip() for item in payload.get("availability_modes") if str(item).strip()]}
    from db.collection_landing_schema import normalize_collection_badges

    badges = payload.get("badges_json")
    if badges is None:
        badges = normalize_collection_badges(payload.get("badges"))
    else:
        badges = normalize_collection_badges(badges)
    landing_metadata = payload.get("landing_metadata_json")
    if landing_metadata is None and isinstance(payload.get("landing_metadata"), dict):
        landing_metadata = payload.get("landing_metadata")
    landing_metadata = merge_location_settings(
        landing_metadata,
        default_location_code=payload.get("default_location_code") if "default_location_code" in payload else None,
        allow_geolocation=payload.get("allow_geolocation") if "allow_geolocation" in payload else None,
        allow_manual_selection=payload.get("allow_manual_selection") if "allow_manual_selection" in payload else None,
        selection_mode=payload.get("location_selection_mode") if "location_selection_mode" in payload else None,
        location_label=payload.get("location_label") if "location_label" in payload else None,
        notes=payload.get("location_settings_notes") if "location_settings_notes" in payload else None,
    )

    row.collection_registry_id = payload.get("collection_registry_id") or None
    row.category_l1 = _clean_str(payload.get("category_l1"))
    row.collection_name = _clean_str(payload.get("collection_name"))
    row.channel_code = channel_code
    row.connection_id = connection_id
    row.sample_master_sku = _clean_str(payload.get("sample_master_sku"))
    row.sample_category_path_slug = _clean_str(payload.get("sample_category_path_slug"))
    row.availability_modes_json = availability_modes
    row.appointment_url = _clean_str(payload.get("appointment_url"))
    row.sample_cta_label = _clean_str(payload.get("sample_cta_label"))
    row.appointment_cta_label = _clean_str(payload.get("appointment_cta_label"))
    row.badges_json = badges
    row.landing_metadata_json = landing_metadata
    row.is_active = _as_bool(payload.get("is_active"), default=getattr(row, "is_active", True))
    row.notes = _clean_str(payload.get("notes"))
    row.updated_at = now
    session.flush()
    return _collection_commerce_dict(row)


def delete_collection_commerce_profile(session: Session, profile_id: int) -> Dict[str, Any]:
    row = session.get(CollectionCommerceProfile, int(profile_id))
    if row is None:
        raise KeyError(f"collection commerce profile {profile_id} not found")
    session.delete(row)
    session.flush()
    return {"status": "deleted", "id": int(profile_id)}


def projection_status(
    session: Session,
    *,
    channel_code: str,
    connection_id: Optional[int] = None,
) -> Dict[str, Any]:
    from db.channel_projection_cache import _native_connection_id
    from db.channel_projection_invalidation import latest_projection_run

    channel = resolve_pipeline_channel_code(channel_code)
    connection_id = _native_connection_id(channel, connection_id)
    stmt = select(ChannelProjectionRow.projection_status, func.count()).where(
        ChannelProjectionRow.channel_code == channel
    )
    if connection_id is None:
        stmt = stmt.where(ChannelProjectionRow.connection_id.is_(None))
    else:
        stmt = stmt.where(ChannelProjectionRow.connection_id == connection_id)
    counts = {str(status or "unknown"): int(count) for status, count in session.execute(stmt.group_by(ChannelProjectionRow.projection_status)).all()}
    run = latest_projection_run(session, channel_code=channel, connection_id=connection_id)
    return {
        "channel_code": channel,
        "connection_id": connection_id,
        "counts": counts,
        "ready": int(counts.get("ready") or 0),
        "stale": int(counts.get("stale") or 0),
        "total": sum(counts.values()),
        "latest_run": _run_dict(run) if run else None,
    }


def inspect_projection(
    session: Session,
    *,
    channel_code: str,
    connection_id: Optional[int] = None,
    skus: Optional[Sequence[str]] = None,
    apply: bool = False,
) -> Dict[str, Any]:
    from db.channel_projection_invalidation import inspect_projection_drift

    return inspect_projection_drift(
        session,
        channel_code=channel_code,
        connection_id=connection_id,
        skus=skus,
        apply=apply,
    )


def rebuild_projection(
    session: Session,
    *,
    channel_code: str,
    connection_id: Optional[int] = None,
    skus: Optional[Sequence[str]] = None,
    only_assigned: bool = True,
    replace_scope: bool = False,
) -> Dict[str, Any]:
    from db.channel_capabilities import resolve_product_structure_for_connection
    from db.channel_exports import build_channel_product_payloads
    from db.channel_projection_cache import _native_connection_id, store_projection_payloads

    channel = resolve_pipeline_channel_code(channel_code)
    connection_id = _native_connection_id(channel, connection_id)
    if connection_id is None:
        return {"status": "skipped", "reason": "no_connection", "channel_code": channel}

    structure = resolve_product_structure_for_connection(
        session,
        channel,
        connection_id=connection_id,
    )
    sku_list = [str(s).strip() for s in (skus or []) if str(s).strip()] or None
    payloads = build_channel_product_payloads(
        session,
        channel,
        skus=sku_list,
        only_assigned=only_assigned if sku_list is None else False,
        connection_id=connection_id,
        product_structure=structure,
        prefer_projection_cache=False,
    )
    stored = store_projection_payloads(
        session,
        channel_code=channel,
        connection_id=connection_id,
        payloads=payloads,
        product_structure=structure,
        scope_key="assigned" if only_assigned and sku_list is None else "rebuild",
        notes="Rebuilt via API",
        replace_scope=replace_scope and sku_list is None,
    )
    return {
        "status": "ok",
        "channel_code": channel,
        "connection_id": connection_id,
        "product_structure": structure,
        "sku_count": len(payloads),
        **stored,
    }


def list_projection_rows(
    session: Session,
    *,
    channel_code: str,
    connection_id: Optional[int] = None,
    status: Optional[str] = None,
    sku: Optional[str] = None,
    limit: int = 50,
    offset: int = 0,
) -> Dict[str, Any]:
    from db.channel_projection_cache import _native_connection_id

    channel = resolve_pipeline_channel_code(channel_code)
    connection_id = _native_connection_id(channel, connection_id)
    filters = [ChannelProjectionRow.channel_code == channel]
    if connection_id is None:
        filters.append(ChannelProjectionRow.connection_id.is_(None))
    else:
        filters.append(ChannelProjectionRow.connection_id == connection_id)
    if status:
        filters.append(ChannelProjectionRow.projection_status == status)
    if sku:
        filters.append(ChannelProjectionRow.master_sku == str(sku).strip())

    count_stmt = select(func.count()).select_from(ChannelProjectionRow)
    stmt = select(ChannelProjectionRow).order_by(ChannelProjectionRow.master_sku)
    for clause in filters:
        count_stmt = count_stmt.where(clause)
        stmt = stmt.where(clause)

    total = session.scalar(count_stmt) or 0
    rows = list(session.scalars(stmt.offset(offset).limit(limit)).all())
    return {
        "channel_code": channel,
        "connection_id": connection_id,
        "total": int(total),
        "limit": limit,
        "offset": offset,
        "rows": [_projection_row_dict(row) for row in rows],
    }


def _review_item_dict(row: CatalogReviewQueueItem) -> Dict[str, Any]:
    return {
        "id": row.id,
        "master_product_id": row.master_product_id,
        "sku": row.sku,
        "field_name": row.field_name,
        "issue_code": row.issue_code,
        "severity": row.severity,
        "status": row.status,
        "raw_value": row.raw_value,
        "proposed_value": row.proposed_value,
        "source_rule_id": row.source_rule_id,
        "context_json": row.context_json,
        "assigned_to": row.assigned_to,
        "resolved_at": row.resolved_at.isoformat() if row.resolved_at else None,
        "notes": row.notes,
        "created_at": row.created_at.isoformat() if row.created_at else None,
        "updated_at": row.updated_at.isoformat() if row.updated_at else None,
    }


def _policy_dict(session: Session, policy: ChannelCatalogPolicy) -> Dict[str, Any]:
    scopes = list(
        session.scalars(
            select(ChannelCatalogPolicyScope)
            .where(ChannelCatalogPolicyScope.policy_id == policy.id)
            .order_by(ChannelCatalogPolicyScope.priority, ChannelCatalogPolicyScope.id)
        ).all()
    )
    return {
        "id": policy.id,
        "channel_code": policy.channel_code,
        "connection_id": policy.connection_id,
        "policy_name": policy.policy_name,
        "publish_mode": policy.publish_mode,
        "variation_mode": policy.variation_mode,
        "shell_product_mode": policy.shell_product_mode,
        "shell_sku_template": policy.shell_sku_template,
        "publish_shell_products": policy.publish_shell_products,
        "exclude_rta": policy.exclude_rta,
        "include_assembled": policy.include_assembled,
        "include_accessories_without_brand": policy.include_accessories_without_brand,
        "max_axes": policy.max_axes,
        "default_allowed_axes": policy.default_allowed_axes,
        "notes": policy.notes,
        "is_active": policy.is_active,
        "scopes": [
            {
                "id": scope.id,
                "scope_kind": scope.scope_kind,
                "scope_code": scope.scope_code,
                "action": scope.action,
                "priority": scope.priority,
                "notes": scope.notes,
            }
            for scope in scopes
        ],
        "updated_at": policy.updated_at.isoformat() if policy.updated_at else None,
    }


def _profile_dict(row: CatalogNormalizationProfile) -> Dict[str, Any]:
    return {
        "id": row.id,
        "code": row.code,
        "name": row.name,
        "family_id": row.family_id,
        "config_json": row.config_json,
        "is_active": row.is_active,
        "notes": row.notes,
    }


def _rule_dict(session: Session, row: CatalogNormalizationRule) -> Dict[str, Any]:
    targets = list(
        session.scalars(
            select(CatalogNormalizationRuleTarget).where(CatalogNormalizationRuleTarget.rule_id == row.id)
        ).all()
    )
    return {
        "id": row.id,
        "profile_id": row.profile_id,
        "rule_code": row.rule_code,
        "scope_kind": row.scope_kind,
        "scope_value": row.scope_value,
        "field_name": row.field_name,
        "match_operator": row.match_operator,
        "match_value": row.match_value,
        "priority": row.priority,
        "stop_processing": row.stop_processing,
        "rule_status": row.rule_status,
        "notes": row.notes,
        "targets": [
            {
                "id": target.id,
                "target_field": target.target_field,
                "target_value": target.target_value,
                "target_code": target.target_code,
                "target_payload": target.target_payload,
            }
            for target in targets
        ],
    }


def _alias_dict(row: CatalogValueAlias) -> Dict[str, Any]:
    return {
        "id": row.id,
        "entity_kind": row.entity_kind,
        "field_name": row.field_name,
        "alias_value": row.alias_value,
        "canonical_code": row.canonical_code,
        "canonical_label": row.canonical_label,
        "match_operator": row.match_operator,
        "sort_order": row.sort_order,
        "is_active": row.is_active,
        "notes": row.notes,
    }


def _collection_commerce_dict(row: CollectionCommerceProfile) -> Dict[str, Any]:
    location_settings = extract_location_settings(row.landing_metadata_json)
    return {
        "id": row.id,
        "collection_registry_id": row.collection_registry_id,
        "category_l1": row.category_l1,
        "collection_name": row.collection_name,
        "channel_code": row.channel_code,
        "connection_id": row.connection_id,
        "sample_master_sku": row.sample_master_sku,
        "sample_category_path_slug": row.sample_category_path_slug,
        "availability_modes_json": row.availability_modes_json,
        "appointment_url": row.appointment_url,
        "sample_cta_label": row.sample_cta_label,
        "appointment_cta_label": row.appointment_cta_label,
        "badges_json": row.badges_json,
        "landing_metadata_json": row.landing_metadata_json,
        "default_location_code": location_settings.get("default_location_code"),
        "allow_geolocation": location_settings.get("allow_geolocation"),
        "allow_manual_selection": location_settings.get("allow_manual_selection"),
        "location_selection_mode": location_settings.get("selection_mode"),
        "location_label": location_settings.get("location_label"),
        "location_settings_notes": location_settings.get("notes"),
        "is_active": row.is_active,
        "notes": row.notes,
        "updated_at": row.updated_at.isoformat() if row.updated_at else None,
    }


def _run_dict(row: ChannelProjectionRun) -> Dict[str, Any]:
    return {
        "id": row.id,
        "channel_code": row.channel_code,
        "connection_id": row.connection_id,
        "scope_key": row.scope_key,
        "status": row.status,
        "product_structure": row.product_structure,
        "sku_count": row.sku_count,
        "started_at": row.started_at.isoformat() if row.started_at else None,
        "completed_at": row.completed_at.isoformat() if row.completed_at else None,
        "invalidated_at": row.invalidated_at.isoformat() if row.invalidated_at else None,
        "notes": row.notes,
    }


def _projection_row_dict(row: ChannelProjectionRow) -> Dict[str, Any]:
    return {
        "id": row.id,
        "master_sku": row.master_sku,
        "channel_sku": row.channel_sku,
        "product_role": row.product_role,
        "product_structure": row.product_structure,
        "projection_status": row.projection_status,
        "payload_hash": row.payload_hash,
        "source_row_hash": row.source_row_hash,
        "normalized_at": row.normalized_at.isoformat() if row.normalized_at else None,
        "updated_at": row.updated_at.isoformat() if row.updated_at else None,
    }