"""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, "") 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