from __future__ import annotations # === Writable config dirs === import os os.environ.setdefault("APP_DATA_DIR", "/data/app_data" if os.access("/data", os.W_OK) else "/tmp/app_data") os.environ.setdefault("MPLCONFIGDIR", os.path.join(os.environ["APP_DATA_DIR"], "mplconfig")) os.makedirs(os.environ["MPLCONFIGDIR"], exist_ok=True) import json, uuid, random from datetime import datetime from typing import Dict import gradio as gr import pandas as pd from app.storage import ( init_db, insert_variant, upsert_campaign, get_variant, get_metrics, get_campaign, set_campaign_settings, get_campaign_value_per_conversion, log_event, export_csv, reset_all, evaluate_stop_rules, record_compliance_log, audit, get_variants ) from app.bandit import ThompsonBandit from app.linucb import LinUCB from app.forecast import SeasonalityModel from app.compliance import rule_based_check, llm_check_and_fix from app.openai_client import openai_chat_json from app.adapters import XAdapter, MetaAdapter, GoogleAdsAdapter # 初期化 init_db() _seasonality_cache: Dict[str, SeasonalityModel] = {} GENERATE_COLUMNS = ["variant_id", "status", "rejection_reason", "text"] REPORT_COLUMNS = ["variant_id","impressions","clicks","conversions","ctr","cvr","expected_value"] GEN_SYSTEM = """ あなたは日本語広告コピーのプロフェッショナルコピーライターです。 出力は**次のJSONオブジェクトのみ**で厳密に返してください。余計な文章・説明・前置きは禁止です。 形式: {"variants":[{"headline":"全角15-25字程度","body":"全角40-90字程度"}, ...]} ルール: - 医薬効能の断定、100%、永久、即効、根拠のない数値などの誇大表現は禁止 - CTAは自然に - 日本語で、句読点や記号は自然に """ GEN_USER_TEMPLATE = """ ブランド: {brand} 商品/サービス: {product} 想定ターゲット: {target} トーン: {tone} 制約: {constraints} 生成本数: {k} 要件: - "variants" 配列の要素数は **ちょうど {k}** 件にしてください - 各要素は {{"headline":"...","body":"..."}} のみ """ def _seasonal(campaign_id: str) -> SeasonalityModel: if campaign_id not in _seasonality_cache: m = SeasonalityModel(campaign_id) try: m.fit() except Exception: pass _seasonality_cache[campaign_id] = m return _seasonality_cache[campaign_id] def _safe_get_variants(data, k: int): items = [] if isinstance(data, dict) and isinstance(data.get("variants"), list): items = data["variants"] elif isinstance(data, list): items = data if not items or not all(isinstance(x, dict) for x in items): return None out = [] for it in items[:k]: out.append({ "headline": str(it.get("headline", "")).strip(), "body": str(it.get("body", "")).strip(), }) return out def _local_variants(brand: str, product: str, k: int): base_head = ["使いやすさで選ばれています","日々の習慣をシンプルに","はじめてでも安心","続けやすいサポートを","いま必要な機能だけを"] base_body = [ "{brand}の「{product}」。生活になじむ設計で、今日からムリなく始められます。まずは詳細をご覧ください。", "毎日を少しラクに。{brand}の{product}が、あなたの習慣づくりを後押しします。今すぐチェック。", "難しい操作は不要。{brand}の{product}なら、使い始めから自然に続けられます。詳しくはサイトへ。", "必要な情報をひと目で。{brand}の{product}で、日々の管理をシンプルに。詳細を見る。", "続けやすさを重視。{brand}の{product}で、小さな一歩から。" ] out = [] for i in range(k): out.append({"headline": base_head[i % len(base_head)], "body": base_body[i % len(base_body)].format(brand=brand, product=product)}) return out async def ui_generate(campaign_id: str, brand: str, product: str, target: str, tone: str, k_variants: int, ng_words: str, value_per_conversion: float): k_variants = int(k_variants) constraints = {"ng_words": [w.strip() for w in ng_words.splitlines() if w.strip()]} if ng_words else {} upsert_campaign(campaign_id, brand, product, target, tone, "ja", constraints, value_per_conversion) user = GEN_USER_TEMPLATE.format( brand=brand, product=product, target=target, tone=tone, constraints=json.dumps(constraints, ensure_ascii=False), k=k_variants ) items = None try: data = await openai_chat_json( [{"role": "system", "content": GEN_SYSTEM},{"role": "user", "content": user}], temperature=0.2, max_tokens=1200, ) items = _safe_get_variants(data, k_variants) except Exception: items = None if not items: try: retry_user = user + "\n\n注意: 'variants' は必ず指定件数、各要素は {\"headline\":\"...\",\"body\":\"...\"} のみ。" data = await openai_chat_json( [{"role": "system", "content": GEN_SYSTEM},{"role": "user", "content": retry_user}], temperature=0.1, max_tokens=1000, ) items = _safe_get_variants(data, k_variants) except Exception: items = None if not items: items = _local_variants(brand, product, k_variants) rows = [] for it in items[:k_variants]: headline, body = it["headline"], it["body"] text = f"{headline}\n{body}".strip() vid = str(uuid.uuid4())[:8] ok_rule, bads = rule_based_check(text, (constraints or {}).get("ng_words")) status, rejection = "approved", None ok_llm, reasons, fixed = llm_check_and_fix(text) if not ok_rule and not ok_llm: status, rejection = "rejected", "; ".join(bads + reasons) elif not ok_rule and ok_llm: text = fixed or text elif ok_rule and not ok_llm: text = fixed or text insert_variant(campaign_id, vid, text, status, rejection) record_compliance_log(campaign_id, vid, status, bads, ok_llm, reasons, fixed) rows.append({"variant_id": vid, "status": status, "rejection_reason": rejection or "", "text": text}) df = pd.DataFrame(rows, columns=GENERATE_COLUMNS) return df def _policy_of(campaign_id: str) -> str: cfg = get_campaign(campaign_id) return str(cfg["policy"] or "thompson") if cfg else "thompson" def _holdout_ratio_of(campaign_id: str) -> float: cfg = get_campaign(campaign_id) return float(cfg["holdout_ratio"] or 0.0) if cfg else 0.0 def _context(hour: int, segment: str) -> Dict[str, str | int | None]: return {"hour": int(hour), "segment": (segment or "").strip() or None} def ui_set_settings(campaign_id: str, policy: str, holdout: float, stop_min_impr: int, stop_rel_ev: float): set_campaign_settings(campaign_id, policy, holdout, stop_min_impr, stop_rel_ev) return f"Updated: policy={policy}, holdout={holdout}, stop_min_impressions={stop_min_impr}, stop_rel_ev_threshold={stop_rel_ev}" def _uniform_variant(campaign_id: str) -> str | None: vs = get_variants(campaign_id) if not vs: return None return random.choice(vs)["variant_id"] def ui_serve(campaign_id: str, hour: int, segment: str, aa_min_impr: int): ctx = _context(hour, segment) m = _seasonal(campaign_id) policy = _policy_of(campaign_id) holdout = _holdout_ratio_of(campaign_id) mets = get_metrics(campaign_id) if not mets: raise gr.Error("配信可能なバリアントがありません。まずは Generate してください。") # A/Aテスト：各バリアントのimpressionsがしきい値未満なら一様ランダム if aa_min_impr and aa_min_impr > 0: for r in mets: if int(r["impressions"]) < int(aa_min_impr): vid = _uniform_variant(campaign_id) row = get_variant(campaign_id, vid) log_event(campaign_id, vid, "impression", datetime.utcnow().isoformat(), None) ThompsonBandit.update_with_event(campaign_id, vid, "impression") audit(campaign_id, "serve", {"variant_id": vid, "policy": "AA"}) return vid, row["text"] # ホールドアウト：一定確率で一様ランダム if holdout > 0 and random.random() < float(holdout): vid = _uniform_variant(campaign_id) row = get_variant(campaign_id, vid) log_event(campaign_id, vid, "impression", datetime.utcnow().isoformat(), None) ThompsonBandit.update_with_event(campaign_id, vid, "impression") audit(campaign_id, "serve", {"variant_id": vid, "policy": "holdout"}) return vid, row["text"] # 通常ポリシー if policy == "linucb": bandit = LinUCB(campaign_id) vid, _ = bandit.choose(ctx) else: bandit = ThompsonBandit(campaign_id) vid, _ = bandit.sample_arm(ctx, m.expected_ctr) if not vid: raise gr.Error("バリアントが見つかりません。") row = get_variant(campaign_id, vid) log_event(campaign_id, vid, "impression", datetime.utcnow().isoformat(), None) ThompsonBandit.update_with_event(campaign_id, vid, "impression") audit(campaign_id, "serve", {"variant_id": vid, "policy": policy}) return vid, row["text"] def ui_feedback(campaign_id: str, variant_id: str, event_type: str, hour: int, segment: str): if not variant_id: raise gr.Error("先に Serve してください。") ctx = _context(hour, segment) log_event(campaign_id, variant_id, event_type, datetime.utcnow().isoformat(), None) ThompsonBandit.update_with_event(campaign_id, variant_id, event_type) # LinUCBはclickのみで学習（CTRモデル） if event_type == "click" and _policy_of(campaign_id) == "linucb": LinUCB(campaign_id).update_click(variant_id, ctx, reward=1.0) audit(campaign_id, "feedback", {"variant_id": variant_id, "event": event_type}) return f"{event_type} を記録しました。" def ui_report(campaign_id: str): mets = get_metrics(campaign_id) vpc = get_campaign_value_per_conversion(campaign_id) rows = [] for r in mets: imp = int(r["impressions"]); clk = int(r["clicks"]); conv = int(r["conversions"]) ctr = (clk / imp) if imp > 0 else 0.0 cvr = (conv / clk) if clk > 0 else 0.0 ev = ctr * cvr * vpc rows.append({ "variant_id": r["variant_id"], "impressions": imp, "clicks": clk, "conversions": conv, "ctr": round(ctr, 4), "cvr": round(cvr, 4), "expected_value": round(ev, 6), }) return pd.DataFrame(rows, columns=REPORT_COLUMNS) def ui_apply_stop(campaign_id: str): paused = evaluate_stop_rules(campaign_id) if not paused: return "No changes." return "Paused: " + ", ".join([f"{vid}({reason})" for vid, reason in paused]) def ui_export_csv(campaign_id: str, table: str): path = export_csv(campaign_id, table) return path def ui_reset_db(confirm_text: str): if (confirm_text or "").strip().upper() != "RESET": raise gr.Error("タイプミス: RESET と入力してください。") reset_all() return "DB was reset." def ui_adapter_send(campaign_id: str, platform: str, variant_id: str, text: str, hour: int, segment: str): ctx = _context(hour, segment) if platform == "x": ad = XAdapter(campaign_id) elif platform == "meta": ad = MetaAdapter(campaign_id) else: ad = GoogleAdsAdapter(campaign_id) res = ad.send(variant_id, text, ctx) return json.dumps(res, ensure_ascii=False) with gr.Blocks(title="AdCopy MAB Optimizer", fill_height=True) as demo: gr.Markdown(""" # AdCopy MAB Optimizer（HF UI・拡張版） - 生成→審査→配信（Thompson / LinUCB）→レポート - A/Aテスト・ホールドアウト、撤退基準 - CSVエクスポート、DBリセット、外部配信スタブ """) with gr.Tab("1) Generate"): with gr.Row(): campaign_id = gr.Textbox(label="campaign_id", value="cmp-demo", scale=1) k_variants = gr.Slider(1, 10, value=5, step=1, label="生成本数") value_per_conv = gr.Number(value=5000, label="value_per_conversion") brand = gr.Textbox(label="ブランド", value="SFM") product = gr.Textbox(label="商品/サービス", value="HbA1c測定アプリ") target = gr.Textbox(label="ターゲット", value="30-50代の健康意識が高い層") tone = gr.Textbox(label="トーン", value="エビデンス重視で安心感") ng_words = gr.Textbox(label="NGワード（改行区切り）", value="治る\n奇跡") btn_gen = gr.Button("広告案を生成＆審査＆保存") table_gen = gr.Dataframe(headers=GENERATE_COLUMNS, interactive=False) btn_gen.click(ui_generate, [campaign_id, brand, product, target, tone, k_variants, ng_words, value_per_conv], [table_gen]) with gr.Tab("2) Settings"): with gr.Row(): campaign_id_set = gr.Textbox(label="campaign_id", value="cmp-demo", scale=1) policy = gr.Dropdown(choices=["thompson", "linucb"], value="thompson", label="Policy") holdout = gr.Slider(0.0, 0.5, value=0.0, step=0.05, label="Holdout ratio") with gr.Row(): stop_min_impr = gr.Number(value=200, label="撤退判定の最小impressions") stop_rel_ev = gr.Slider(0.1, 0.9, value=0.5, step=0.05, label="EVの相対しきい値（劣後停止）") btn_set = gr.Button("設定を保存") msg_set = gr.Markdown() btn_set.click(ui_set_settings, [campaign_id_set, policy, holdout, stop_min_impr, stop_rel_ev], [msg_set]) with gr.Tab("3) Serve & Feedback"): with gr.Row(): campaign_id2 = gr.Textbox(label="campaign_id", value="cmp-demo", scale=1) hour = gr.Slider(0, 23, value=20, step=1, label="hour") segment = gr.Textbox(label="segment (任意)") aa_min_impr = gr.Number(value=0, label="A/Aテスト閾値（各variantのimpressionsがこの数に達するまで均等ランダム）") btn_serve = gr.Button("Serve Ad（impressionを記録）") served_vid = gr.Textbox(label="served variant_id", interactive=False) served_text = gr.Textbox(label="served text", lines=6, interactive=False) btn_serve.click(ui_serve, [campaign_id2, hour, segment, aa_min_impr], [served_vid, served_text]) with gr.Row(): btn_click = gr.Button("Clickを記録") btn_conv = gr.Button("Conversionを記録") msg = gr.Markdown() btn_click.click(lambda cid, vid, h, s: ui_feedback(cid, vid, "click", h, s), [campaign_id2, served_vid, hour, segment], [msg]) btn_conv.click(lambda cid, vid, h, s: ui_feedback(cid, vid, "conversion", h, s), [campaign_id2, served_vid, hour, segment], [msg]) gr.Markdown("### 外部配信スタブ") with gr.Row(): platform = gr.Dropdown(choices=["x", "meta", "google"], value="x", label="プラットフォーム") btn_send = gr.Button("この広告を外部に送る（スタブ）") send_res = gr.Textbox(label="送信結果", lines=3) btn_send.click(ui_adapter_send, [campaign_id2, platform, served_vid, served_text, hour, segment], [send_res]) with gr.Tab("4) Report & Ops"): campaign_id3 = gr.Textbox(label="campaign_id", value="cmp-demo") with gr.Row(): btn_rep = gr.Button("レポート更新") table_rep = gr.Dataframe(headers=REPORT_COLUMNS, interactive=False) btn_rep.click(ui_report, [campaign_id3], [table_rep]) with gr.Row(): btn_stop = gr.Button("撤退基準を適用（自動pause）") msg_stop = gr.Markdown() btn_stop.click(ui_apply_stop, [campaign_id3], [msg_stop]) gr.Markdown("#### CSVエクスポート") table_sel = gr.Dropdown(choices=["events","metrics","variants","compliance_logs","audit_logs"], value="metrics", label="テーブル") btn_exp = gr.Button("エクスポート") file_out = gr.File(label="ダウンロード", interactive=False) btn_exp.click(ui_export_csv, [campaign_id3, table_sel], [file_out]) gr.Markdown("#### 管理用：DBリセット（全削除）") confirm = gr.Textbox(label="タイプ: RESET", value="") btn_reset = gr.Button("DBを初期化") msg_reset = gr.Markdown() btn_reset.click(ui_reset_db, [confirm], [msg_reset]) if __name__ == "__main__": demo.queue().launch(server_name="0.0.0.0", server_port=7860)