""" 报告生成工具 — auto_put_ad_mini V3 职责： - 读取决策引擎输出 - 生成 CSV（程序用）+ XLSX（人工查阅，带条件格式） - 输出到 outputs/reports/ """ import logging from datetime import datetime, timedelta from pathlib import Path from typing import Optional import pandas as pd from agent.tools import tool from agent.tools.models import ToolContext, ToolResult logger = logging.getLogger(__name__) _MINI_DIR = Path(__file__).resolve().parent.parent _REPORTS_DIR = _MINI_DIR / "outputs" / "reports" # 最终输出列顺序（审批表格：简洁版，去掉技术性列） # 与 im_approval.py 的 APPROVAL_COLUMNS 保持一致（15列精简版） OUTPUT_COLUMNS = [ # 核心标识（前5列，含决策动作） "approval_date", "account_id", "ad_id", "cost_7d_avg", "action", # 基础信息 "ad_name", "audience_tier", "ad_age_days", "bid_amount", # 关键指标 "动态ROI_7日均值", "cost_7d_total", "revenue_7d_total", # 决策详情 "dimension", "reason", "recommended_change_pct", ] # 中文列名映射 CN_COLUMNS = { "approval_date": "日期", "account_id": "账户ID", "ad_id": "广告ID", "cost_7d_avg": "广告消耗(7日日均/元)", "ad_name": "广告名称", "audience_tier": "人群包", "create_time": "创建时间", "ad_age_days": "投放天数", "configured_status": "广告状态", "bid_amount": "出价(元)", "yesterday_cost": "昨日消耗(元)", "yesterday_revenue": "昨日收入(元)", "yesterday_roi": "昨日ROI", "cost_7d_total": "7日总消耗(元)", "revenue_7d_total": "7日总收入(元)", "T0裂变系数_latest": "T0裂变系数(最新)", "arpu_latest": "ARPU(最新)", "a_latest": "a值(最新)", "b_7d_mean": "b值(7日均值)", "T0裂变系数_7d_mean": "T0裂变系数(7日均值)", "e_factor": "e因子", "f_7日动态ROI": "7日均值动态ROI", "动态ROI": "动态ROI(单日)", "动态ROI_7日均值": "动态ROI(7日均值)", "cost_30d_total": "30日总消耗(元)", "cost_30d_avg": "30日日均消耗(元)", "stable_spend_days_30d": "稳定消耗天数(30日)", "creative_count": "创意数", "action": "决策动作", "dimension": "命中维度", "reason": "决策理由", "recommended_change_pct": "建议调幅(%)", "current_bid": "当前出价(元)", "recommended_bid": "建议出价(元)", "guardrail_reason": "护栏说明", "execution_status": "执行状态", "f_7日动态ROI_mean_all": "全体动态ROI均值", "source": "数据来源", } # 动作中文映射 ACTION_CN_MAP = { "pause": "关停", "bid_down": "降价", "bid_up": "提价", "hold": "保持", "scale_up": "扩量", "observe": "观察", } def _write_xlsx_with_format(df: pd.DataFrame, path: Path) -> None: """生成带条件格式的 XLSX 文件。""" try: import openpyxl from openpyxl.styles import Font, Alignment, PatternFill from openpyxl.utils import get_column_letter except ImportError: logger.warning("openpyxl 未安装，跳过 XLSX 生成") return # 动作中文化 if "action" in df.columns: df["action"] = df["action"].map(ACTION_CN_MAP).fillna(df["action"]) # 中文列名 df_cn = df.rename(columns=CN_COLUMNS) df_cn.to_excel(path, index=False, engine="openpyxl") wb = openpyxl.load_workbook(path) ws = wb.active # 表头样式（加粗 + 灰色背景） header_font = Font(bold=True, size=10) header_fill = PatternFill(start_color="D3D3D3", end_color="D3D3D3", fill_type="solid") # 灰色背景 center_alignment = Alignment(horizontal="center", vertical="center") for cell in ws[1]: cell.font = header_font cell.fill = header_fill cell.alignment = center_alignment # 所有数据单元格居中对齐 for row in ws.iter_rows(min_row=2, max_row=ws.max_row): for cell in row: cell.alignment = center_alignment # 自动列宽（增加30%） for col_idx in range(1, ws.max_column + 1): col_letter = get_column_letter(col_idx) max_len = max( len(str(ws.cell(row=r, column=col_idx).value or "")) for r in range(1, min(ws.max_row + 1, 50)) ) # 列宽增加30%：原公式 max_len + 4，现改为 (max_len + 4) * 1.3 ws.column_dimensions[col_letter].width = min((max_len + 4) * 1.3, 40) # 决策动作列条件格式化 action_col_idx = None for idx, cell in enumerate(ws[1], 1): if cell.value == "决策动作": action_col_idx = idx # E1 单元格（决策动作表头）黄色高亮 yellow_header_fill = PatternFill(start_color="FFFF00", end_color="FFFF00", fill_type="solid") cell.fill = yellow_header_fill break if action_col_idx: # 颜色定义 yellow_fill = PatternFill(start_color="FFFF00", end_color="FFFF00", fill_type="solid") # 黄色（整列默认） green_fill = PatternFill(start_color="90EE90", end_color="90EE90", fill_type="solid") # 浅绿色 red_fill = PatternFill(start_color="FF6B6B", end_color="FF6B6B", fill_type="solid") # 红色 orange_fill = PatternFill(start_color="FFB84D", end_color="FFB84D", fill_type="solid") # 橘黄色 # 为决策动作列的所有数据单元格设置颜色 for row_idx in range(2, ws.max_row + 1): cell = ws.cell(row=row_idx, column=action_col_idx) value = str(cell.value).strip() if cell.value else "" # 根据值设置颜色 if value == "保持": cell.fill = green_fill elif value == "关停": cell.fill = red_fill elif value == "降价": cell.fill = orange_fill else: # 其他值（如"提价"）使用黄色 cell.fill = yellow_fill # 启用自动筛选（首行） ws.auto_filter.ref = ws.dimensions # 冻结窗格：锁定第一行和前五列（冻结到F2单元格） ws.freeze_panes = "F2" wb.save(path) logger.info("XLSX 已生成: %s", path) @tool(description="生成决策报告（CSV + XLSX 带条件格式）") async def generate_report( ctx: ToolContext = None, decision_csv: str = "", end_date: str = "yesterday", ) -> ToolResult: """ 生成决策报告。 Args: decision_csv: 决策结果 CSV 路径（apply_decisions 输出） end_date: 结束日期 Returns: ToolResult，包含报告路径 """ try: if not decision_csv: decision_csv = str(_MINI_DIR / "outputs" / "decision_temp.csv") df = pd.read_csv(decision_csv) if df.empty: return ToolResult(title="报告生成", output="决策数据为空") if end_date == "yesterday": end_date = (datetime.now() - timedelta(days=1)).strftime("%Y%m%d") _REPORTS_DIR.mkdir(parents=True, exist_ok=True) # 添加审批日期列（当前日期） df["approval_date"] = datetime.now().strftime("%Y-%m-%d") # 选择输出列（存在的列） cols = [c for c in OUTPUT_COLUMNS if c in df.columns] df_out = df[cols].copy() # 排序：有消耗的在前，无消耗的在后，同组内按消耗降序 sort_cols = [] ascending_flags = [] if "cost_7d_avg" in df_out.columns: # 无消耗（cost_7d_avg<10元）放最后 df_out["_has_spend"] = (df_out["cost_7d_avg"] >= 10.0).astype(int) sort_cols.append("_has_spend") ascending_flags.append(False) # 1在前，0在后 if "cost_7d_total" in df_out.columns: sort_cols.append("cost_7d_total") ascending_flags.append(False) if sort_cols: df_out = df_out.sort_values(sort_cols, ascending=ascending_flags) if "_has_spend" in df_out.columns: df_out.drop(columns=["_has_spend"], inplace=True) # CSV csv_path = _REPORTS_DIR / f"decision_{end_date}.csv" df_out.to_csv(csv_path, index=False, encoding="utf-8-sig") logger.info("CSV 已生成: %s", csv_path) # XLSX xlsx_path = _REPORTS_DIR / f"decision_{end_date}.xlsx" _write_xlsx_with_format(df_out, xlsx_path) pause_count = (df_out["action"] == "pause").sum() if "action" in df_out.columns else 0 total_count = len(df_out) return ToolResult( title=f"报告已生成（{total_count}个广告）", output=f"报告已生成:\n CSV: {csv_path}\n XLSX: {xlsx_path}\n\n共 {total_count} 个广告，其中 {pause_count} 个建议关停", metadata={ "csv_path": str(csv_path), "xlsx_path": str(xlsx_path), "total_ads": total_count, "pause_ads": pause_count, }, ) except Exception as e: logger.error("报告生成失败: %s", e, exc_info=True) return ToolResult(title="报告生成失败", output=str(e))