"""Versioned T15 propagation-fission multipliers relative to T0.

This module is deliberately independent from ODPS, policy, persistence,
approval, and Tencent writes. Daily ROI consumes a published parameter release;
it never recomputes cohort parameters during a production run.
"""

from __future__ import annotations

import hashlib
import json
import math
from dataclasses import dataclass
from pathlib import Path
from typing import Mapping

import pandas as pd


LEGACY_FISSION_PARAMETER_VERSION = "20260712_A0-A15_v1"
DEFAULT_FISSION_PARAMETER_VERSION = "20260712_A0-A15_v2"
FISSION_HORIZON_DAYS = 15
PARAMETER_ROOT = Path(__file__).resolve().parent / "data" / "fission_multiplier"
SOURCE_MULTIPLIER_COLUMN = "T15成熟系数（相对T0）"
DISPLAY_MULTIPLIER_COLUMN = "传播裂变系数-对T0裂变"
SOURCE_TOTAL_TO_FIRST_COLUMN = "每首层累计裂变活跃人次"
DISPLAY_TOTAL_TO_FIRST_COLUMN = "裂变系数-总裂变UV/首层UV"

MINIAPP_FILE = "小程序T15成熟系数_中文_20260712.csv"
GZH_FILE = "公众号T15成熟系数_中文_20260712.csv"
QIWEI_FILE = "企微T15成熟系数_中文_20260712.csv"
FALLBACK_FILE = "T15分级回退系数_中文_20260712.csv"

MATCH_MINIAPP_EXACT = "miniapp_package_goal_exact"
MATCH_MINIAPP_PACKAGE = "miniapp_package_fallback"
MATCH_MINIAPP_GOAL = "miniapp_goal_fallback"
MATCH_MINIAPP_CHANNEL = "miniapp_channel_fallback"
MATCH_GZH_EXACT = "gzh_partner_account_exact"
MATCH_GZH_PARTNER = "gzh_partner_fallback"
MATCH_GZH_CHANNEL = "gzh_channel_fallback"
MATCH_QIWEI_PARTNER = "qiwei_partner_exact"
MATCH_QIWEI_REFERENCE = "qiwei_reference_pending"
QIWEI_REFERENCE_MULTIPLIER = 2.5
QIWEI_REFERENCE_VERSION = "qiwei_reference_2p5_v1"
QIWEI_REFERENCE_RUN_SUFFIX = "q2p5v1"


@dataclass(frozen=True)
class ParameterRelease:
    version: str
    cohort_date: str
    observation_end_date: str
    run_suffix: str
    file_sha256: Mapping[str, str]
    source_version: str
    includes_total_to_first: bool


RELEASES: Mapping[str, ParameterRelease] = {
    LEGACY_FISSION_PARAMETER_VERSION: ParameterRelease(
        version=LEGACY_FISSION_PARAMETER_VERSION,
        cohort_date="20260712",
        observation_end_date="20260727",
        run_suffix="f0712v1",
        file_sha256={
            MINIAPP_FILE: "9358408ad4957bbd22f689f8bbf66b7abe765f210e5af444d11e329815fee6a4",
            GZH_FILE: "1f6f17b1403aa9b56a33195da3c324fbe06cbb714aa2ebabcdbceb891aa7a951",
            FALLBACK_FILE: "6e4d517595ec2632dc54a65000abe94000b5313e77420547759b2419314d60ac",
        },
        source_version=LEGACY_FISSION_PARAMETER_VERSION,
        includes_total_to_first=False,
    ),
    DEFAULT_FISSION_PARAMETER_VERSION: ParameterRelease(
        version=DEFAULT_FISSION_PARAMETER_VERSION,
        cohort_date="20260712",
        observation_end_date="20260727",
        run_suffix="f0712v2",
        file_sha256={
            MINIAPP_FILE: "9358408ad4957bbd22f689f8bbf66b7abe765f210e5af444d11e329815fee6a4",
            GZH_FILE: "1f6f17b1403aa9b56a33195da3c324fbe06cbb714aa2ebabcdbceb891aa7a951",
            FALLBACK_FILE: "6e4d517595ec2632dc54a65000abe94000b5313e77420547759b2419314d60ac",
        },
        source_version=LEGACY_FISSION_PARAMETER_VERSION,
        includes_total_to_first=True,
    ),
}


@dataclass(frozen=True)
class FissionMultiplierMatch:
    multiplier: float
    multiplier_vs_first: float | None
    match_level: str
    source: str


@dataclass(frozen=True)
class FissionMultiplierParameters:
    release: ParameterRelease
    miniapp_exact: Mapping[tuple[str, str], float]
    miniapp_by_package: Mapping[str, float]
    miniapp_by_goal: Mapping[str, float]
    miniapp_channel: float
    gzh_exact: Mapping[tuple[str, str], float]
    gzh_by_partner: Mapping[str, float]
    gzh_channel: float
    total_to_first: Mapping[tuple[str, str, str, str], float]
    qiwei_by_partner: Mapping[str, float]
    miniapp_exact_rows: int
    miniapp_exact_available_rows: int
    gzh_exact_rows: int
    gzh_exact_available_rows: int
    qiwei_exact_rows: int
    qiwei_exact_available_rows: int

    def snapshot(self) -> dict[str, object]:
        snapshot = {
            "version": self.release.version,
            "cohort_date": self.release.cohort_date,
            "observation_end_date": self.release.observation_end_date,
            "horizon_days": FISSION_HORIZON_DAYS,
            "miniapp_exact_rows": self.miniapp_exact_rows,
            "miniapp_exact_available_rows": self.miniapp_exact_available_rows,
            "gzh_exact_rows": self.gzh_exact_rows,
            "gzh_exact_available_rows": self.gzh_exact_available_rows,
            "miniapp_channel_multiplier": self.miniapp_channel,
            "gzh_channel_multiplier": self.gzh_channel,
        }
        if self.release.includes_total_to_first:
            snapshot["includes_total_to_first"] = True
            snapshot["source_version"] = self.release.source_version
        # Keep legacy release hashes stable until a release explicitly carries
        # audited WeCom parameters.
        if self.qiwei_by_partner:
            snapshot.update(
                {
                    "qiwei_exact_rows": self.qiwei_exact_rows,
                    "qiwei_exact_available_rows": self.qiwei_exact_available_rows,
                }
            )
        return snapshot

    def _match_result(
        self,
        multiplier: float,
        entity_type: str,
        match_level: str,
        key_primary: str,
        key_secondary: str,
        source: str,
    ) -> FissionMultiplierMatch:
        return FissionMultiplierMatch(
            multiplier=multiplier,
            multiplier_vs_first=self.total_to_first.get(
                (entity_type, match_level, key_primary, key_secondary)
            ),
            match_level=match_level,
            source=source,
        )

    def match(
        self,
        entity_type: str,
        *,
        package: object = "",
        optimize_goal: object = "",
        partner: object = "",
        official_account: object = "",
    ) -> FissionMultiplierMatch:
        if entity_type == "self":
            package_key = normalize_package(package)
            goal_key = normalize_name(optimize_goal)
            exact_key = (package_key, goal_key)
            if goal_key and exact_key in self.miniapp_exact:
                return self._match_result(
                    self.miniapp_exact[exact_key],
                    "self",
                    MATCH_MINIAPP_EXACT,
                    package_key,
                    goal_key,
                    f"人群包+转化目标精确:{package_key}|{goal_key}",
                )
            if package_key in self.miniapp_by_package:
                return self._match_result(
                    self.miniapp_by_package[package_key],
                    "self",
                    MATCH_MINIAPP_PACKAGE,
                    package_key,
                    "",
                    f"人群包回退:{package_key}",
                )
            if goal_key and goal_key in self.miniapp_by_goal:
                return self._match_result(
                    self.miniapp_by_goal[goal_key],
                    "self",
                    MATCH_MINIAPP_GOAL,
                    goal_key,
                    "",
                    f"转化目标回退:{goal_key}",
                )
            return self._match_result(
                self.miniapp_channel,
                "self",
                MATCH_MINIAPP_CHANNEL,
                "",
                "",
                "小程序渠道回退",
            )

        if entity_type == "gzh":
            partner_key = normalize_name(partner)
            account_key = normalize_name(official_account)
            exact_key = (partner_key, account_key)
            if partner_key and account_key and exact_key in self.gzh_exact:
                return self._match_result(
                    self.gzh_exact[exact_key],
                    "gzh",
                    MATCH_GZH_EXACT,
                    partner_key,
                    account_key,
                    f"合作方+公众号精确:{partner_key}|{account_key}",
                )
            if partner_key in self.gzh_by_partner:
                return self._match_result(
                    self.gzh_by_partner[partner_key],
                    "gzh",
                    MATCH_GZH_PARTNER,
                    partner_key,
                    "",
                    f"合作方回退:{partner_key}",
                )
            return self._match_result(
                self.gzh_channel,
                "gzh",
                MATCH_GZH_CHANNEL,
                "",
                "",
                "公众号渠道回退",
            )

        if entity_type == "qiwei":
            partner_key = normalize_name(partner)
            if partner_key and partner_key in self.qiwei_by_partner:
                return self._match_result(
                    self.qiwei_by_partner[partner_key],
                    "qiwei",
                    MATCH_QIWEI_PARTNER,
                    partner_key,
                    "",
                    f"企微合作方精确:{partner_key}",
                )
            return FissionMultiplierMatch(
                QIWEI_REFERENCE_MULTIPLIER,
                None,
                MATCH_QIWEI_REFERENCE,
                f"企微默认参考系数{QIWEI_REFERENCE_MULTIPLIER:g}_待发布合作方参数:"
                f"{partner_key}",
            )

        raise ValueError(f"不支持的ROI实体类型: {entity_type}")


def parameter_values(
    parameters: FissionMultiplierParameters,
) -> list[dict[str, object]]:
    """Serialize one validated release into database lookup rows."""

    rows: list[dict[str, object]] = []

    def append(
        entity_type: str,
        match_level: str,
        key_primary: str,
        key_secondary: str,
        multiplier: float,
    ) -> None:
        row = {
            "entity_type": entity_type,
            "match_level": match_level,
            "key_primary": key_primary,
            "key_secondary": key_secondary,
            "multiplier": float(multiplier),
        }
        if parameters.release.includes_total_to_first:
            key = (entity_type, match_level, key_primary, key_secondary)
            multiplier_vs_first = parameters.total_to_first.get(key)
            if multiplier_vs_first is None:
                raise ValueError(f"传播裂变系数缺少总裂变UV/首层UV参数: {key}")
            row["multiplier_vs_first"] = float(multiplier_vs_first)
        rows.append(row)

    for (package, goal), multiplier in parameters.miniapp_exact.items():
        append("self", MATCH_MINIAPP_EXACT, package, goal, multiplier)
    for package, multiplier in parameters.miniapp_by_package.items():
        append("self", MATCH_MINIAPP_PACKAGE, package, "", multiplier)
    for goal, multiplier in parameters.miniapp_by_goal.items():
        append("self", MATCH_MINIAPP_GOAL, goal, "", multiplier)
    append("self", MATCH_MINIAPP_CHANNEL, "", "", parameters.miniapp_channel)

    for (partner, account), multiplier in parameters.gzh_exact.items():
        append("gzh", MATCH_GZH_EXACT, partner, account, multiplier)
    for partner, multiplier in parameters.gzh_by_partner.items():
        append("gzh", MATCH_GZH_PARTNER, partner, "", multiplier)
    append("gzh", MATCH_GZH_CHANNEL, "", "", parameters.gzh_channel)

    for partner, multiplier in parameters.qiwei_by_partner.items():
        append("qiwei", MATCH_QIWEI_PARTNER, partner, "", multiplier)
    return sorted(
        rows,
        key=lambda row: (
            str(row["entity_type"]),
            str(row["match_level"]),
            str(row["key_primary"]),
            str(row["key_secondary"]),
        ),
    )


def parameter_content_sha256(
    parameters: FissionMultiplierParameters,
) -> str:
    payload = {
        "release": parameters.snapshot(),
        "file_sha256": dict(parameters.release.file_sha256),
        "values": parameter_values(parameters),
    }
    encoded = json.dumps(
        payload,
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":"),
    ).encode("utf-8")
    return hashlib.sha256(encoded).hexdigest()


def parameters_from_database(
    release_row: Mapping[str, object],
    value_rows: list[Mapping[str, object]],
) -> FissionMultiplierParameters:
    """Build runtime parameters from one published database version."""

    metadata = json.loads(str(release_row.get("metadata_json") or "{}"))
    file_sha256 = metadata.get("file_sha256") or {}
    registered_release = RELEASES.get(str(release_row["version"]))
    release = ParameterRelease(
        version=str(release_row["version"]),
        cohort_date=str(release_row["cohort_date"]).replace("-", ""),
        observation_end_date=str(release_row["observation_end_date"]).replace(
            "-", ""
        ),
        run_suffix=str(release_row["run_suffix"]),
        file_sha256=file_sha256,
        source_version=str(
            metadata.get("source_version")
            or (registered_release.source_version if registered_release else release_row["version"])
        ),
        includes_total_to_first=bool(
            metadata.get("includes_total_to_first")
            or (registered_release.includes_total_to_first if registered_release else False)
        ),
    )
    indexed: dict[tuple[str, str, str, str], float] = {}
    total_to_first: dict[tuple[str, str, str, str], float] = {}
    for row in value_rows:
        key = (
            str(row["entity_type"]),
            str(row["match_level"]),
            str(row.get("key_primary") or ""),
            str(row.get("key_secondary") or ""),
        )
        if key in indexed:
            raise ValueError(f"数据库传播裂变系数存在重复匹配键: {key}")
        multiplier = float(row["multiplier"])
        if not math.isfinite(multiplier) or multiplier < 1:
            raise ValueError(f"数据库传播裂变系数无效: {key}={multiplier}")
        indexed[key] = multiplier
        raw_vs_first = row.get("multiplier_vs_first")
        if raw_vs_first is not None:
            multiplier_vs_first = float(raw_vs_first)
            if not math.isfinite(multiplier_vs_first) or multiplier_vs_first < 0:
                raise ValueError(
                    f"数据库总裂变UV/首层UV系数无效: {key}={multiplier_vs_first}"
                )
            total_to_first[key] = multiplier_vs_first
    if release.includes_total_to_first and set(total_to_first) != set(indexed):
        raise ValueError("数据库总裂变UV/首层UV参数与T0参数匹配键不一致")

    unsupported_qiwei_levels = sorted(
        {
            match_level
            for entity_type, match_level, _, _ in indexed
            if entity_type == "qiwei" and match_level != MATCH_QIWEI_PARTNER
        }
    )
    if unsupported_qiwei_levels:
        raise ValueError(
            "数据库企微传播裂变参数只支持合作方精确匹配: "
            + ", ".join(unsupported_qiwei_levels)
        )

    def values(entity_type: str, level: str) -> dict[tuple[str, str], float]:
        return {
            (primary, secondary): multiplier
            for (entity, match_level, primary, secondary), multiplier in indexed.items()
            if entity == entity_type and match_level == level
        }

    def single(entity_type: str, level: str) -> float:
        matched = values(entity_type, level)
        if set(matched) != {("", "")}:
            raise ValueError(f"数据库{level}必须且只能有一条渠道参数")
        return matched[("", "")]

    mini_exact = values("self", MATCH_MINIAPP_EXACT)
    mini_package = {
        key[0]: multiplier
        for key, multiplier in values("self", MATCH_MINIAPP_PACKAGE).items()
    }
    mini_goal = {
        key[0]: multiplier
        for key, multiplier in values("self", MATCH_MINIAPP_GOAL).items()
    }
    gzh_exact = values("gzh", MATCH_GZH_EXACT)
    gzh_partner = {
        key[0]: multiplier
        for key, multiplier in values("gzh", MATCH_GZH_PARTNER).items()
    }
    qiwei_partner_values = values("qiwei", MATCH_QIWEI_PARTNER)
    if any(secondary for _, secondary in qiwei_partner_values):
        raise ValueError("数据库企微合作方参数不能包含次级匹配键")
    if any(not primary for primary, _ in qiwei_partner_values):
        raise ValueError("数据库企微合作方参数不能包含空合作方键")
    qiwei_partner = {
        key[0]: multiplier for key, multiplier in qiwei_partner_values.items()
    }
    qiwei_metadata = {"qiwei_exact_rows", "qiwei_exact_available_rows"}
    if qiwei_partner and not qiwei_metadata.issubset(metadata):
        missing = sorted(qiwei_metadata - set(metadata))
        raise ValueError(f"数据库企微传播裂变参数元数据缺失: {', '.join(missing)}")
    result = FissionMultiplierParameters(
        release=release,
        miniapp_exact=mini_exact,
        miniapp_by_package=mini_package,
        miniapp_by_goal=mini_goal,
        miniapp_channel=single("self", MATCH_MINIAPP_CHANNEL),
        gzh_exact=gzh_exact,
        gzh_by_partner=gzh_partner,
        gzh_channel=single("gzh", MATCH_GZH_CHANNEL),
        total_to_first=total_to_first,
        qiwei_by_partner=qiwei_partner,
        miniapp_exact_rows=int(metadata["miniapp_exact_rows"]),
        miniapp_exact_available_rows=int(metadata["miniapp_exact_available_rows"]),
        gzh_exact_rows=int(metadata["gzh_exact_rows"]),
        gzh_exact_available_rows=int(metadata["gzh_exact_available_rows"]),
        qiwei_exact_rows=(
            int(metadata["qiwei_exact_rows"]) if qiwei_partner else 0
        ),
        qiwei_exact_available_rows=(
            int(metadata["qiwei_exact_available_rows"]) if qiwei_partner else 0
        ),
    )
    expected_hash = str(release_row["content_sha256"])
    if parameter_content_sha256(result) != expected_hash:
        raise ValueError(
            f"数据库传播裂变系数内容校验失败: version={release.version}"
        )
    return result


def normalize_name(value: object) -> str:
    if value is None or pd.isna(value):
        return ""
    text = str(value).strip()
    return "" if not text or text.lower() == "null" else text


def normalize_package(value: object) -> str:
    return normalize_name(value) or "泛人群"


def _sha256(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def _read_parameter_file(
    path: Path,
    *,
    required: set[str],
    label: str,
    expected_sha256: str,
) -> pd.DataFrame:
    if not path.is_file():
        raise FileNotFoundError(f"{label}参数文件不存在: {path}")
    actual_sha256 = _sha256(path)
    if actual_sha256 != expected_sha256:
        raise ValueError(
            f"{label}参数文件校验失败: expected={expected_sha256} actual={actual_sha256}"
        )
    frame = pd.read_csv(path, encoding="utf-8-sig")
    missing = sorted(required - set(frame.columns))
    if missing:
        raise ValueError(f"{label}参数缺少字段: {', '.join(missing)}")
    return frame


def _validated_multipliers(
    frame: pd.DataFrame,
    *,
    label: str,
) -> pd.DataFrame:
    result = frame.copy()
    numeric_columns = [
        "首层UV",
        "A0活跃裂变UV",
        "A0-A15累计裂变活跃人次",
        "T1-T15相对T0尾部系数",
        SOURCE_MULTIPLIER_COLUMN,
        SOURCE_TOTAL_TO_FIRST_COLUMN,
    ]
    for column in numeric_columns:
        if column in result:
            result[column] = pd.to_numeric(result[column], errors="coerce")

    checkable = result[
        result["A0活跃裂变UV"].gt(0)
        & result[SOURCE_MULTIPLIER_COLUMN].notna()
    ]
    if checkable.empty:
        raise ValueError(f"{label}参数没有可校验的传播裂变系数")
    expected = (
        checkable["A0-A15累计裂变活跃人次"]
        / checkable["A0活跃裂变UV"]
    )
    errors = (expected - checkable[SOURCE_MULTIPLIER_COLUMN]).abs()
    if bool(errors.gt(1e-9).any()):
        raise ValueError(f"{label}参数不满足累计活跃人次/A0恒等式")
    tails = checkable["T1-T15相对T0尾部系数"]
    tail_errors = (
        tails - (checkable[SOURCE_MULTIPLIER_COLUMN] - 1)
    ).abs()
    if bool(tail_errors.gt(1e-9).any()):
        raise ValueError(f"{label}参数不满足尾部系数=传播裂变系数-1")

    expected_vs_first = (
        result["A0-A15累计裂变活跃人次"] / result["首层UV"]
    ).where(result["首层UV"].gt(0))
    if SOURCE_TOTAL_TO_FIRST_COLUMN in frame:
        checkable_vs_first = result["首层UV"].gt(0) & result[
            SOURCE_TOTAL_TO_FIRST_COLUMN
        ].notna()
        errors_vs_first = (
            result.loc[checkable_vs_first, SOURCE_TOTAL_TO_FIRST_COLUMN]
            - expected_vs_first[checkable_vs_first]
        ).abs()
        if bool(errors_vs_first.gt(1e-9).any()):
            raise ValueError(f"{label}参数不满足累计活跃人次/首层UV恒等式")
    result[SOURCE_TOTAL_TO_FIRST_COLUMN] = expected_vs_first

    available = result[result["样本状态"].eq("可用")].copy()
    valid = available[SOURCE_MULTIPLIER_COLUMN].map(
        lambda value: bool(pd.notna(value) and math.isfinite(float(value)) and value >= 1)
    )
    if not bool(valid.all()):
        raise ValueError(f"{label}存在无效的可用传播裂变系数")
    return result


def _mapping(
    frame: pd.DataFrame,
    key_columns: list[str],
    *,
    normalizers: list,
    label: str,
    value_column: str = SOURCE_MULTIPLIER_COLUMN,
) -> dict:
    available = frame[frame["样本状态"].eq("可用")].copy()
    keys = [
        tuple(normalizer(row[column]) for column, normalizer in zip(key_columns, normalizers))
        for _, row in available.iterrows()
    ]
    if any(not all(key) for key in keys):
        raise ValueError(f"{label}存在空匹配键")
    if len(keys) != len(set(keys)):
        raise ValueError(f"{label}存在重复匹配键")
    values = available[value_column].astype(float).tolist()
    if len(key_columns) == 1:
        return {key[0]: value for key, value in zip(keys, values)}
    return dict(zip(keys, values))


def _single_channel_multiplier(
    frame: pd.DataFrame,
    level: str,
    value_column: str = SOURCE_MULTIPLIER_COLUMN,
) -> float:
    rows = frame[
        frame["参数层级"].eq(level) & frame["样本状态"].eq("可用")
    ]
    if len(rows) != 1:
        raise ValueError(f"{level}必须且只能有一条可用参数，实际为{len(rows)}")
    return float(rows.iloc[0][value_column])


def load_fission_multiplier_parameters(
    version: str = DEFAULT_FISSION_PARAMETER_VERSION,
    *,
    parameter_root: Path | None = None,
) -> FissionMultiplierParameters:
    try:
        release = RELEASES[version]
    except KeyError as exc:
        raise ValueError(f"未知传播裂变系数参数版本: {version}") from exc
    directory = (parameter_root or PARAMETER_ROOT) / release.source_version

    exact_required = {
        "首层投放日期",
        "首层UV",
        "A0活跃裂变UV",
        "A0-A15累计裂变活跃人次",
        "T1-T15相对T0尾部系数",
        SOURCE_MULTIPLIER_COLUMN,
        "样本状态",
    }
    if release.includes_total_to_first:
        exact_required.add(SOURCE_TOTAL_TO_FIRST_COLUMN)
    miniapp = _read_parameter_file(
        directory / MINIAPP_FILE,
        required=exact_required | {"人群包", "转化目标"},
        label="小程序",
        expected_sha256=release.file_sha256[MINIAPP_FILE],
    )
    gzh = _read_parameter_file(
        directory / GZH_FILE,
        required=exact_required | {"合作方", "公众号"},
        label="公众号",
        expected_sha256=release.file_sha256[GZH_FILE],
    )
    has_qiwei_file = QIWEI_FILE in release.file_sha256
    qiwei = None
    if has_qiwei_file:
        qiwei = _read_parameter_file(
            directory / QIWEI_FILE,
            required=exact_required | {"合作方"},
            label="企微",
            expected_sha256=release.file_sha256[QIWEI_FILE],
        )
    fallback = _read_parameter_file(
        directory / FALLBACK_FILE,
        required={
            "参数层级",
            "人群包",
            "转化目标",
            "合作方",
            "公众号",
            "首层UV",
            "A0活跃裂变UV",
            "A0-A15累计裂变活跃人次",
            "T1-T15相对T0尾部系数",
            SOURCE_MULTIPLIER_COLUMN,
            "样本状态",
        },
        label="分级回退",
        expected_sha256=release.file_sha256[FALLBACK_FILE],
    )
    miniapp = _validated_multipliers(miniapp, label="小程序")
    gzh = _validated_multipliers(gzh, label="公众号")
    if qiwei is not None:
        qiwei = _validated_multipliers(qiwei, label="企微")
    fallback = _validated_multipliers(fallback, label="分级回退")

    exact_frames = [("小程序", miniapp), ("公众号", gzh)]
    if qiwei is not None:
        exact_frames.append(("企微", qiwei))
    for label, frame in exact_frames:
        dates = {str(value) for value in frame["首层投放日期"].dropna()}
        if dates != {release.cohort_date}:
            raise ValueError(
                f"{label}参数cohort日期不一致: expected={release.cohort_date} actual={sorted(dates)}"
            )

    mini_package_rows = fallback[fallback["参数层级"].eq("小程序人群包回退")]
    mini_goal_rows = fallback[fallback["参数层级"].eq("小程序转化目标回退")]
    gzh_partner_rows = fallback[fallback["参数层级"].eq("公众号合作方回退")]
    qiwei_by_partner = (
        _mapping(
            qiwei,
            ["合作方"],
            normalizers=[normalize_name],
            label="企微合作方参数",
        )
        if qiwei is not None
        else {}
    )

    miniapp_exact = _mapping(
        miniapp,
        ["人群包", "转化目标"],
        normalizers=[normalize_package, normalize_name],
        label="小程序精确参数",
    )
    miniapp_by_package = _mapping(
        mini_package_rows,
        ["人群包"],
        normalizers=[normalize_package],
        label="小程序人群包回退参数",
    )
    miniapp_by_goal = _mapping(
        mini_goal_rows,
        ["转化目标"],
        normalizers=[normalize_name],
        label="小程序转化目标回退参数",
    )
    gzh_exact = _mapping(
        gzh,
        ["合作方", "公众号"],
        normalizers=[normalize_name, normalize_name],
        label="公众号精确参数",
    )
    gzh_by_partner = _mapping(
        gzh_partner_rows,
        ["合作方"],
        normalizers=[normalize_name],
        label="公众号合作方回退参数",
    )

    total_to_first: dict[tuple[str, str, str, str], float] = {}

    def add_total_to_first(
        entity_type: str,
        match_level: str,
        mapping: Mapping,
    ) -> None:
        for raw_key, value in mapping.items():
            keys = raw_key if isinstance(raw_key, tuple) else (raw_key, "")
            total_to_first[(entity_type, match_level, keys[0], keys[1])] = value

    if release.includes_total_to_first:
        add_total_to_first(
            "self",
            MATCH_MINIAPP_EXACT,
            _mapping(
                miniapp,
                ["人群包", "转化目标"],
                normalizers=[normalize_package, normalize_name],
                label="小程序精确参数",
                value_column=SOURCE_TOTAL_TO_FIRST_COLUMN,
            ),
        )
        add_total_to_first(
            "self",
            MATCH_MINIAPP_PACKAGE,
            _mapping(
                mini_package_rows,
                ["人群包"],
                normalizers=[normalize_package],
                label="小程序人群包回退参数",
                value_column=SOURCE_TOTAL_TO_FIRST_COLUMN,
            ),
        )
        add_total_to_first(
            "self",
            MATCH_MINIAPP_GOAL,
            _mapping(
                mini_goal_rows,
                ["转化目标"],
                normalizers=[normalize_name],
                label="小程序转化目标回退参数",
                value_column=SOURCE_TOTAL_TO_FIRST_COLUMN,
            ),
        )
        total_to_first[("self", MATCH_MINIAPP_CHANNEL, "", "")] = (
            _single_channel_multiplier(
                fallback,
                "小程序渠道回退",
                SOURCE_TOTAL_TO_FIRST_COLUMN,
            )
        )
        add_total_to_first(
            "gzh",
            MATCH_GZH_EXACT,
            _mapping(
                gzh,
                ["合作方", "公众号"],
                normalizers=[normalize_name, normalize_name],
                label="公众号精确参数",
                value_column=SOURCE_TOTAL_TO_FIRST_COLUMN,
            ),
        )
        add_total_to_first(
            "gzh",
            MATCH_GZH_PARTNER,
            _mapping(
                gzh_partner_rows,
                ["合作方"],
                normalizers=[normalize_name],
                label="公众号合作方回退参数",
                value_column=SOURCE_TOTAL_TO_FIRST_COLUMN,
            ),
        )
        total_to_first[("gzh", MATCH_GZH_CHANNEL, "", "")] = (
            _single_channel_multiplier(
                fallback,
                "公众号渠道回退",
                SOURCE_TOTAL_TO_FIRST_COLUMN,
            )
        )
        if qiwei is not None:
            add_total_to_first(
                "qiwei",
                MATCH_QIWEI_PARTNER,
                _mapping(
                    qiwei,
                    ["合作方"],
                    normalizers=[normalize_name],
                    label="企微合作方参数",
                    value_column=SOURCE_TOTAL_TO_FIRST_COLUMN,
                ),
            )

    return FissionMultiplierParameters(
        release=release,
        miniapp_exact=miniapp_exact,
        miniapp_by_package=miniapp_by_package,
        miniapp_by_goal=miniapp_by_goal,
        miniapp_channel=_single_channel_multiplier(
            fallback, "小程序渠道回退"
        ),
        gzh_exact=gzh_exact,
        gzh_by_partner=gzh_by_partner,
        gzh_channel=_single_channel_multiplier(
            fallback, "公众号渠道回退"
        ),
        total_to_first=total_to_first,
        qiwei_by_partner=qiwei_by_partner,
        miniapp_exact_rows=len(miniapp),
        miniapp_exact_available_rows=int(miniapp["样本状态"].eq("可用").sum()),
        gzh_exact_rows=len(gzh),
        gzh_exact_available_rows=int(gzh["样本状态"].eq("可用").sum()),
        qiwei_exact_rows=len(qiwei) if qiwei is not None else 0,
        qiwei_exact_available_rows=(
            int(qiwei["样本状态"].eq("可用").sum())
            if qiwei is not None
            else 0
        ),
    )


def apply_fission_multiplier(
    daily: pd.DataFrame,
    parameters: FissionMultiplierParameters,
) -> pd.DataFrame:
    required = {"entity_type", "包名", "广告优化目标", "合作方名", "公众号名"}
    missing = sorted(required - set(daily.columns))
    if missing:
        raise ValueError(f"传播裂变系数匹配缺少字段: {', '.join(missing)}")

    result = daily.copy()
    multipliers: list[float] = []
    multipliers_vs_first: list[float | None] = []
    levels: list[str] = []
    sources: list[str] = []
    for row in result.to_dict("records"):
        entity_type = str(row["entity_type"])
        match = parameters.match(
            "self" if entity_type == "self_ad" else entity_type,
            package=row["包名"],
            optimize_goal=row["广告优化目标"],
            partner=row["合作方名"],
            official_account=row["公众号名"],
        )
        multipliers.append(match.multiplier)
        multipliers_vs_first.append(match.multiplier_vs_first)
        levels.append(match.match_level)
        sources.append(match.source)

    result[DISPLAY_MULTIPLIER_COLUMN] = multipliers
    result[DISPLAY_TOTAL_TO_FIRST_COLUMN] = multipliers_vs_first
    result["传播裂变系数匹配层级"] = levels
    result["传播裂变系数来源"] = sources
    result["传播裂变参数版本"] = parameters.release.version
    result["传播裂变参数cohort日期"] = parameters.release.cohort_date
    qiwei = result["entity_type"].eq("qiwei")
    qiwei_reference = qiwei & result["传播裂变系数匹配层级"].eq(
        MATCH_QIWEI_REFERENCE
    )
    qiwei_formal = qiwei & ~qiwei_reference
    result["传播裂变参数状态"] = "正式参数"
    result["调控参与状态"] = "参与阈值和调控"
    result.loc[result["entity_type"].eq("self_ad"), "调控参与状态"] = (
        "广告级复用统一阈值_审批后可暂停广告"
    )
    result.loc[qiwei, "调控参与状态"] = "仅展示_不进入阈值和调控"
    result["ROI计算口径"] = (
        "首层实际效率收入+T0实际裂变收入+预测T1-T15裂变收入"
    )
    result.loc[qiwei_formal, "传播裂变参数状态"] = "企微正式参数_仅展示"
    result.loc[qiwei_reference, "传播裂变参数版本"] = QIWEI_REFERENCE_VERSION
    result.loc[qiwei_reference, "传播裂变参数cohort日期"] = ""
    result.loc[qiwei_reference, "传播裂变参数状态"] = (
        f"默认参考系数{QIWEI_REFERENCE_MULTIPLIER:g}_待发布合作方参数"
    )
    result.loc[qiwei_reference, "ROI计算口径"] = (
        "首层实际效率收入+T0实际裂变收入+预测T1-T15裂变收入_"
        f"传播裂变系数默认取{QIWEI_REFERENCE_MULTIPLIER:g}"
    )
    return result
