import hashlib
import json
import shutil
import tempfile
import unittest
from dataclasses import replace
from datetime import date
from pathlib import Path
from unittest.mock import patch

import pandas as pd

from roi_control.data_source import build_daily_sql
from roi_control.fission_multiplier import (
    DEFAULT_FISSION_PARAMETER_VERSION,
    DISPLAY_MULTIPLIER_COLUMN,
    DISPLAY_TOTAL_TO_FIRST_COLUMN,
    FALLBACK_FILE,
    GZH_FILE,
    LEGACY_FISSION_PARAMETER_VERSION,
    MATCH_GZH_EXACT,
    MATCH_GZH_PARTNER,
    MATCH_MINIAPP_EXACT,
    MATCH_MINIAPP_GOAL,
    MATCH_MINIAPP_PACKAGE,
    MATCH_QIWEI_REFERENCE,
    MATCH_QIWEI_PARTNER,
    MINIAPP_FILE,
    ParameterRelease,
    QIWEI_FILE,
    QIWEI_REFERENCE_MULTIPLIER,
    QIWEI_REFERENCE_VERSION,
    RELEASES,
    SOURCE_TOTAL_TO_FIRST_COLUMN,
    apply_fission_multiplier,
    load_fission_multiplier_parameters,
    parameter_content_sha256,
    parameter_values,
    parameters_from_database,
)
from roi_control.metrics import prepare_daily_metrics
from roi_control.fission_multiplier_refresh import (
    build_gzh_activity_curve_sql,
    build_miniapp_activity_curve_sql,
    build_qiwei_activity_curve_sql,
    build_qiwei_release_frame,
)


PROJECT_DIR = Path(__file__).resolve().parent


class RoiFissionMultiplierTest(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.parameters = load_fission_multiplier_parameters()

    def test_published_release_metadata_and_counts(self):
        snapshot = self.parameters.snapshot()
        self.assertEqual(snapshot["version"], DEFAULT_FISSION_PARAMETER_VERSION)
        self.assertEqual(snapshot["cohort_date"], "20260712")
        self.assertEqual(snapshot["observation_end_date"], "20260727")
        self.assertEqual(snapshot["miniapp_exact_rows"], 40)
        self.assertEqual(snapshot["miniapp_exact_available_rows"], 21)
        self.assertEqual(snapshot["gzh_exact_rows"], 110)
        self.assertEqual(snapshot["gzh_exact_available_rows"], 93)
        self.assertAlmostEqual(snapshot["miniapp_channel_multiplier"], 2.608491, places=6)
        self.assertAlmostEqual(snapshot["gzh_channel_multiplier"], 2.476153, places=6)
        self.assertFalse(self.parameters.qiwei_by_partner)
        self.assertNotIn("qiwei_formal", snapshot)
        self.assertNotIn("qiwei_channel_multiplier", snapshot)
        self.assertFalse(
            any(row["entity_type"] == "qiwei" for row in parameter_values(self.parameters))
        )
        self.assertEqual(
            parameter_content_sha256(self.parameters),
            "af62237a0f6ee8de75ad90f9be82b0c880a783ee5a8ecb3e42c27b777e6dd213",
        )
        legacy = load_fission_multiplier_parameters(LEGACY_FISSION_PARAMETER_VERSION)
        self.assertEqual(
            parameter_content_sha256(legacy),
            "c447dc53fb560a9ca7dd954b27966ec46bb4e25e1dfd6916d607431c106242c6",
        )

    def test_miniapp_uses_exact_then_published_fallbacks(self):
        exact = self.parameters.match(
            "self",
            package="泛人群",
            optimize_goal="关键页面访问次数",
        )
        self.assertEqual(exact.match_level, MATCH_MINIAPP_EXACT)
        self.assertAlmostEqual(exact.multiplier, 2.515898565120092)

        missing_goal = self.parameters.match(
            "self",
            package="泛人群",
            optimize_goal="",
        )
        self.assertEqual(missing_goal.match_level, MATCH_MINIAPP_PACKAGE)
        self.assertAlmostEqual(missing_goal.multiplier, 2.5182863849765257)

        insufficient_exact = self.parameters.match(
            "self",
            package="unexp*day*商业",
            optimize_goal="关键页面访问次数",
        )
        self.assertEqual(insufficient_exact.match_level, MATCH_MINIAPP_GOAL)
        self.assertAlmostEqual(insufficient_exact.multiplier, 2.610471, places=6)

    def test_r500_uses_fixed_cohort_coefficients(self):
        match = self.parameters.match(
            "self",
            package="R500*全品类",
            optimize_goal="关键页面访问次数",
        )
        self.assertEqual(match.match_level, MATCH_MINIAPP_EXACT)
        self.assertAlmostEqual(match.multiplier, 49567 / 18467)
        self.assertAlmostEqual(match.multiplier_vs_first, 49567 / 7068)

    def test_gzh_insufficient_exact_uses_partner_fallback(self):
        exact = self.parameters.match(
            "gzh", partner="szhx", official_account="尊闻"
        )
        self.assertEqual(exact.match_level, MATCH_GZH_EXACT)

        fallback = self.parameters.match(
            "gzh", partner="yy", official_account="异闻怪谈"
        )
        self.assertEqual(fallback.match_level, MATCH_GZH_PARTNER)
        self.assertAlmostEqual(fallback.multiplier, 1.887225, places=6)

    def test_qiwei_uses_reference_multiplier_without_control_semantics(self):
        match = self.parameters.match("qiwei", partner="企微合作方")
        self.assertEqual(match.match_level, MATCH_QIWEI_REFERENCE)
        self.assertEqual(match.multiplier, QIWEI_REFERENCE_MULTIPLIER)
        self.assertEqual(QIWEI_REFERENCE_MULTIPLIER, 2.5)
        self.assertIn("待发布合作方参数", match.source)

    def test_qiwei_roi_uses_reference_multiplier(self):
        frame = pd.DataFrame(
            [
                {
                    "dt": "20260727",
                    "entity_type": "qiwei",
                    "channel": "群/企微合作-稳定",
                    "代理名称": "",
                    "账号id": "",
                    "账号名称": "",
                    "广告id": "",
                    "广告名称": "",
                    "包名": "",
                    "广告优化目标": "",
                    "创意id": "",
                    "合作方名": "企微合作方",
                    "公众号名": "",
                    "首层UV": 100,
                    "T0裂变数": 20,
                    "成本": 100,
                    "效率收入": 50,
                    "裂变效率收入": 30,
                }
            ]
        )
        result = prepare_daily_metrics(frame, self.parameters).iloc[0]
        self.assertAlmostEqual(
            result["预测全链路效率收入"],
            50 + 30 * QIWEI_REFERENCE_MULTIPLIER,
        )
        self.assertAlmostEqual(
            result["ROI"],
            (50 + 30 * QIWEI_REFERENCE_MULTIPLIER) / 100,
        )
        self.assertEqual(result["传播裂变参数版本"], QIWEI_REFERENCE_VERSION)
        self.assertEqual(result["传播裂变参数cohort日期"], "")
        self.assertEqual(result["调控参与状态"], "仅展示_不进入阈值和调控")

    def test_qiwei_parameters_match_partner_then_default(self):
        formal = replace(
            self.parameters,
            qiwei_by_partner={"合作方A": 2.2},
            total_to_first={
                **self.parameters.total_to_first,
                ("qiwei", MATCH_QIWEI_PARTNER, "合作方A", ""): 0.75,
            },
            qiwei_exact_rows=2,
            qiwei_exact_available_rows=1,
        )

        exact = formal.match("qiwei", partner="合作方A")
        fallback = formal.match("qiwei", partner="未知合作方")
        self.assertEqual(exact.match_level, MATCH_QIWEI_PARTNER)
        self.assertEqual(exact.multiplier, 2.2)
        self.assertEqual(exact.multiplier_vs_first, 0.75)
        self.assertEqual(fallback.match_level, MATCH_QIWEI_REFERENCE)
        self.assertEqual(fallback.multiplier, QIWEI_REFERENCE_MULTIPLIER)

    def test_qiwei_formal_and_default_rows_keep_separate_audit_fields(self):
        formal = replace(
            self.parameters,
            qiwei_by_partner={"合作方A": 2.2},
            total_to_first={
                **self.parameters.total_to_first,
                ("qiwei", MATCH_QIWEI_PARTNER, "合作方A", ""): 0.75,
            },
            qiwei_exact_rows=2,
            qiwei_exact_available_rows=1,
        )
        daily = pd.DataFrame(
            [
                {
                    "entity_type": "qiwei",
                    "包名": "",
                    "广告优化目标": "",
                    "合作方名": "合作方A",
                    "公众号名": "",
                },
                {
                    "entity_type": "qiwei",
                    "包名": "",
                    "广告优化目标": "",
                    "合作方名": "未知合作方",
                    "公众号名": "",
                },
            ]
        )

        result = apply_fission_multiplier(daily, formal)
        exact, fallback = result.iloc[0], result.iloc[1]
        self.assertEqual(exact["传播裂变参数版本"], formal.release.version)
        self.assertEqual(exact["传播裂变参数状态"], "企微正式参数_仅展示")
        self.assertEqual(fallback["传播裂变参数版本"], QIWEI_REFERENCE_VERSION)
        self.assertEqual(
            fallback[DISPLAY_MULTIPLIER_COLUMN],
            QIWEI_REFERENCE_MULTIPLIER,
        )
        self.assertIn("默认参考系数", fallback["传播裂变参数状态"])

    def test_formal_qiwei_database_round_trip(self):
        formal = replace(
            self.parameters,
            qiwei_by_partner={"合作方A": 2.2},
            total_to_first={
                **self.parameters.total_to_first,
                ("qiwei", MATCH_QIWEI_PARTNER, "合作方A", ""): 0.75,
            },
            qiwei_exact_rows=2,
            qiwei_exact_available_rows=1,
        )
        metadata = formal.snapshot()
        metadata["file_sha256"] = dict(formal.release.file_sha256)
        restored = parameters_from_database(
            {
                "version": formal.release.version,
                "cohort_date": date(2026, 7, 12),
                "observation_end_date": date(2026, 7, 27),
                "run_suffix": formal.release.run_suffix,
                "content_sha256": parameter_content_sha256(formal),
                "metadata_json": json.dumps(metadata, ensure_ascii=False),
            },
            parameter_values(formal),
        )

        self.assertEqual(restored.qiwei_by_partner, {"合作方A": 2.2})
        self.assertNotIn("qiwei_formal", restored.snapshot())
        self.assertNotIn("qiwei_channel_multiplier", restored.snapshot())
        self.assertEqual(
            parameter_content_sha256(restored),
            parameter_content_sha256(formal),
        )

    def test_formal_qiwei_parameter_files_are_loaded_and_serialized(self):
        version = "20260712_A0-A15_qiwei_test"
        source = (
            PROJECT_DIR
            / "roi_control"
            / "data"
            / "fission_multiplier"
            / self.parameters.release.source_version
        )
        with tempfile.TemporaryDirectory() as temporary:
            parameter_root = Path(temporary)
            target = parameter_root / version
            target.mkdir()
            shutil.copy2(source / MINIAPP_FILE, target / MINIAPP_FILE)
            shutil.copy2(source / GZH_FILE, target / GZH_FILE)
            shutil.copy2(source / FALLBACK_FILE, target / FALLBACK_FILE)

            qiwei = pd.DataFrame(
                [
                    {
                        "首层投放日期": "20260712",
                        "合作方": "合作方A",
                        "首层UV": 600,
                        "A0活跃裂变UV": 200,
                        "A0-A15累计裂变活跃人次": 400,
                        SOURCE_TOTAL_TO_FIRST_COLUMN: 400 / 600,
                        "T1-T15相对T0尾部系数": 1.0,
                        "T15成熟系数（相对T0）": 2.0,
                        "样本状态": "可用",
                    }
                ]
            )
            qiwei.to_csv(target / QIWEI_FILE, index=False, encoding="utf-8-sig")

            def sha256(path: Path) -> str:
                return hashlib.sha256(path.read_bytes()).hexdigest()

            release = ParameterRelease(
                version=version,
                cohort_date="20260712",
                observation_end_date="20260727",
                run_suffix="qtest",
                file_sha256={
                    name: sha256(target / name)
                    for name in (MINIAPP_FILE, GZH_FILE, QIWEI_FILE, FALLBACK_FILE)
                },
                source_version=version,
                includes_total_to_first=True,
            )
            with patch.dict(RELEASES, {version: release}):
                parameters = load_fission_multiplier_parameters(
                    version,
                    parameter_root=parameter_root,
                )

        self.assertEqual(parameters.qiwei_by_partner, {"合作方A": 2.0})
        qiwei_values = [
            row for row in parameter_values(parameters) if row["entity_type"] == "qiwei"
        ]
        self.assertEqual(len(qiwei_values), 1)
        self.assertEqual(qiwei_values[0]["match_level"], MATCH_QIWEI_PARTNER)

    def test_database_parameter_round_trip_preserves_release(self):
        metadata = self.parameters.snapshot()
        metadata["file_sha256"] = dict(self.parameters.release.file_sha256)
        restored = parameters_from_database(
            {
                "version": self.parameters.release.version,
                "cohort_date": date(2026, 7, 12),
                "observation_end_date": date(2026, 7, 27),
                "run_suffix": self.parameters.release.run_suffix,
                "content_sha256": parameter_content_sha256(self.parameters),
                "metadata_json": json.dumps(metadata, ensure_ascii=False),
            },
            parameter_values(self.parameters),
        )
        self.assertEqual(
            parameter_content_sha256(restored),
            parameter_content_sha256(self.parameters),
        )
        self.assertEqual(restored.snapshot(), self.parameters.snapshot())

    def test_roi_formula_does_not_add_t0_twice(self):
        frame = pd.DataFrame(
            [
                {
                    "dt": "20260727",
                    "entity_type": "self",
                    "channel": "小程序投流-稳定",
                    "代理名称": "",
                    "账号id": "1",
                    "账号名称": "测试账户",
                    "广告id": "2",
                    "广告名称": "测试广告",
                    "包名": "泛人群",
                    "广告优化目标": "关键页面访问次数",
                    "创意id": "3",
                    "合作方名": "",
                    "公众号名": "",
                    "首层UV": 100,
                    "T0裂变数": 20,
                    "成本": 100,
                    "效率收入": 50,
                    "裂变效率收入": 30,
                }
            ]
        )
        result = prepare_daily_metrics(frame, self.parameters).iloc[0]
        expected_tail = 30 * (result[DISPLAY_MULTIPLIER_COLUMN] - 1)
        expected_fission = 30 + expected_tail
        self.assertAlmostEqual(result["T0实际裂变收入"], 30)
        self.assertAlmostEqual(result["预测T1-T15裂变收入"], expected_tail)
        self.assertAlmostEqual(result["预测T0-T15裂变收入"], expected_fission)
        self.assertAlmostEqual(result["预测全链路效率收入"], 50 + expected_fission)
        self.assertAlmostEqual(result["ROI"], (50 + expected_fission) / 100)
        self.assertAlmostEqual(result["实际ROI"], 0.8)
        self.assertAlmostEqual(
            result[DISPLAY_TOTAL_TO_FIRST_COLUMN],
            self.parameters.match(
                "self",
                package="泛人群",
                optimize_goal="关键页面访问次数",
            ).multiplier_vs_first,
        )

    def test_daily_sql_preserves_explicit_optimize_goal(self):
        sql = build_daily_sql("20260725", "20260727")
        self.assertIn("AS 广告优化目标", sql)
        self.assertIn("广告优化目标='null'", sql)
        self.assertIn("GROUP BY\n  dt, channel, 账号id, 广告id, 包名", sql)
        self.assertEqual(
            sql.count("SUM(NVL(裂变效率收入, 0)) AS 裂变效率收入"),
            3,
        )
        self.assertNotIn("多层t0裂变收入", sql)

    def test_recompute_sql_matches_published_release_audit_sql(self):
        audit_dir = (
            PROJECT_DIR
            / "roi_control"
            / "sql"
            / "fission_multiplier"
            / self.parameters.release.source_version
        )
        expected_miniapp = (
            audit_dir / "小程序人群包转化目标_A0-A15_20260712.sql"
        ).read_text(encoding="utf-8").strip()
        expected_gzh = (
            audit_dir / "公众号合作即转_A0-A15_20260712.sql"
        ).read_text(encoding="utf-8").strip()
        self.assertEqual(
            build_miniapp_activity_curve_sql("20260712"),
            expected_miniapp,
        )
        self.assertEqual(
            build_gzh_activity_curve_sql("20260712"),
            expected_gzh,
        )

    def test_qiwei_recompute_sql_groups_by_partner(self):
        sql = build_qiwei_activity_curve_sql("20260712")

        self.assertIn("d.channel='群/企微合作-稳定'", sql)
        self.assertIn("JOIN loghubods.user_share_log l", sql)
        self.assertIn(
            "ON d.subsessionid=l.subsessionid AND d.mid=l.machinecode",
            sql,
        )
        self.assertIn("d.dt='20260712' AND l.dt='20260712'", sql)
        self.assertIn("FROM loghubods.ods_user_active_log_info_day", sql)
        self.assertIn("user_share_depth=0", sql)
        self.assertIn("a.machinecode=b.machinecode", sql)
        self.assertIn("a.subsessionid=b.subsessionid", sql)
        self.assertIn("b.machinecode IS NULL", sql)
        self.assertIn("a.topic='click'", sql)
        self.assertIn("CAST(a.usersharedepth AS BIGINT)>0", sql)
        self.assertIn("s.dt BETWEEN '20260712' AND '20260727'", sql)
        self.assertIn("GROUP BY partner_name", sql)
        self.assertIn("HAVING COUNT(DISTINCT first_mid)>500", sql)
        self.assertIn("AS a0_active_uv", sql)
        self.assertIn("AS a15_active_uv", sql)
        self.assertIn("END AS activity_maturity_vs_a0", sql)
        self.assertNotIn("-NVL(a.a0_active_uv,0)", sql)
        self.assertNotIn("account_name", sql)
        self.assertNotIn("公众号名", sql)

    def test_qiwei_recompute_sql_validates_arguments(self):
        with self.assertRaisesRegex(ValueError, "horizon不能小于0"):
            build_qiwei_activity_curve_sql("20260712", horizon=-1)
        with self.assertRaisesRegex(ValueError, "首层UV门槛不能小于0"):
            build_qiwei_activity_curve_sql("20260712", min_first_uv=-1)

    def test_qiwei_release_frame_has_no_channel_fallback(self):
        row = {
            "cohort_dt": "20260712",
            "partner_name": "合作方A",
            "first_uv": 600,
            "root_session_cnt": 600,
            "active_person_days": 400,
            "activity_rate_sum": 2 / 3,
            "active_tail_vs_a0": 1.0,
            "activity_maturity_vs_a0": 2.0,
            "steady_dau_multiplier": 5 / 3,
        }
        for lag in range(16):
            row[f"a{lag}_active_uv"] = 200 if lag == 0 else 0
            row[f"a{lag}_active_rate"] = 1 / 3 if lag == 0 else 0
        frame = build_qiwei_release_frame(pd.DataFrame([row]))
        self.assertEqual(frame["参数层级"].tolist(), ["企微合作方"])
        self.assertNotIn("企微渠道回退", frame.to_string())

if __name__ == "__main__":
    unittest.main()
