Agent/examples/auto_put_ad_mini/roi_control/service.py

"""Orchestrate one idempotent daily ROI metric and approval batch."""

from __future__ import annotations

import hashlib
import logging
import os
import re
from datetime import datetime, timedelta
from pathlib import Path
from typing import Any
from zoneinfo import ZoneInfo

from storage import initialize_schema, load_managed_accounts
from tencent_client import ACTIVE_STATUS, TencentClient

from .agency_delivery import agency_route_key, publish_agency_reports
from .config import AgencyWebhookConfig, RoiConfig
from .data_source import (
    ODPSClient,
    date_window,
    fetch_ad_age,
    fetch_daily_data,
    resolve_end_date,
)
from .feishu import RoiFeishuPublisher
from .metrics import ENTITY_SELF, METRIC_RUN_SUFFIX, METRIC_VERSION
from .fission_multiplier import (
    DEFAULT_FISSION_PARAMETER_VERSION,
    FissionMultiplierParameters,
    parameters_from_database,
)
from .policy import annotate_execution
from .reporting import (
    REPORT_RUN_SUFFIX,
    REPORT_VERSION,
    write_agency_workbooks,
    write_workbook,
)
from .rules import (
    POLICY_RUN_SUFFIX,
    POLICY_VERSION,
    RuleConfig,
    evaluate_rules,
)
from .repository import (
    FINAL_STATUSES,
    create_or_load_run,
    load_agency_deliveries,
    load_fission_parameter_release,
    mark_failed,
    mark_published,
    replace_run_results,
)


SHANGHAI = ZoneInfo("Asia/Shanghai")
logger = logging.getLogger("auto_put_ad_mini.roi_control")


def _redact_agency_webhooks(
    error: Exception,
    config: AgencyWebhookConfig,
) -> str:
    message = str(error)
    for webhook_url in (config.webhooks or {}).values():
        message = message.replace(webhook_url, "<redacted>")
    return message


def _internal_source_revision(config: RoiConfig, now: datetime) -> str:
    if config.internal_test_dedup_enabled:
        return "internal_test"
    return f"it_{now.strftime('%H%M%S%f')}"


def _build_internal_reports(
    *,
    agency_reports: list[dict[str, object]],
) -> list[dict[str, object]]:
    reports = [
        {
            **report,
            "title": f"内部测试_{Path(str(report['report'])).stem}",
        }
        for report in agency_reports
        if agency_route_key(str(report["agency_name"])) == "自动化投放"
    ]
    if len(reports) != 1:
        raise RuntimeError(
            "Internal ROI test requires exactly one 自动化投放 agency report"
        )
    return reports


def _internal_webhook_config(
    reports: list[dict[str, object]],
    webhook_url: str,
) -> AgencyWebhookConfig:
    return AgencyWebhookConfig(
        enabled=True,
        webhooks={
            agency_route_key(str(report["agency_name"])): webhook_url
            for report in reports
        },
    )


def _annotate_current_creative_status(
    rows,
    tencent: TencentClient | None = None,
):
    """Read current Tencent status only for creative-level stop decisions."""

    result = rows.copy()
    result["当前创意状态"] = ""
    mask = result["entity_type"].eq(ENTITY_SELF) & result["动作"].eq("关停")
    if not mask.any():
        return result

    client = tencent or TencentClient()
    cache: dict[tuple[int, int], str] = {}
    try:
        for index, row in result.loc[mask].iterrows():
            try:
                account_id = int(row["账号id"])
                creative_id = int(row["创意id"])
            except (TypeError, ValueError):
                result.at[index, "当前创意状态"] = "读取失败"
                continue
            key = (account_id, creative_id)
            if key not in cache:
                try:
                    creative = client.get_dynamic_creative(account_id, creative_id)
                    status = str(creative.get("configured_status") or "")
                    cache[key] = (
                        "正常"
                        if status == ACTIVE_STATUS
                        else "已停止" if status else "读取失败"
                    )
                except Exception as exc:
                    logger.warning(
                        "Failed to read creative status account=%s creative=%s: %s",
                        account_id,
                        creative_id,
                        exc,
                    )
                    cache[key] = "读取失败"
            result.at[index, "当前创意状态"] = cache[key]
    finally:
        if tencent is None:
            client.session.close()
    return result


def _rule_config(config: RoiConfig) -> RuleConfig:
    return RuleConfig(
        self_stop_min_age=config.self_stop_min_age,
        self_up_min_age=config.self_up_min_age,
        self_min_daily_uv=config.self_min_daily_uv,
        partner_min_daily_uv=config.partner_min_daily_uv,
        observe_min_latest_uv=config.observe_min_latest_uv,
        one_day_min_uv=config.one_day_min_uv,
        one_day_p30_min_uv=config.one_day_p30_min_uv,
        one_day_hard_stop_roi=config.one_day_hard_stop_roi,
        one_day_stop_quantile=config.one_day_stop_quantile,
        stop_quantile=config.stop_quantile,
        up_quantile=config.up_quantile,
    )


def _dates(start_date: str) -> list[str]:
    start = datetime.strptime(start_date, "%Y%m%d")
    return [
        (start + timedelta(days=offset)).strftime("%Y%m%d")
        for offset in range(3)
    ]


def _run_identity(
    end_date: str,
    fission_parameters: FissionMultiplierParameters,
    source_revision: str | None = None,
) -> tuple[str, str]:
    if source_revision and not re.fullmatch(
        r"[a-z0-9][a-z0-9_-]{0,63}", source_revision
    ):
        raise ValueError(
            "source_revision must use lowercase letters, numbers, '_' or '-'"
        )
    release = fission_parameters.release
    run_suffixes = [release.run_suffix]
    run_key_versions = [release.version]
    run_id = (
        f"roi_{end_date}_{METRIC_RUN_SUFFIX}_{POLICY_RUN_SUFFIX}_"
        f"{'_'.join(run_suffixes)}_{REPORT_RUN_SUFFIX}"
    )
    run_key = (
        f"{METRIC_VERSION}:{POLICY_VERSION}:{':'.join(run_key_versions)}:"
        f"{REPORT_VERSION}:{end_date}"
    )
    if source_revision:
        run_id = f"{run_id}_{source_revision}"
        revision_hash = hashlib.sha256(source_revision.encode("utf-8")).hexdigest()
        run_key = f"{run_key}:sr:{revision_hash[:16]}"
    if len(run_id) > 64:
        raise ValueError("ROI run_id exceeds database limit")
    if len(run_key) > 128:
        raise ValueError("ROI run_key exceeds database limit")
    return run_id, run_key


def run_daily_roi(
    *,
    requested_end_date: str | None = None,
    output_dir: Path,
    send_feishu: bool,
    now: datetime | None = None,
    source_revision: str | None = None,
    internal_test: bool = False,
) -> dict[str, Any]:
    """Compute, snapshot, report, and optionally publish one ROI batch."""

    config = RoiConfig.from_env()
    agency_webhook_config = AgencyWebhookConfig.from_env()
    if internal_test and send_feishu:
        raise ValueError("internal_test cannot publish formal Feishu notifications")
    if internal_test and not (agency_webhook_config.webhooks or {}).get("内部"):
        raise ValueError("Internal ROI test requires agency webhook route: 内部")
    effective_now = now or datetime.now(SHANGHAI)
    if effective_now.tzinfo is None:
        effective_now = effective_now.replace(tzinfo=SHANGHAI)
    if internal_test:
        if source_revision:
            raise ValueError("internal_test manages source_revision automatically")
        source_revision = _internal_source_revision(config, effective_now)

    client = ODPSClient(project=os.getenv("ODPS_PROJECT", "loghubods"))
    end_date = resolve_end_date(
        client,
        requested_end_date,
        now=effective_now,
    )
    start_date, end_date = date_window(end_date)
    expected_dates = _dates(start_date)

    initialize_schema()
    fission_version = os.getenv(
        "ROI_FISSION_PARAMETER_VERSION",
        DEFAULT_FISSION_PARAMETER_VERSION,
    )
    fission_parameters = parameters_from_database(
        *load_fission_parameter_release(fission_version)
    )
    run_config = config.snapshot()
    run_config["fission_multiplier"] = fission_parameters.snapshot()
    run_config["report_version"] = REPORT_VERSION
    run_config["agency_webhook"] = agency_webhook_config.snapshot()
    run_config["internal_test"] = internal_test
    if source_revision:
        run_config["source_revision"] = source_revision
    run_id, run_key = _run_identity(
        end_date,
        fission_parameters,
        source_revision,
    )
    run = create_or_load_run(
        {
            "run_id": run_id,
            "run_key": run_key,
            "metric_version": METRIC_VERSION,
            "policy_version": POLICY_VERSION,
            "fission_parameter_version": fission_parameters.release.version,
            "fission_cohort_date": datetime.strptime(
                fission_parameters.release.cohort_date, "%Y%m%d"
            ).date(),
            "start_date": datetime.strptime(start_date, "%Y%m%d").date(),
            "end_date": datetime.strptime(end_date, "%Y%m%d").date(),
            "config": run_config,
        }
    )
    reusable_statuses = FINAL_STATUSES | {"PENDING_APPROVAL", "EXECUTING"}
    publish_requested = send_feishu or internal_test
    if run.get("status") in reusable_statuses or (
        run.get("status") == "COMPUTED" and not publish_requested
    ):
        logger.info("Reuse ROI run=%s status=%s", run_id, run.get("status"))
        reused_result: dict[str, Any] = {
            "run_id": run_id,
            "status": run.get("status"),
            "reused": True,
            "sheet_url": run.get("sheet_url"),
        }
        if send_feishu and agency_webhook_config.enabled:
            stored_deliveries = load_agency_deliveries(run_id)
            if stored_deliveries:
                retry_reports = [
                    {
                        "agency_name": row["agency_name"],
                        "report_version": row["agency_report_version"],
                        "report": row["file_path"],
                        "creative_rows": row.get("creative_rows") or 0,
                        "ad_rows": row.get("ad_rows") or 0,
                    }
                    for row in stored_deliveries
                ]
                publisher = RoiFeishuPublisher()
                try:
                    try:
                        reused_result["agency_deliveries"] = publish_agency_reports(
                            run_id=run_id,
                            reports=retry_reports,
                            config=agency_webhook_config,
                            publisher=publisher,
                            now=now,
                        )
                    except Exception as exc:
                        safe_error = _redact_agency_webhooks(
                            exc,
                            agency_webhook_config,
                        )
                        logger.error("Agency ROI retry failed: %s", safe_error)
                        reused_result["agency_delivery_error"] = safe_error
                finally:
                    publisher.close()
        return reused_result

    try:
        logger.info(
            "ROI %s loading ODPS daily data %s..%s fission_parameter=%s",
            run_id,
            start_date,
            end_date,
            fission_parameters.release.version,
        )
        daily = fetch_daily_data(client, start_date, end_date)
        ad_age = fetch_ad_age(client, end_date)
        _, thresholds, summary = evaluate_rules(
            daily,
            expected_dates,
            ad_age,
            _rule_config(config),
            fission_parameters=fission_parameters,
        )
        managed_ids = (
            set()
            if internal_test
            else {
                int(row["account_id"])
                for row in load_managed_accounts()
            }
        )
        annotated, snapshots, actions = annotate_execution(
            summary,
            managed_ids,
            run_id,
        )
        annotated = _annotate_current_creative_status(annotated)
        fission_match_summary = (
            summary.groupby(
                ["entity_type", "传播裂变系数匹配层级"],
                as_index=False,
                dropna=False,
            )
            .size()
            .rename(
                columns={
                    "entity_type": "实体类型",
                    "传播裂变系数匹配层级": "匹配层级",
                    "size": "实体数",
                }
            )
        )
        channel_totals = fission_match_summary.groupby("实体类型")[
            "实体数"
        ].transform("sum")
        fission_match_summary["渠道实体数"] = channel_totals
        fission_match_summary["匹配率"] = (
            fission_match_summary["实体数"] / channel_totals
        )
        fission_match_summary["参数版本"] = fission_parameters.release.version
        recommendation_rows = annotated[annotated["动作"].ne("")].copy()
        report_rows = annotated[
            annotated["阈值样本状态"].isin(
                [
                    "进入三日统一阈值样本池",
                    "广告级三日合格_不进入阈值样本池",
                    "单日补充决策_昨日UV>200",
                    "补充观察_昨日UV>200",
                ]
            )
        ].copy()
        for row in snapshots:
            row["run_id"] = run_id
        for row in actions:
            row["run_id"] = run_id
        threshold_record = {
            "统计窗口": f"{start_date} 至 {end_date}",
            "整体三日关停线": thresholds.to_dict("records"),
        }
        replace_run_results(
            run_id,
            snapshots=snapshots,
            actions=actions,
            thresholds=threshold_record,
        )

        output_dir.mkdir(parents=True, exist_ok=True)
        batch_name = f"ROI调控_{start_date}-{end_date}"
        if source_revision:
            batch_name = f"{batch_name}_{source_revision}"
        output_path = output_dir / f"{batch_name}.xlsx"
        if not internal_test:
            write_workbook(
                report_rows,
                thresholds,
                expected_dates,
                output_path,
                run_config,
                fission_match_summary,
            )
        agency_report_date = effective_now.strftime("%Y%m%d")
        agency_reports = write_agency_workbooks(
            report_rows,
            output_dir / f"{agency_report_date}_调控建议",
            agency_report_date,
            agency_names={"自动化投放"} if internal_test else None,
        )
        internal_reports = (
            _build_internal_reports(agency_reports=agency_reports)
            if internal_test
            else []
        )
        result_report = (
            str(internal_reports[0]["report"])
            if internal_test
            else str(output_path)
        )

        result: dict[str, Any] = {
            "run_id": run_id,
            "batch_name": batch_name,
            "status": "COMPUTED",
            "reused": False,
            "start_date": start_date,
            "end_date": end_date,
            "source_revision": source_revision,
            "entity_count": len(snapshots),
            "candidate_count": len(recommendation_rows),
            "actionable_count": len(actions),
            "thresholds": threshold_record,
            "fission_multiplier_matches": fission_match_summary.to_dict(
                "records"
            ),
            "report": result_report,
            "agency_reports": agency_reports,
        }
        if not publish_requested:
            return result

        if internal_test:
            internal_url = (agency_webhook_config.webhooks or {})["内部"]
            internal_config = _internal_webhook_config(
                internal_reports,
                internal_url,
            )
            publisher = RoiFeishuPublisher(require_chat_ids=False)
            try:
                deliveries = publish_agency_reports(
                    run_id=run_id,
                    reports=internal_reports,
                    config=internal_config,
                    publisher=publisher,
                    now=effective_now,
                )
            finally:
                publisher.close()
            result["internal_deliveries"] = deliveries
            failed = [row for row in deliveries if row.get("status") != "SENT"]
            if failed:
                raise RuntimeError(
                    f"Internal ROI test delivery failed for {len(failed)} report(s)"
                )
            main_delivery = next(
                row for row in deliveries if row["agency_name"] == "自动化投放"
            )
            mark_published(
                run_id,
                sheet_token=str(main_delivery["sheet_token"]),
                sheet_url=str(main_delivery["sheet_url"]),
                message_id="internal-webhook",
                expires_at=None,
                requires_approval=False,
            )
            result.update(
                {
                    "status": "COMPLETED",
                    "sheet_url": main_delivery["sheet_url"],
                }
            )
            return result

        counts = recommendation_rows.groupby("动作").size().to_dict()
        summary_text = (
            f"统计窗口：{start_date} - {end_date}\n"
            f"关停建议：{counts.get('关停', 0)}\n"
            f"观察：{counts.get('观察', 0)}\n"
            f"可执行动作：{len(actions)}\n"
            f"审批有效期：发送后 {config.approval_ttl_minutes} 分钟"
        )
        publisher = RoiFeishuPublisher()
        try:
            published = publisher.publish(
                output_path,
                run_id=run_id,
                batch_name=batch_name,
                summary=summary_text,
                requires_approval=bool(actions),
            )
            expires_at = (
                effective_now
                + timedelta(minutes=config.approval_ttl_minutes)
                if actions
                else None
            )
            mark_published(
                run_id,
                sheet_token=published["sheet_token"],
                sheet_url=published["url"],
                message_id=published["message_id"],
                expires_at=expires_at,
                requires_approval=bool(actions),
            )
            try:
                result["agency_deliveries"] = publish_agency_reports(
                    run_id=run_id,
                    reports=agency_reports,
                    config=agency_webhook_config,
                    publisher=publisher,
                    now=now,
                )
            except Exception as exc:
                safe_error = _redact_agency_webhooks(
                    exc,
                    agency_webhook_config,
                )
                logger.error("Agency ROI delivery phase failed: %s", safe_error)
                result["agency_delivery_error"] = safe_error
        finally:
            publisher.close()
        result.update(
            {
                "status": "PENDING_APPROVAL" if actions else "COMPLETED",
                "sheet_url": published["url"],
                "expires_at": expires_at.isoformat() if expires_at else None,
            }
        )
        return result
    except Exception as exc:
        mark_failed(run_id, str(exc))
        raise