"""Offline T15 cohort parameter recomputation.

The functions here are read-only DataFrame/SQL transformations. Generated
parameters are drafts; publishing requires adding a reviewed release to
``fission_multiplier.RELEASES`` with file checksums.
"""

from __future__ import annotations

from datetime import datetime, timedelta
from pathlib import Path

import pandas as pd
from openpyxl.styles import Font


HORIZON = 15


def _parse_date(value: str) -> datetime:
    try:
        return datetime.strptime(value, "%Y%m%d")
    except ValueError as exc:
        raise ValueError(f"日期必须是YYYYMMDD，实际为: {value}") from exc


def build_miniapp_activity_curve_sql(
    cohort_date: str,
    horizon: int = HORIZON,
) -> str:
    """Build the audited miniapp package + conversion-goal cohort SQL."""

    start = _parse_date(cohort_date)
    if horizon < 0:
        raise ValueError("horizon不能小于0")
    dates = [
        (start + timedelta(days=lag)).strftime("%Y%m%d")
        for lag in range(horizon + 1)
    ]
    end_date = dates[-1]
    active_counts = ",\n".join(
        f"    SUM(CASE WHEN active_dt='{date}' THEN 1 ELSE 0 END) AS a{lag}_active_uv"
        for lag, date in enumerate(dates)
    )
    active_select = ",\n".join(
        f"  NVL(a.a{lag}_active_uv,0) AS a{lag}_active_uv"
        for lag in range(horizon + 1)
    )
    rate_select = ",\n".join(
        f"  CASE WHEN b.first_uv>0 THEN NVL(a.a{lag}_active_uv,0)/b.first_uv END AS a{lag}_active_rate"
        for lag in range(horizon + 1)
    )
    active_sum = "+".join(
        f"NVL(a.a{lag}_active_uv,0)" for lag in range(horizon + 1)
    )
    return f"""
WITH first_cohort AS (
  SELECT DISTINCT
    CASE WHEN 包名 IS NULL OR 包名='' OR 包名='null' THEN '泛人群' ELSE 包名 END AS package_name,
    CASE WHEN 广告优化目标 IS NULL OR 广告优化目标='' OR 广告优化目标='null' THEN '未知' ELSE 广告优化目标 END AS optimize_goal,
    mid AS first_mid,
    subsessionid AS rootsessionid
  FROM loghubods.opengid_base_data
  WHERE dt='{cohort_date}'
    AND channel='小程序投流-稳定'
    AND usersharedepth=0
    AND hotsencetype != 1167
    AND videoid IS NOT NULL
    AND mid IS NOT NULL
    AND subsessionid IS NOT NULL
),
root_entity AS (
  SELECT DISTINCT package_name, optimize_goal, rootsessionid FROM first_cohort
),
first_users AS (
  SELECT DISTINCT package_name, optimize_goal, first_mid FROM first_cohort
),
base_stats AS (
  SELECT package_name, optimize_goal,
         COUNT(DISTINCT first_mid) AS first_uv,
         COUNT(DISTINCT rootsessionid) AS root_session_cnt
  FROM first_cohort
  GROUP BY package_name, optimize_goal
),
raw_active_fission AS (
  SELECT DISTINCT
    r.package_name,
    r.optimize_goal,
    s.machinecode AS active_mid,
    s.dt AS active_dt
  FROM root_entity r
  JOIN loghubods.user_share_log s
    ON s.rootsessionid=r.rootsessionid
  WHERE s.dt BETWEEN '{cohort_date}' AND '{end_date}'
    AND s.topic='click'
    AND s.machinecode IS NOT NULL
    AND CAST(s.usersharedepth AS BIGINT)>0
),
active_fission AS (
  SELECT c.*
  FROM raw_active_fission c
  LEFT JOIN first_users f
    ON c.package_name=f.package_name AND c.optimize_goal=f.optimize_goal AND c.active_mid=f.first_mid
  WHERE f.first_mid IS NULL
),
active_counts AS (
  SELECT package_name, optimize_goal,
{active_counts},
    COUNT(*) AS active_person_days
  FROM active_fission
  GROUP BY package_name, optimize_goal
)
SELECT
  '{cohort_date}' AS cohort_dt,
  b.package_name,
  b.optimize_goal,
  b.first_uv,
  b.root_session_cnt,
{active_select},
{rate_select},
  NVL(a.active_person_days,0) AS active_person_days,
  CASE WHEN b.first_uv>0 THEN ({active_sum})/b.first_uv END AS activity_rate_sum,
  CASE WHEN NVL(a.a0_active_uv,0)>0
       THEN (({active_sum})-NVL(a.a0_active_uv,0))/a.a0_active_uv
  END AS active_tail_vs_a0,
  CASE WHEN NVL(a.a0_active_uv,0)>0
       THEN ({active_sum})/a.a0_active_uv
  END AS activity_maturity_vs_a0,
  CASE WHEN b.first_uv>0 THEN 1+({active_sum})/b.first_uv END AS steady_dau_multiplier
FROM base_stats b
LEFT JOIN active_counts a ON b.package_name=a.package_name AND b.optimize_goal=a.optimize_goal
ORDER BY b.first_uv DESC
""".strip()


def build_gzh_activity_curve_sql(
    cohort_date: str,
    horizon: int = HORIZON,
    min_first_uv: int = 500,
) -> str:
    """Build the audited official-account partner + account cohort SQL."""

    start = _parse_date(cohort_date)
    if horizon < 0:
        raise ValueError("horizon不能小于0")
    if min_first_uv < 0:
        raise ValueError("首层UV门槛不能小于0")
    dates = [
        (start + timedelta(days=lag)).strftime("%Y%m%d")
        for lag in range(horizon + 1)
    ]
    end_date = dates[-1]
    active_counts = ",\n".join(
        f"    SUM(CASE WHEN active_dt='{date}' THEN 1 ELSE 0 END) AS a{lag}_active_uv"
        for lag, date in enumerate(dates)
    )
    active_select = ",\n".join(
        f"  NVL(a.a{lag}_active_uv,0) AS a{lag}_active_uv"
        for lag in range(horizon + 1)
    )
    rate_select = ",\n".join(
        f"  CASE WHEN b.first_uv>0 THEN NVL(a.a{lag}_active_uv,0)/b.first_uv END AS a{lag}_active_rate"
        for lag in range(horizon + 1)
    )
    active_sum = "+".join(
        f"NVL(a.a{lag}_active_uv,0)" for lag in range(horizon + 1)
    )
    return f"""
WITH first_cohort AS (
  SELECT DISTINCT
    CASE WHEN 合作方名 IS NULL OR 合作方名='' THEN 'null' ELSE 合作方名 END AS partner_name,
    CASE WHEN 公众号名 IS NULL OR 公众号名='' THEN 'null' ELSE 公众号名 END AS account_name,
    mid AS first_mid,
    subsessionid AS rootsessionid
  FROM loghubods.opengid_base_data
  WHERE dt='{cohort_date}'
    AND channel='公众号合作-即转-稳定'
    AND usersharedepth=0
    AND hotsencetype != 1167
    AND videoid IS NOT NULL
    AND mid IS NOT NULL
    AND subsessionid IS NOT NULL
),
base_stats AS (
  SELECT partner_name, account_name,
         COUNT(DISTINCT first_mid) AS first_uv,
         COUNT(DISTINCT rootsessionid) AS root_session_cnt
  FROM first_cohort
  GROUP BY partner_name, account_name
  HAVING COUNT(DISTINCT first_mid)>{min_first_uv}
),
root_entity AS (
  SELECT DISTINCT f.partner_name, f.account_name, f.rootsessionid
  FROM first_cohort f
  JOIN base_stats b ON f.partner_name=b.partner_name AND f.account_name=b.account_name
),
first_users AS (
  SELECT DISTINCT f.partner_name, f.account_name, f.first_mid
  FROM first_cohort f
  JOIN base_stats b ON f.partner_name=b.partner_name AND f.account_name=b.account_name
),
raw_active_fission AS (
  SELECT DISTINCT r.partner_name, r.account_name, s.machinecode AS active_mid, s.dt AS active_dt
  FROM root_entity r
  JOIN loghubods.user_share_log s
    ON s.rootsessionid=r.rootsessionid
  WHERE s.dt BETWEEN '{cohort_date}' AND '{end_date}'
    AND s.topic='click'
    AND s.machinecode IS NOT NULL
    AND CAST(s.usersharedepth AS BIGINT)>0
),
active_fission AS (
  SELECT c.*
  FROM raw_active_fission c
  LEFT JOIN first_users f
    ON c.partner_name=f.partner_name AND c.account_name=f.account_name AND c.active_mid=f.first_mid
  WHERE f.first_mid IS NULL
),
active_counts AS (
  SELECT partner_name, account_name,
{active_counts},
    COUNT(*) AS active_person_days
  FROM active_fission
  GROUP BY partner_name, account_name
)
SELECT
  '{cohort_date}' AS cohort_dt,
  '公众号合作-即转-稳定' AS channel,
  b.partner_name,
  b.account_name,
  b.first_uv,
  b.root_session_cnt,
{active_select},
{rate_select},
  NVL(a.active_person_days,0) AS active_person_days,
  CASE WHEN b.first_uv>0 THEN ({active_sum})/b.first_uv END AS activity_rate_sum,
  CASE WHEN NVL(a.a0_active_uv,0)>0
       THEN (({active_sum})-NVL(a.a0_active_uv,0))/a.a0_active_uv
  END AS active_tail_vs_a0,
  CASE WHEN NVL(a.a0_active_uv,0)>0
       THEN ({active_sum})/a.a0_active_uv
  END AS activity_maturity_vs_a0,
  CASE WHEN b.first_uv>0 THEN 1+({active_sum})/b.first_uv END AS steady_dau_multiplier
FROM base_stats b
LEFT JOIN active_counts a ON b.partner_name=a.partner_name AND b.account_name=a.account_name
ORDER BY b.first_uv DESC
""".strip()


def build_qiwei_activity_curve_sql(
    cohort_date: str,
    horizon: int = HORIZON,
    min_first_uv: int = 500,
) -> str:
    """Build the audited WeCom partner-level cohort SQL."""

    start = _parse_date(cohort_date)
    if horizon < 0:
        raise ValueError("horizon不能小于0")
    if min_first_uv < 0:
        raise ValueError("首层UV门槛不能小于0")
    dates = [
        (start + timedelta(days=lag)).strftime("%Y%m%d")
        for lag in range(horizon + 1)
    ]
    end_date = dates[-1]
    active_counts = ",\n".join(
        f"    SUM(CASE WHEN active_dt='{date}' THEN 1 ELSE 0 END) AS a{lag}_active_uv"
        for lag, date in enumerate(dates)
    )
    active_select = ",\n".join(
        f"  NVL(a.a{lag}_active_uv,0) AS a{lag}_active_uv"
        for lag in range(horizon + 1)
    )
    rate_select = ",\n".join(
        f"  CASE WHEN b.first_uv>0 THEN NVL(a.a{lag}_active_uv,0)/b.first_uv END AS a{lag}_active_rate"
        for lag in range(horizon + 1)
    )
    active_sum = "+".join(
        f"NVL(a.a{lag}_active_uv,0)" for lag in range(horizon + 1)
    )
    active_tail_sum = "+".join(
        f"NVL(a.a{lag}_active_uv,0)" for lag in range(1, horizon + 1)
    ) or "0"
    return f"""
WITH first_cohort AS (
  SELECT DISTINCT
    CASE WHEN d.合作方名 IS NULL OR d.合作方名='' THEN 'null' ELSE d.合作方名 END AS partner_name,
    d.mid AS first_mid,
    l.rootsessionid AS rootsessionid
  FROM loghubods.opengid_base_data d
  JOIN loghubods.user_share_log l
    ON d.subsessionid=l.subsessionid AND d.mid=l.machinecode
  WHERE d.dt='{cohort_date}' AND l.dt='{cohort_date}'
    AND d.channel='群/企微合作-稳定'
    AND d.usersharedepth=0
    AND d.hotsencetype != 1167
    AND d.videoid IS NOT NULL
    AND d.mid IS NOT NULL
    AND d.subsessionid IS NOT NULL
),
base_stats AS (
  SELECT partner_name,
         COUNT(DISTINCT first_mid) AS first_uv,
         COUNT(DISTINCT rootsessionid) AS root_session_cnt
  FROM first_cohort
  GROUP BY partner_name
  HAVING COUNT(DISTINCT first_mid)>{min_first_uv}
),
root_entity AS (
  SELECT DISTINCT f.partner_name, f.rootsessionid
  FROM first_cohort f
  JOIN base_stats b ON f.partner_name=b.partner_name
),
first_users AS (
  SELECT DISTINCT f.partner_name, f.first_mid
  FROM first_cohort f
  JOIN base_stats b ON f.partner_name=b.partner_name
),
raw_active_fission AS (
  SELECT DISTINCT r.partner_name, s.machinecode AS active_mid, s.dt AS active_dt
  FROM root_entity r
  JOIN (
    SELECT DISTINCT a.dt, a.machinecode, a.rootsessionid
    FROM loghubods.user_share_log a
    LEFT JOIN (
      SELECT DISTINCT
        apptype,
        machinecode,
        rootsessionid,
        channel_type,
        user_share_depth,
        real_user_share_depth,
        subsessionid
      FROM loghubods.ods_user_active_log_info_day
      WHERE dt BETWEEN '{cohort_date}' AND '{end_date}'
        AND user_share_depth=0
    ) b
      ON a.machinecode=b.machinecode AND a.subsessionid=b.subsessionid
    WHERE a.dt BETWEEN '{cohort_date}' AND '{end_date}'
      AND a.topic='click'
      AND b.machinecode IS NULL
      AND CAST(a.usersharedepth AS BIGINT)>0
  ) s
    ON s.rootsessionid=r.rootsessionid
  WHERE s.dt BETWEEN '{cohort_date}' AND '{end_date}'
    AND s.machinecode IS NOT NULL
),
active_fission AS (
  SELECT c.*
  FROM raw_active_fission c
  LEFT JOIN first_users f
    ON c.partner_name=f.partner_name AND c.active_mid=f.first_mid
  WHERE f.first_mid IS NULL
),
active_counts AS (
  SELECT partner_name,
{active_counts},
    COUNT(*) AS active_person_days
  FROM active_fission
  GROUP BY partner_name
)
SELECT
  '{cohort_date}' AS cohort_dt,
  '群/企微合作-稳定' AS channel,
  b.partner_name,
  b.first_uv,
  b.root_session_cnt,
{active_select},
{rate_select},
  NVL(a.active_person_days,0) AS active_person_days,
  CASE WHEN b.first_uv>0 THEN ({active_sum})/b.first_uv END AS activity_rate_sum,
  CASE WHEN NVL(a.a0_active_uv,0)>0
       THEN ({active_tail_sum})/a.a0_active_uv
  END AS active_tail_vs_a0,
  CASE WHEN NVL(a.a0_active_uv,0)>0
       THEN ({active_sum})/a.a0_active_uv
  END AS activity_maturity_vs_a0,
  CASE WHEN b.first_uv>0 THEN 1+({active_sum})/b.first_uv END AS steady_dau_multiplier
FROM base_stats b
LEFT JOIN active_counts a ON b.partner_name=a.partner_name
ORDER BY b.first_uv DESC
""".strip()


def _sample_status(first_uv: pd.Series, a0_uv: pd.Series) -> pd.Series:
    status = pd.Series("可用", index=first_uv.index, dtype="object")
    status.loc[first_uv < 500] = "首层UV不足500"
    status.loc[(first_uv >= 500) & (a0_uv < 200)] = "A0裂变UV不足200"
    return status


def localize_exact(raw: pd.DataFrame, entity: str) -> pd.DataFrame:
    result = raw.copy()
    if "optimize_goal" in result:
        result["optimize_goal"] = (
            result["optimize_goal"]
            .fillna("未知")
            .replace({"": "未知", "null": "未知"})
        )
    for column in ["first_uv", "root_session_cnt", "active_person_days"]:
        result[column] = pd.to_numeric(result[column], errors="coerce").fillna(0)
    for lag in range(HORIZON + 1):
        for suffix in ("active_uv", "active_rate"):
            column = f"a{lag}_{suffix}"
            result[column] = pd.to_numeric(result[column], errors="coerce").fillna(0)
    for column in (
        "activity_rate_sum",
        "active_tail_vs_a0",
        "activity_maturity_vs_a0",
        "steady_dau_multiplier",
    ):
        result[column] = pd.to_numeric(result[column], errors="coerce")

    mapping = {
        "cohort_dt": "首层投放日期",
        "channel": "渠道",
        "package_name": "人群包",
        "optimize_goal": "转化目标",
        "partner_name": "合作方",
        "account_name": "公众号",
        "first_uv": "首层UV",
        "root_session_cnt": "根会话数",
        "active_person_days": "A0-A15累计裂变活跃人次",
        "activity_rate_sum": "每首层累计裂变活跃人次",
        "active_tail_vs_a0": "T1-T15相对T0尾部系数",
        "activity_maturity_vs_a0": "T15成熟系数（相对T0）",
        "steady_dau_multiplier": "稳态DAU倍数",
    }
    for lag in range(HORIZON + 1):
        mapping[f"a{lag}_active_uv"] = f"A{lag}活跃裂变UV"
        mapping[f"a{lag}_active_rate"] = f"A{lag}活跃率"
    result["样本状态"] = _sample_status(result["first_uv"], result["a0_active_uv"])
    result["参数层级"] = entity
    mapping["样本状态"] = "样本状态"
    mapping["参数层级"] = "参数层级"
    available = {key: value for key, value in mapping.items() if key in result}
    return result.rename(columns=available)[list(available.values())]


def _rollup(raw: pd.DataFrame, groups: list[str], level: str) -> pd.DataFrame:
    columns = ["first_uv", "a0_active_uv", "active_person_days"]
    work = raw.copy()
    for column in columns:
        work[column] = pd.to_numeric(work[column], errors="coerce").fillna(0)
    if groups:
        result = work.groupby(groups, as_index=False, dropna=False)[columns].sum()
    else:
        result = pd.DataFrame([work[columns].sum()])
    result["参数层级"] = level
    result["T1-T15相对T0尾部系数"] = (
        (result["active_person_days"] - result["a0_active_uv"])
        / result["a0_active_uv"].replace(0, pd.NA)
    )
    result["T15成熟系数（相对T0）"] = (
        result["active_person_days"]
        / result["a0_active_uv"].replace(0, pd.NA)
    )
    result["样本状态"] = _sample_status(result["first_uv"], result["a0_active_uv"])
    result = result.rename(
        columns={
            "package_name": "人群包",
            "optimize_goal": "转化目标",
            "partner_name": "合作方",
            "account_name": "公众号",
            "first_uv": "首层UV",
            "a0_active_uv": "A0活跃裂变UV",
            "active_person_days": "A0-A15累计裂变活跃人次",
        }
    )
    for column in ("人群包", "转化目标", "合作方", "公众号"):
        if column not in result:
            result[column] = ""
    return result[
        [
            "参数层级",
            "人群包",
            "转化目标",
            "合作方",
            "公众号",
            "首层UV",
            "A0活跃裂变UV",
            "A0-A15累计裂变活跃人次",
            "T1-T15相对T0尾部系数",
            "T15成熟系数（相对T0）",
            "样本状态",
        ]
    ]


def build_qiwei_release_frame(qiwei_raw: pd.DataFrame) -> pd.DataFrame:
    return localize_exact(qiwei_raw, "企微合作方")


def build_release_frames(
    miniapp_raw: pd.DataFrame,
    gzh_raw: pd.DataFrame,
) -> tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]:
    miniapp = localize_exact(miniapp_raw, "人群包+转化目标")
    gzh = localize_exact(gzh_raw, "合作方+公众号")
    fallback = pd.concat(
        [
            _rollup(miniapp_raw, ["package_name"], "小程序人群包回退"),
            _rollup(miniapp_raw, ["optimize_goal"], "小程序转化目标回退"),
            _rollup(miniapp_raw, [], "小程序渠道回退"),
            _rollup(gzh_raw, ["partner_name"], "公众号合作方回退"),
            _rollup(gzh_raw, [], "公众号渠道回退"),
        ],
        ignore_index=True,
    )
    return miniapp, gzh, fallback


def _format_workbook(path: Path) -> None:
    from openpyxl import load_workbook

    workbook = load_workbook(path)
    for sheet in workbook.worksheets:
        sheet.freeze_panes = "A2"
        sheet.auto_filter.ref = sheet.dimensions
        for cell in sheet[1]:
            cell.font = Font(bold=True)
        for cells in sheet.columns:
            width = min(max(len(str(cell.value or "")) for cell in cells) + 2, 32)
            sheet.column_dimensions[cells[0].column_letter].width = width
    workbook.save(path)


def _draft_notes(cohort_date: str) -> pd.DataFrame:
    return pd.DataFrame(
        {
            "项目": ["Cohort日期", "观察周期", "参数状态", "收入公式"],
            "定义": [
                cohort_date,
                "A0至A15，同一用户同日去重、隔日重新计入",
                "草稿；完成审计并登记版本与SHA256后才能被日级ROI消费",
                "首层实际效率收入+T0实际裂变收入+T0实际裂变收入×(传播裂变系数-对T0裂变-1)",
            ],
        }
    )


def write_draft_release(
    output_dir: Path,
    cohort_date: str,
    miniapp_raw: pd.DataFrame,
    gzh_raw: pd.DataFrame,
) -> dict[str, Path]:
    output_dir.mkdir(parents=True, exist_ok=True)
    miniapp, gzh, fallback = build_release_frames(miniapp_raw, gzh_raw)
    paths = {
        "miniapp": output_dir / f"小程序T15成熟系数_中文_{cohort_date}.csv",
        "gzh": output_dir / f"公众号T15成熟系数_中文_{cohort_date}.csv",
        "fallback": output_dir / f"T15分级回退系数_中文_{cohort_date}.csv",
        "workbook": output_dir / f"最终T15成熟系数_中文_{cohort_date}.xlsx",
    }
    miniapp.to_csv(paths["miniapp"], index=False, encoding="utf-8-sig")
    gzh.to_csv(paths["gzh"], index=False, encoding="utf-8-sig")
    fallback.to_csv(paths["fallback"], index=False, encoding="utf-8-sig")
    with pd.ExcelWriter(paths["workbook"], engine="openpyxl") as writer:
        miniapp.to_excel(writer, sheet_name="小程序精确系数", index=False)
        gzh.to_excel(writer, sheet_name="公众号精确系数", index=False)
        fallback.to_excel(writer, sheet_name="分级回退系数", index=False)
        _draft_notes(cohort_date).to_excel(writer, sheet_name="计算说明", index=False)
    _format_workbook(paths["workbook"])
    return paths


def write_qiwei_draft(
    output_dir: Path,
    cohort_date: str,
    qiwei_raw: pd.DataFrame,
) -> dict[str, Path]:
    exact = build_qiwei_release_frame(qiwei_raw)
    output_dir.mkdir(parents=True, exist_ok=True)
    paths = {
        "exact": output_dir / f"企微T15成熟系数_中文_{cohort_date}.csv",
        "workbook": output_dir / f"企微T15成熟系数_中文_{cohort_date}.xlsx",
    }
    exact.to_csv(paths["exact"], index=False, encoding="utf-8-sig")
    with pd.ExcelWriter(paths["workbook"], engine="openpyxl") as writer:
        exact.to_excel(writer, sheet_name="企微合作方系数", index=False)
        _draft_notes(cohort_date).to_excel(writer, sheet_name="计算说明", index=False)
    _format_workbook(paths["workbook"])
    return paths
