from __future__ import annotations # === Writable config dirs (must come right after future import) === 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 import uuid 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_value_per_conversion, log_event ) from app.bandit import ThompsonBandit from app.forecast import SeasonalityModel from app.compliance import rule_based_check, llm_check_and_fix from app.openai_client import openai_chat_json # 初期化 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"] # JSONモード前提の厳格プロンプト 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): """LLM応答から variants 配列を安全に取り出して正規化。失敗時は None を返す。""" 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): """LLMが不調でも UI を止めないための最終フォールバック生成（簡易・無害表現）。""" base_head = [ "使いやすさで選ばれています", "日々の習慣をシンプルに", "はじめてでも安心", "続けやすいサポートを", "いま必要な機能だけを" ] base_body = [ "{brand}の「{product}」。生活になじむ設計で、今日からムリなく始められます。まずは詳細をご覧ください。", "毎日を少しラクに。{brand}の{product}が、あなたの習慣づくりを後押しします。今すぐチェック。", "難しい操作は不要。{brand}の{product}なら、使い始めから自然に続けられます。詳しくはサイトへ。", "必要な情報をひと目で。{brand}の{product}で、日々の管理をシンプルに。詳細を見る。", "続けやすさを重視。{brand}の{product}で、小さな一歩から。" ] out = [] for i in range(k): hi = base_head[i % len(base_head)] bo = base_body[i % len(base_body)].format(brand=brand, product=product) out.append({"headline": hi, "body": bo}) 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) # Slider の float を明示的に int 化 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, ) # まずは通常プロンプトで JSON モード呼び出し 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 # それでも無理ならローカル生成（UIを止めない） if not items: items = _local_variants(brand, product, k_variants) rows = [] for it in items[:k_variants]: headline = it["headline"] body = 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")) rejection_reason = None status = "approved" if not ok_rule: ok_llm, reasons, fixed = llm_check_and_fix(text) if ok_llm: text = fixed or text else: status = "rejected" rejection_reason = "; ".join(bads + reasons) else: ok_llm, reasons, fixed = llm_check_and_fix(text) if not ok_llm: text = fixed or text insert_variant(campaign_id, vid, text, status, rejection_reason) rows.append({ "variant_id": vid, "status": status, "rejection_reason": rejection_reason or "", "text": text, }) # 常に固定カラムで返す（空でもカラムを持つDataFrame） df = pd.DataFrame(rows, columns=GENERATE_COLUMNS) return df def ui_serve(campaign_id: str, hour: int, segment: str): ctx = {"hour": int(hour), "segment": (segment or "").strip() or None} m = _seasonal(campaign_id) bandit = ThompsonBandit(campaign_id) vid, _ = bandit.sample_arm(ctx, m.expected_ctr) if not vid: raise gr.Error("配信可能なバリアントがありません。まずは Generate してください。") row = get_variant(campaign_id, vid) if not row: raise gr.Error("バリアントが見つかりません。") # impression 記録 log_event(campaign_id, vid, "impression", datetime.utcnow().isoformat(), None) ThompsonBandit.update_with_event(campaign_id, vid, "impression") return vid, row["text"] def ui_feedback(campaign_id: str, variant_id: str, event_type: str): if not variant_id: raise gr.Error("先に Serve してください。") log_event(campaign_id, variant_id, event_type, datetime.utcnow().isoformat(), None) ThompsonBandit.update_with_event(campaign_id, variant_id, 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), }) # 常に固定カラムで返す（空でもカラムを持つDataFrame） df = pd.DataFrame(rows, columns=REPORT_COLUMNS) return df def ui_check(text: str): ok_rule, bads = rule_based_check(text, []) ok_llm, reasons, fixed = llm_check_and_fix(text) status = "pass" if (ok_rule and ok_llm) else "needs_fix" fixed_text = fixed or (text if status == "pass" else "") reasons_joined = "; ".join(bads + reasons) return status, reasons_joined, fixed_text with gr.Blocks(title="AdCopy MAB Optimizer", fill_height=True) as demo: gr.Markdown(""" # AdCopy MAB Optimizer（HF UI） **広告コピー自動生成 → Thompson Sampling（CTR×CVR） → レポート** を、Hugging Face Spaces 上で完結。 - LLM: OpenAI （`OPENAI_API_KEY` を Space Secrets に設定） - DB: SQLite（`/data/app_data/data.db` など、書き込み可能ディレクトリ） - 季節性: Prophet/NeuralProphet（なければ簡易ヒューリスティック） """) 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) 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 (任意)") 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], [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: ui_feedback(cid, vid, "click"), [campaign_id2, served_vid], [msg]) btn_conv.click(lambda cid, vid: ui_feedback(cid, vid, "conversion"), [campaign_id2, served_vid], [msg]) with gr.Tab("3) Report"): campaign_id3 = gr.Textbox(label="campaign_id", value="cmp-demo") btn_rep = gr.Button("更新") table_rep = gr.Dataframe(headers=REPORT_COLUMNS, interactive=False) btn_rep.click(ui_report, [campaign_id3], [table_rep]) with gr.Tab("4) Compliance Check"): cand = gr.Textbox(label="チェックする文面", lines=5) btn_chk = gr.Button("判定") status = gr.Textbox(label="status") reasons = gr.Textbox(label="reasons") fixed = gr.Textbox(label="fixed (修正案)", lines=5) btn_chk.click(ui_check, [cand], [status, reasons, fixed]) if __name__ == "__main__": demo.queue().launch(server_name="0.0.0.0", server_port=7860)