# -*- coding: utf-8 -*-
"""ODS member_profiles birthday 字段提取验证。

任务 3.2: 确认 ODS 入库逻辑能从 JSON payload 中提取 birthday 字段。
ODS_MEMBER 任务使用 schema-aware 动态入库（_insert_records_schema_aware），
当 ods.member_profiles 表包含 birthday 列时，会自动从 API JSON 中按列名匹配提取。
本测试通过 FakeDB 模拟包含 birthday 列的表结构，验证提取行为。
"""
from __future__ import annotations

import logging
import os
import sys
from pathlib import Path

import pytest

PROJECT_ROOT = Path(__file__).resolve().parents[2]
if str(PROJECT_ROOT) not in sys.path:
    sys.path.insert(0, str(PROJECT_ROOT))

os.environ.setdefault("ETL_SKIP_DOTENV", "1")

from tasks.ods.ods_tasks import ODS_TASK_CLASSES, ODS_TASK_SPECS, BaseOdsTask
from .task_test_utils import (
    create_test_config,
    FakeDBOperations,
    FakeAPIClient,
    FakeCursor,
    FakeConnection,
)

# ---------------------------------------------------------------------------
# 常量
# ---------------------------------------------------------------------------
_MEMBER_CODE = "ODS_MEMBER"
_MEMBER_TABLE = "ods.member_profiles"
_MEMBER_ENDPOINT = "/MemberProfile/GetTenantMemberList"

# member_profiles 表列定义（含 birthday DATE 列 — 迁移 C1 后的状态）
_MEMBER_COLUMNS_WITH_BIRTHDAY = [
    ("id", "bigint", "int8"),
    ("record_index", "integer", "int4"),
    ("content_hash", "text", "text"),
    ("payload", "jsonb", "jsonb"),
    ("birthday", "date", "date"),
]


# ---------------------------------------------------------------------------
# FakeDB：支持自定义列定义
# ---------------------------------------------------------------------------
class _MemberFakeCursor(FakeCursor):
    """扩展 FakeCursor，支持自定义列定义和 PK 查询。"""

    def __init__(self, recorder, db_ops=None, columns_map=None, pk_map=None):
        super().__init__(recorder, db_ops)
        self._columns_map = columns_map or {}
        self._pk_map = pk_map or {}

    def execute(self, sql, params=None):
        sql_text = sql.decode("utf-8", errors="ignore") if isinstance(sql, (bytes, bytearray)) else str(sql)
        self.recorder.append({"sql": sql_text.strip(), "params": params})
        self._fetchall_rows = []

        lowered = sql_text.lower()

        # 拦截 information_schema.columns 查询 → 返回自定义列定义
        if "from information_schema.columns" in lowered:
            table_name = None
            if params and len(params) >= 2:
                table_name = params[1]
            full_table = f"{params[0]}.{params[1]}" if params and len(params) >= 2 else None
            if full_table and full_table in self._columns_map:
                self._fetchall_rows = list(self._columns_map[full_table])
            elif table_name and table_name in self._columns_map:
                self._fetchall_rows = list(self._columns_map[table_name])
            else:
                self._fetchall_rows = self._fake_columns(table_name)
            return

        # 拦截 PK 约束查询
        if "from information_schema.table_constraints" in lowered:
            table_name = None
            if params and len(params) >= 2:
                table_name = params[1]
            full_table = f"{params[0]}.{params[1]}" if params and len(params) >= 2 else None
            if full_table and full_table in self._pk_map:
                self._fetchall_rows = [(col,) for col in self._pk_map[full_table]]
            elif table_name and table_name in self._pk_map:
                self._fetchall_rows = [(col,) for col in self._pk_map[table_name]]
            else:
                self._fetchall_rows = [("id",)]
            return

        # 拦截 content_hash 查询
        if "distinct on" in lowered and "content_hash" in lowered:
            self._fetchall_rows = []
            self._pending_rows = []
            return

        # 默认：处理 INSERT 等语句
        if self._pending_rows:
            self.rowcount = len(self._pending_rows)
            self._record_upserts(sql_text)
            if "returning" in lowered:
                self._fetchall_rows = [(True,)] * len(self._pending_rows)
            self._pending_rows = []
        else:
            self.rowcount = 0


class _MemberFakeConnection(FakeConnection):
    def __init__(self, db_ops, columns_map=None, pk_map=None):
        super().__init__(db_ops)
        self._columns_map = columns_map or {}
        self._pk_map = pk_map or {}

    def cursor(self):
        return _MemberFakeCursor(
            self.statements, self._db_ops,
            columns_map=self._columns_map,
            pk_map=self._pk_map,
        )


class _MemberFakeDB(FakeDBOperations):
    def __init__(self, columns_map=None, pk_map=None):
        super().__init__()
        self._columns_map = columns_map or {}
        self._pk_map = pk_map or {}
        self.conn = _MemberFakeConnection(
            self, self._columns_map, self._pk_map,
        )


# ---------------------------------------------------------------------------
# 辅助
# ---------------------------------------------------------------------------
def _build_config(tmp_path):
    archive_dir = tmp_path / "archive"
    temp_dir = tmp_path / "temp"
    cfg = create_test_config("ONLINE", archive_dir, temp_dir)
    cfg.config.setdefault("run", {})
    cfg.config["run"]["ods_conflict_mode"] = "update"
    cfg.config["run"]["snapshot_missing_delete"] = False
    return cfg


def _get_member_spec():
    for spec in ODS_TASK_SPECS:
        if spec.code == _MEMBER_CODE:
            return spec
    raise KeyError(f"未找到任务 spec: {_MEMBER_CODE}")


# ---------------------------------------------------------------------------
# 测试
# ---------------------------------------------------------------------------
class TestOdsBirthdayExtraction:
    """验证 ODS_MEMBER 任务在表包含 birthday 列时能正确提取该字段。"""

    def test_birthday_extracted_from_json_payload(self, tmp_path):
        """当 ods.member_profiles 包含 birthday 列时，
        API JSON 中的 birthday 值应被提取并写入对应列位置。
        """
        spec = _get_member_spec()
        task_cls = ODS_TASK_CLASSES[_MEMBER_CODE]

        config = _build_config(tmp_path)
        db = _MemberFakeDB(
            columns_map={_MEMBER_TABLE: _MEMBER_COLUMNS_WITH_BIRTHDAY},
            pk_map={_MEMBER_TABLE: ["id", "content_hash"]},
        )
        # 模拟 API 返回包含 birthday 的会员记录
        member_record = {
            "id": 12345,
            "birthday": "1990-05-15",
            "nickname": "测试会员",
            "mobile": "13800138000",
        }
        api = FakeAPIClient({spec.endpoint: [member_record]})
        logger = logging.getLogger("test_birthday")

        task = task_cls(config, db, api, logger)
        result = task.execute()

        counts = result["counts"]
        assert counts["fetched"] == 1
        assert counts["inserted"] == 1

        # 验证写入的 SQL 参数中包含 birthday 值
        insert_stmts = [
            s for s in db.conn.statements
            if "INSERT" in s.get("sql", "").upper() and "member_profiles" in s.get("sql", "")
        ]
        assert len(insert_stmts) >= 1, "应有至少一条 INSERT 语句"

        # INSERT SQL 应包含 birthday 列
        assert any("birthday" in s["sql"].lower() for s in insert_stmts), (
            "INSERT SQL 应包含 birthday 列"
        )

    def test_birthday_null_when_not_in_json(self, tmp_path):
        """当 API JSON 中不包含 birthday 字段时，
        对应列应写入 None。
        """
        spec = _get_member_spec()
        task_cls = ODS_TASK_CLASSES[_MEMBER_CODE]

        config = _build_config(tmp_path)
        db = _MemberFakeDB(
            columns_map={_MEMBER_TABLE: _MEMBER_COLUMNS_WITH_BIRTHDAY},
            pk_map={_MEMBER_TABLE: ["id", "content_hash"]},
        )
        # API 返回不含 birthday 的记录
        member_record = {
            "id": 67890,
            "nickname": "无生日会员",
        }
        api = FakeAPIClient({spec.endpoint: [member_record]})
        logger = logging.getLogger("test_birthday_null")

        task = task_cls(config, db, api, logger)
        result = task.execute()

        counts = result["counts"]
        assert counts["fetched"] == 1
        assert counts["inserted"] == 1

    def test_birthday_column_in_insert_sql(self, tmp_path):
        """INSERT SQL 中应包含 birthday 列名。"""
        spec = _get_member_spec()
        task_cls = ODS_TASK_CLASSES[_MEMBER_CODE]

        config = _build_config(tmp_path)
        db = _MemberFakeDB(
            columns_map={_MEMBER_TABLE: _MEMBER_COLUMNS_WITH_BIRTHDAY},
            pk_map={_MEMBER_TABLE: ["id", "content_hash"]},
        )
        member_record = {"id": 11111, "birthday": "2000-01-01"}
        api = FakeAPIClient({spec.endpoint: [member_record]})
        logger = logging.getLogger("test_birthday_sql")

        task = task_cls(config, db, api, logger)
        task.execute()

        # 找到 INSERT 语句，确认包含 birthday 列
        insert_stmts = [
            s for s in db.conn.statements
            if "INSERT" in s.get("sql", "").upper() and "member_profiles" in s.get("sql", "")
        ]
        assert any("birthday" in s["sql"].lower() for s in insert_stmts), (
            "INSERT SQL 应包含 birthday 列，"
            f"实际 SQL: {[s['sql'][:200] for s in insert_stmts]}"
        )

    def test_ods_member_spec_has_no_extra_columns(self):
        """确认 ODS_MEMBER 任务没有 extra_columns（依赖 schema-aware 动态映射）。

        这意味着 birthday 字段无需在 OdsTaskSpec 中显式声明，
        只要 DB 表有该列，就会自动从 JSON 提取。
        """
        spec = _get_member_spec()
        assert spec.extra_columns == (), (
            "ODS_MEMBER 不应有 extra_columns，"
            "它依赖 schema-aware 动态入库自动映射所有 DB 列"
        )
        assert spec.table_name == "ods.member_profiles"
        assert spec.code == "ODS_MEMBER"
