"""Preview and execute confirmed Feishu operator commands.""" from __future__ import annotations import logging import uuid from datetime import datetime, timedelta from typing import Any from operator_commands import ( ACTION_DAY_PAUSE, ACTION_RESUME, ACTION_STOP, ParsedCommand, ) from storage import ( advisory_lock, clear_operator_pause, create_operator_command_with_items, find_pending_command_conflict, insert_operator_command_item, load_ad_states, load_operator_command_items, load_operator_command, load_operator_pauses, load_realtime_accounts, set_operator_pause, transition_operator_command, update_operator_command, update_operator_command_item, upsert_ad_state, ) from tencent_client import ( ACTIVE_STATUS, SUSPEND_STATUS, PostWriteVerificationError, TencentWriteNotSentError, TencentWriteOutcomeUnknownError, TencentWriteRejectedError, TencentClient, current_bid_fen, resolve_bid_field, ) COMMAND_PENDING = "PENDING_CONFIRMATION" COMMAND_EXECUTING = "EXECUTING" COMMAND_SUCCEEDED = "SUCCEEDED" COMMAND_PARTIAL = "PARTIAL" COMMAND_FAILED = "FAILED" COMMAND_CANCELLED = "CANCELLED" COMMAND_EXPIRED = "EXPIRED" FINAL_COMMAND_STATUSES = { COMMAND_SUCCEEDED, COMMAND_PARTIAL, COMMAND_FAILED, COMMAND_CANCELLED, COMMAND_EXPIRED, } PAUSE_UNTIL_NEXT_DELIVERY = "UNTIL_NEXT_DELIVERY" PAUSE_UNTIL_MANUAL = "UNTIL_MANUAL" logger = logging.getLogger("tencent_realtime_control.operator_control") def _managed_account_map() -> dict[int, dict[str, Any]]: return { int(row["account_id"]): row for row in load_realtime_accounts() } def resolve_target_accounts(parsed: ParsedCommand) -> list[dict[str, Any]]: managed = _managed_account_map() if parsed.scope_type == "ALL": accounts = list(managed.values()) if not accounts: raise ValueError("当前没有纳入实时控制的账户") return accounts missing = [account_id for account_id in parsed.account_ids if account_id not in managed] if missing: raise ValueError(f"账户不在实时控制范围: {missing}") return [managed[account_id] for account_id in parsed.account_ids] def _next_delivery_start(now: datetime, start_hour: int) -> datetime: return datetime.combine( now.date() + timedelta(days=1), datetime.min.time().replace(hour=start_hour), now.tzinfo, ) def _load_today_metrics( client: TencentClient, account_id: int, adgroup_ids: list[int], now: datetime, ) -> dict[int, dict[str, int]]: metrics: dict[int, dict[str, int]] = {} for start in range(0, len(adgroup_ids), 100): metrics.update( client.get_today_ad_metrics( account_id, adgroup_ids[start:start + 100], now.date(), ) ) return metrics def _operator_pause_is_active(state: dict[str, Any], now: datetime) -> bool: mode = str(state.get("operator_pause_mode") or "") if mode == PAUSE_UNTIL_MANUAL: return True if mode != PAUSE_UNTIL_NEXT_DELIVERY: return False resume_at = state.get("operator_resume_at") if not resume_at: return False if resume_at.tzinfo is None: resume_at = resume_at.replace(tzinfo=now.tzinfo) return resume_at > now def preview_write_command( parsed: ParsedCommand, *, now: datetime, source_message_id: str, chat_id: str, sender_open_id: str, sender_name: str | None, confirmation_ttl_minutes: int, start_hour: int, raw_text: str = "", parse_source: str = "deterministic", intent: dict[str, Any] | None = None, preview_lock_name: str = "tencent_operator_command_preview", tencent: TencentClient | None = None, ) -> dict[str, Any]: accounts = resolve_target_accounts(parsed) client = tencent or TencentClient() next_day = (now.date() + timedelta(days=1)).isoformat() preview_items: list[dict[str, Any]] = [] account_summaries: list[dict[str, Any]] = [] if parsed.action in {ACTION_DAY_PAUSE, ACTION_STOP}: for account in accounts: account_id = int(account["account_id"]) operator_pauses = { int(row["adgroup_id"]): row for row in load_operator_pauses([account_id]) if _operator_pause_is_active(row, now) } ads = [ ad for ad in client.get_ads(account_id) if ( ( str(ad.get("configured_status") or "") == ACTIVE_STATUS and ( parsed.action == ACTION_STOP or not str(ad.get("begin_date") or "") or str(ad.get("begin_date")) <= now.date().isoformat() ) and ( parsed.action == ACTION_STOP or not str(ad.get("end_date") or "") or str(ad.get("end_date")) >= next_day ) and ( parsed.action == ACTION_STOP or int(ad.get("adgroup_id") or 0) not in operator_pauses ) ) or ( parsed.action == ACTION_STOP and str(ad.get("configured_status") or "") == SUSPEND_STATUS and ( operator_pauses.get( int(ad.get("adgroup_id") or 0), {}, ).get("operator_pause_mode") == PAUSE_UNTIL_NEXT_DELIVERY ) ) ) ] metrics = _load_today_metrics( client, account_id, [int(ad["adgroup_id"]) for ad in ads], now ) account_cost = 0 for ad in ads: adgroup_id = int(ad["adgroup_id"]) values = metrics.get(adgroup_id, {}) account_cost += int(values.get("cost_fen") or 0) preview_items.append({ "account_id": account_id, "audience_name": account.get("audience_name"), "adgroup_id": adgroup_id, "adgroup_name": ad.get("adgroup_name"), "before_status": ad.get("configured_status"), "target_status": ( SUSPEND_STATUS if parsed.action == ACTION_STOP else ad.get("configured_status") ), "preview_cost_fen": int(values.get("cost_fen") or 0), "preview_impressions": int(values.get("impressions") or 0), "preview_clicks": int(values.get("clicks") or 0), "preview_conversions": int(values.get("conversions") or 0), "previewed_at": now, "execution_status": "PREVIEWED", }) account_summaries.append({ "account_id": account_id, "ad_count": len(ads), "cost_fen": account_cost, }) elif parsed.action == ACTION_RESUME: pauses_by_account: dict[int, list[dict[str, Any]]] = {} for state in load_operator_pauses([int(row["account_id"]) for row in accounts]): if not _operator_pause_is_active(state, now): continue pauses_by_account.setdefault(int(state["account_id"]), []).append(state) for account in accounts: account_id = int(account["account_id"]) states = pauses_by_account.get(account_id, []) ads = { int(ad.get("adgroup_id") or 0): ad for ad in client.get_ads(account_id) } ids = [int(state["adgroup_id"]) for state in states] metrics = _load_today_metrics(client, account_id, ids, now) account_cost = 0 for state in states: adgroup_id = int(state["adgroup_id"]) ad = ads.get(adgroup_id, {}) values = metrics.get(adgroup_id, {}) account_cost += int(values.get("cost_fen") or 0) preview_items.append({ "account_id": account_id, "audience_name": account.get("audience_name"), "adgroup_id": adgroup_id, "adgroup_name": ad.get("adgroup_name") or state.get("adgroup_name"), "before_status": ad.get("configured_status"), "target_status": ( ACTIVE_STATUS if state.get("operator_pause_mode") == PAUSE_UNTIL_MANUAL else ad.get("configured_status") ), "preview_cost_fen": int(values.get("cost_fen") or 0), "preview_impressions": int(values.get("impressions") or 0), "preview_clicks": int(values.get("clicks") or 0), "preview_conversions": int(values.get("conversions") or 0), "previewed_at": now, "execution_status": "PREVIEWED", }) account_summaries.append({ "account_id": account_id, "ad_count": len(states), "cost_fen": account_cost, }) else: raise ValueError(f"Unsupported write action: {parsed.action}") if not preview_items: raise ValueError("当前范围内没有符合操作条件的广告") impacted_account_ids = sorted({int(row["account_id"]) for row in preview_items}) impacted_accounts = [ account for account in accounts if int(account["account_id"]) in impacted_account_ids ] account_summaries = [ row for row in account_summaries if int(row["account_id"]) in impacted_account_ids ] command_id = f"cmd_{now.strftime('%Y%m%d%H%M%S')}_{uuid.uuid4().hex[:8]}" expires_at = now + timedelta(minutes=confirmation_ttl_minutes) resume_at = _next_delivery_start(now, start_hour) if parsed.action == ACTION_DAY_PAUSE else None totals = { "preview_cost_fen": sum(int(row["preview_cost_fen"]) for row in preview_items), "preview_impressions": sum(int(row["preview_impressions"]) for row in preview_items), "preview_clicks": sum(int(row["preview_clicks"]) for row in preview_items), "preview_conversions": sum(int(row["preview_conversions"]) for row in preview_items), } for row in preview_items: row["command_id"] = command_id command_record = { "command_id": command_id, "source_message_id": source_message_id, "chat_id": chat_id, "sender_open_id": sender_open_id, "sender_name": sender_name, "raw_text": raw_text, "parse_source": parse_source, "intent": intent or {}, "action": parsed.action, "scope_type": parsed.scope_type, "target_account_ids": impacted_account_ids, "status": COMMAND_PENDING, "preview_account_count": len(impacted_accounts), "preview_ad_count": len(preview_items), **totals, "previewed_at": now, "resume_at": resume_at, "expires_at": expires_at, } with advisory_lock(preview_lock_name) as acquired: if not acquired: raise RuntimeError("其他运营命令正在生成预览,请稍后重试") conflict = find_pending_command_conflict( [(int(row["account_id"]), int(row["adgroup_id"])) for row in preview_items], now=now, ) if conflict: raise ValueError( "操作范围与待确认命令冲突: " f"{conflict['command_id']},账户 {conflict['account_id']}," f"广告 {conflict['adgroup_id']}" ) command = create_operator_command_with_items(command_record, preview_items) command["account_summaries"] = account_summaries return command def cancel_command(command_id: str, sender_open_id: str, now: datetime) -> dict[str, Any]: command = load_operator_command(command_id) if not command: raise ValueError(f"命令不存在: {command_id}") if command["sender_open_id"] != sender_open_id: raise PermissionError("只能取消自己发起的命令") if command["status"] != COMMAND_PENDING: return command transition_operator_command( command_id, expected_statuses={COMMAND_PENDING}, target_status=COMMAND_CANCELLED, executed_at=now, ) return load_operator_command(command_id) or command def _ensure_state( *, account: dict[str, Any], ad: dict[str, Any], states: dict[int, dict[str, Any]], now: datetime, ) -> dict[str, Any]: adgroup_id = int(ad["adgroup_id"]) state = states.get(adgroup_id) if state: return state bid_field = resolve_bid_field(ad, account.get("bid_scene")) base_bid = current_bid_fen(ad, bid_field) if base_bid is None or base_bid <= 0: raise ValueError( f"广告缺少有效基础出价: account={account['account_id']} ad={adgroup_id}" ) upsert_ad_state( account_id=int(account["account_id"]), adgroup_id=adgroup_id, adgroup_name=str(ad.get("adgroup_name") or ""), bid_field=bid_field, base_bid_fen=base_bid, boosted_date=None, paused_by_strategy=False, pause_reason=None, last_action="REGISTER_BASE", action_at=now, ) state = { "account_id": int(account["account_id"]), "adgroup_id": adgroup_id, "bid_field": bid_field, "base_bid_fen": base_bid, "paused_by_strategy": False, } states[adgroup_id] = state return state def _record_item( command: dict[str, Any], account: dict[str, Any], *, item: dict[str, Any] | None = None, **values: Any, ) -> None: if item and item.get("id"): update_operator_command_item(int(item["id"]), **values) return insert_operator_command_item( { "command_id": command["command_id"], "account_id": int(account["account_id"]), "audience_name": account.get("audience_name"), **values, } ) def _execute_pause( command: dict[str, Any], account: dict[str, Any], *, now: datetime, start_hour: int, client: TencentClient, items: list[dict[str, Any]], ) -> tuple[int, int, int]: account_id = int(account["account_id"]) try: ads = { int(ad.get("adgroup_id") or 0): ad for ad in client.get_ads(account_id) } except Exception as exc: for item in items: _record_item( command, account, item=item, execution_status="FAILED", error_message=f"Tencent ad query failed: {exc}", ) return 0, len(items), 0 states = load_ad_states(account_id) successes = failures = skipped = 0 mode = ( PAUSE_UNTIL_NEXT_DELIVERY if command["action"] == ACTION_DAY_PAUSE else PAUSE_UNTIL_MANUAL ) resume_at = _next_delivery_start(now, start_hour) if mode == PAUSE_UNTIL_NEXT_DELIVERY else None for item in items: adgroup_id = int(item["adgroup_id"]) ad = ads.get(adgroup_id) if not ad: _record_item( command, account, item=item, execution_status="FAILED", error_message="Tencent ad not found", ) failures += 1 continue before_status = str(ad.get("configured_status") or "") if before_status != ACTIVE_STATUS: existing_state = states.get(adgroup_id) if ( command["action"] == ACTION_STOP and before_status == SUSPEND_STATUS and existing_state and existing_state.get("operator_pause_mode") ): set_operator_pause( account_id=account_id, adgroup_id=adgroup_id, mode=PAUSE_UNTIL_MANUAL, resume_at=None, command_id=command["command_id"], paused_from_status=( existing_state.get("operator_paused_from_status") or ACTIVE_STATUS ), paused_at=now, ) _record_item( command, account, item=item, adgroup_id=adgroup_id, adgroup_name=ad.get("adgroup_name"), before_status=before_status, target_status=SUSPEND_STATUS, readback_status=before_status, execution_status="SUCCESS", ) successes += 1 continue _record_item( command, account, item=item, adgroup_id=adgroup_id, adgroup_name=ad.get("adgroup_name"), before_status=before_status, target_status=( before_status if mode == PAUSE_UNTIL_NEXT_DELIVERY else SUSPEND_STATUS ), readback_status=before_status, execution_status="SKIPPED_NOT_ACTIVE", ) skipped += 1 continue if ( mode == PAUSE_UNTIL_NEXT_DELIVERY and str(ad.get("begin_date") or "") > now.date().isoformat() ): _record_item( command, account, item=item, adgroup_id=adgroup_id, adgroup_name=ad.get("adgroup_name"), before_status=before_status, target_status=before_status, readback_status=before_status, execution_status="SKIPPED_ALREADY_DEFERRED", ) skipped += 1 continue if ( mode == PAUSE_UNTIL_NEXT_DELIVERY and str(ad.get("end_date") or "") and str(ad.get("end_date")) < resume_at.date().isoformat() ): _record_item( command, account, item=item, adgroup_id=adgroup_id, adgroup_name=ad.get("adgroup_name"), before_status=before_status, target_status=before_status, readback_status=before_status, execution_status="SKIPPED_END_DATE", ) skipped += 1 continue tencent_updated = False try: _ensure_state(account=account, ad=ad, states=states, now=now) set_operator_pause( account_id=account_id, adgroup_id=adgroup_id, mode=mode, resume_at=resume_at, command_id=command["command_id"], paused_from_status=before_status, paused_at=now, ) if mode == PAUSE_UNTIL_NEXT_DELIVERY: before_begin_date = str(ad.get("begin_date") or "") next_day = resume_at.date().isoformat() target_begin_date = max(before_begin_date, next_day) if before_begin_date != target_begin_date: client.update_ad_begin_dates( account_id, [adgroup_id], target_begin_date, ) tencent_updated = True readback_status = before_status target_status = before_status else: readback = client.update_ad( account_id, adgroup_id, target_status=SUSPEND_STATUS, ) tencent_updated = True readback_status = readback.get("configured_status") target_status = SUSPEND_STATUS _record_item( command, account, item=item, adgroup_id=adgroup_id, adgroup_name=ad.get("adgroup_name"), before_status=before_status, target_status=target_status, readback_status=readback_status, execution_status="SUCCESS", ) successes += 1 except Exception as exc: if isinstance( exc, (PostWriteVerificationError, TencentWriteOutcomeUnknownError), ): tencent_updated = True safe_to_clear = isinstance( exc, (TencentWriteNotSentError, TencentWriteRejectedError), ) if not tencent_updated and safe_to_clear: clear_operator_pause( account_id, adgroup_id, action="OPERATOR_PAUSE_FAILED", action_at=now, ) try: _record_item( command, account, item=item, adgroup_id=adgroup_id, adgroup_name=ad.get("adgroup_name"), before_status=before_status, target_status=( before_status if mode == PAUSE_UNTIL_NEXT_DELIVERY else SUSPEND_STATUS ), readback_status=( exc.actual.get("configured_status") if isinstance(exc, PostWriteVerificationError) else ( before_status if mode == PAUSE_UNTIL_NEXT_DELIVERY else SUSPEND_STATUS ) if tencent_updated else None ), execution_status=( "VERIFY_FAILED" if isinstance(exc, PostWriteVerificationError) else "OUTCOME_UNKNOWN" if isinstance(exc, TencentWriteOutcomeUnknownError) else "AUDIT_FAILED" if tencent_updated else "FAILED" ), error_message=str(exc), ) except Exception: logger.exception( "Failed to persist operator command failure item " "command=%s account=%s ad=%s", command["command_id"], account_id, adgroup_id, ) failures += 1 return successes, failures, skipped def _execute_resume( command: dict[str, Any], account: dict[str, Any], *, now: datetime, client: TencentClient, items: list[dict[str, Any]], ) -> tuple[int, int, int]: account_id = int(account["account_id"]) pauses = { int(row["adgroup_id"]): row for row in load_operator_pauses([account_id]) } try: ads = { int(ad.get("adgroup_id") or 0): ad for ad in client.get_ads(account_id) } except Exception as exc: for item in items: _record_item( command, account, item=item, execution_status="FAILED", error_message=f"Tencent ad query failed: {exc}", ) return 0, len(items), 0 successes = failures = skipped = 0 for item in items: adgroup_id = int(item["adgroup_id"]) state = pauses.get(adgroup_id) if not state: _record_item( command, account, item=item, execution_status="SKIPPED_NOT_PAUSED", ) skipped += 1 continue ad = ads.get(adgroup_id) if not ad: _record_item( command, account, item=item, adgroup_id=adgroup_id, adgroup_name=state.get("adgroup_name"), execution_status="FAILED", error_message="Tencent ad not found", ) failures += 1 continue before_status = str(ad.get("configured_status") or "") try: readback_status = before_status target_status = None if ( state.get("operator_pause_mode") == PAUSE_UNTIL_NEXT_DELIVERY and str(ad.get("begin_date") or "") > now.date().isoformat() and not bool(state.get("paused_by_strategy")) ): client.update_ad_begin_dates( account_id, [adgroup_id], now.date().isoformat(), ) elif ( before_status == SUSPEND_STATUS and not bool(state.get("paused_by_strategy")) ): target_status = ACTIVE_STATUS readback = client.update_ad( account_id, adgroup_id, target_status=ACTIVE_STATUS, ) readback_status = str(readback.get("configured_status") or "") clear_operator_pause( account_id, adgroup_id, action="OPERATOR_RESUME", action_at=now, ) _record_item( command, account, item=item, adgroup_id=adgroup_id, adgroup_name=ad.get("adgroup_name"), before_status=before_status, target_status=target_status, readback_status=readback_status, execution_status="SUCCESS", ) successes += 1 except Exception as exc: _record_item( command, account, item=item, adgroup_id=adgroup_id, adgroup_name=ad.get("adgroup_name"), before_status=before_status, target_status=ACTIVE_STATUS, execution_status="FAILED", error_message=str(exc), ) failures += 1 return successes, failures, skipped def execute_confirmed_command( command_id: str, *, sender_open_id: str, now: datetime, start_hour: int, lock_name: str, tencent: TencentClient | None = None, ) -> dict[str, Any]: command = load_operator_command(command_id) if not command: raise ValueError(f"命令不存在: {command_id}") if command["sender_open_id"] != sender_open_id: raise PermissionError("只能确认自己发起的命令") if command["status"] in FINAL_COMMAND_STATUSES: return command if command["status"] not in {COMMAND_PENDING, COMMAND_EXECUTING}: raise RuntimeError(f"命令当前不可确认: {command['status']}") with advisory_lock(lock_name) as acquired: if not acquired: raise RuntimeError("实时控制正在执行,请稍后再次确认") command = load_operator_command(command_id) or command if command["status"] in FINAL_COMMAND_STATUSES: return command if command["status"] not in {COMMAND_PENDING, COMMAND_EXECUTING}: raise RuntimeError(f"命令当前不可确认: {command['status']}") if ( command["status"] == COMMAND_PENDING and command.get("expires_at") and now.replace(tzinfo=None) > command["expires_at"] ): transition_operator_command( command_id, expected_statuses={COMMAND_PENDING}, target_status=COMMAND_EXPIRED, executed_at=now, ) return load_operator_command(command_id) or command if command["status"] == COMMAND_PENDING: transitioned = transition_operator_command( command_id, expected_statuses={COMMAND_PENDING}, target_status=COMMAND_EXECUTING, confirmed_at=now, ) if not transitioned: latest = load_operator_command(command_id) or command if latest["status"] in FINAL_COMMAND_STATUSES: return latest raise RuntimeError( f"命令状态并发变化: {latest['status']}" ) try: managed = _managed_account_map() missing = [ account_id for account_id in command["target_account_ids"] if account_id not in managed ] if missing: raise RuntimeError( f"确认时账户已不在自动化管理范围: {missing}" ) accounts = [ managed[account_id] for account_id in command["target_account_ids"] ] client = tencent or TencentClient() frozen_items = load_operator_command_items(command_id) if not frozen_items: raise RuntimeError("命令缺少冻结广告明细,请重新发起") items_by_account: dict[int, list[dict[str, Any]]] = {} for item in frozen_items: items_by_account.setdefault(int(item["account_id"]), []).append(item) successes = failures = skipped = 0 for account in accounts: if command["action"] in {ACTION_DAY_PAUSE, ACTION_STOP}: ok, failed, ignored = _execute_pause( command, account, now=now, start_hour=start_hour, client=client, items=items_by_account.get(int(account["account_id"]), []), ) elif command["action"] == ACTION_RESUME: ok, failed, ignored = _execute_resume( command, account, now=now, client=client, items=items_by_account.get(int(account["account_id"]), []), ) else: raise ValueError(f"Unsupported command action: {command['action']}") successes += ok failures += failed skipped += ignored except Exception as exc: update_operator_command( command_id, COMMAND_FAILED, executed_at=now, error_message=str(exc), ) raise status = ( COMMAND_FAILED if failures and not successes else COMMAND_PARTIAL if failures else COMMAND_SUCCEEDED ) update_operator_command(command_id, status, executed_at=now) result = load_operator_command(command_id) or command result["successes"] = successes result["failures"] = failures result["skipped"] = skipped return result def pause_status_summary(now: datetime) -> dict[str, Any]: account_ids = [int(row["account_id"]) for row in load_realtime_accounts()] pauses = [ row for row in load_operator_pauses(account_ids) if _operator_pause_is_active(row, now) ] return { "total": len(pauses), "until_next_delivery": sum( row["operator_pause_mode"] == PAUSE_UNTIL_NEXT_DELIVERY for row in pauses ), "until_manual": sum( row["operator_pause_mode"] == PAUSE_UNTIL_MANUAL for row in pauses ), "accounts": sorted({int(row["account_id"]) for row in pauses}), }