Upload 2 files

app.py

@@ -0,0 +1,413 @@
+# app.py
+# Hugging Face Spaces-ready Gradio + FastAPI app
+# 機能: クリエイティブ自動生成（OpenAI）、ブランドガイド従属性チェック、A/Bテスト（CTR/CVR）、多腕バンディット（ε-greedy）
+# 追跡: /v/<variant> で表示、/click と /convert でクリック/コンバージョン計測。/r はローテータ（割当）URL。
+# メモリ: SQLite（./ab.db）
+# デプロイ: 「Spaces（Gradio）」で動作。環境変数 OPENAI_API_KEY を設定してください。
+
+import os
+import sqlite3
+import json
+import time
+import random
+import hashlib
+from datetime import datetime
+from typing import List, Dict, Any, Optional
+from urllib.parse import quote, unquote
+
+import gradio as gr
+from fastapi import FastAPI, Request
+from fastapi.responses import HTMLResponse, RedirectResponse, PlainTextResponse
+from openai import OpenAI
+
+DB_PATH = "ab.db"
+TABLE_SQL = """
+CREATE TABLE IF NOT EXISTS variants (
+    id TEXT PRIMARY KEY,
+    name TEXT,
+    channel TEXT,
+    target_metric TEXT,
+    kpi_target REAL,
+    brand_guidelines TEXT,
+    banned_words TEXT,
+    product TEXT,
+    audience TEXT,
+    cta_url TEXT,
+    headline TEXT,
+    body TEXT,
+    cta TEXT,
+    image_prompt TEXT,
+    created_at INTEGER,
+    impressions INTEGER DEFAULT 0,
+    clicks INTEGER DEFAULT 0,
+    conversions INTEGER DEFAULT 0
+);
+
+CREATE TABLE IF NOT EXISTS weights (
+    variant_id TEXT PRIMARY KEY,
+    weight REAL
+);
+"""
+
+def db():
+    conn = sqlite3.connect(DB_PATH, check_same_thread=False)
+    conn.row_factory = sqlite3.Row
+    return conn
+
+def init_db():
+    conn = db()
+    cur = conn.cursor()
+    for stmt in TABLE_SQL.strip().split(";"):
+        s = stmt.strip()
+        if s:
+            cur.execute(s)
+    conn.commit()
+    conn.close()
+
+def now_ts() -> int:
+    return int(time.time())
+
+def gen_variant_id(seed: str) -> str:
+    return hashlib.sha256(f"{seed}-{time.time()}-{random.random()}".encode()).hexdigest()[:12]
+
+def upsert_weight(conn, vid: str, w: float):
+    cur = conn.cursor()
+    cur.execute("INSERT INTO weights(variant_id, weight) VALUES(?, ?) ON CONFLICT(variant_id) DO UPDATE SET weight=excluded.weight", (vid, w))
+    conn.commit()
+
+def get_weights(conn):
+    rows = conn.execute("SELECT variant_id, weight FROM weights").fetchall()
+    return {r["variant_id"]: r["weight"] for r in rows}
+
+def epsilon_greedy_pick(conn, epsilon: float = 0.1) -> Optional[str]:
+    # CTR ベースのε-greedy
+    rows = conn.execute("SELECT id, impressions, clicks FROM variants").fetchall()
+    if not rows:
+        return None
+    # 探索
+    if random.random() < epsilon:
+        return random.choice(rows)["id"]
+    # 活用: CTR最大
+    best_vid = None
+    best_ctr = -1
+    for r in rows:
+        imps = r["impressions"]
+        clk = r["clicks"]
+        ctr = (clk / imps) if imps > 0 else 0.0
+        if ctr > best_ctr:
+            best_ctr = ctr
+            best_vid = r["id"]
+    return best_vid or rows[0]["id"]
+
+def record_impression(conn, vid: str):
+    conn.execute("UPDATE variants SET impressions = impressions + 1 WHERE id=?", (vid,))
+    conn.commit()
+
+def record_click(conn, vid: str):
+    conn.execute("UPDATE variants SET clicks = clicks + 1 WHERE id=?", (vid,))
+    conn.commit()
+
+def record_conversion(conn, vid: str):
+    conn.execute("UPDATE variants SET conversions = conversions + 1 WHERE id=?", (vid,))
+    conn.commit()
+
+def fetch_variant(conn, vid: str):
+    row = conn.execute("SELECT * FROM variants WHERE id=?", (vid,)).fetchone()
+    return row
+
+def list_variants(conn):
+    rows = conn.execute("SELECT * FROM variants ORDER BY created_at DESC").fetchall()
+    return rows
+
+def delete_all(conn):
+    conn.execute("DELETE FROM variants")
+    conn.execute("DELETE FROM weights")
+    conn.commit()
+
+# ---- Brand compliance (ローカル簡易チェック) ----
+def check_compliance(text: str, guidelines: str, banned_words: List[str]) -> Dict[str, Any]:
+    violations = []
+    lw = text.lower()
+    for w in banned_words:
+        ww = w.strip().lower()
+        if ww and ww in lw:
+            violations.append(f"banned_word:{w}")
+    # 文字数の過不足などの軽いヒューリスティック
+    if len(text) < 10:
+        violations.append("too_short")
+    if len(text) > 1200:
+        violations.append("too_long")
+    ok = len(violations) == 0
+    return {"ok": ok, "violations": violations}
+
+# ---- OpenAIでクリエイティブ生成 ----
+def openai_client():
+    api_key = os.getenv("OPENAI_API_KEY")
+    if not api_key:
+        raise RuntimeError("OPENAI_API_KEY が未設定です（SpacesのSecretsに設定してください）。")
+    return OpenAI(api_key=api_key)
+
+GEN_SYS_PROMPT = """You are a world-class performance marketer and copywriter.
+You must output STRICT JSON encoded in UTF-8 with this schema:
+{
+  "variants": [
+    {
+      "name": "string short nickname",
+      "headline": "string 60 chars max",
+      "body": "string 280-600 chars",
+      "cta": "string <= 30 chars",
+      "image_prompt": "string concise visual prompt for an image generator"
+    }, ...
+  ]
+}
+Rules:
+- Obey brand guidelines: tone, banned phrases, do-not-claim, legal constraints.
+- Optimize for the stated goal (CTR or CVR) and KPI target.
+- Avoid spammy claims, avoid ALL CAPS.
+- Keep culturally appropriate Japanese copy if input is Japanese.
+"""
+
+def generate_variants_with_openai(goal: str, kpi_target: float, channel: str, product: str, audience: str, brand_guidelines: str, banned_words: List[str], n_variants: int = 3) -> List[Dict[str, str]]:
+    client = openai_client()
+    user_prompt = {
+        "goal": goal,
+        "kpi_target": kpi_target,
+        "channel": channel,
+        "product": product,
+        "audience": audience,
+        "brand_guidelines": brand_guidelines,
+        "banned_words": banned_words,
+        "n_variants": n_variants
+    }
+    # Chat Completions (安定): モデルは適宜変更可
+    resp = client.chat.completions.create(
+        model="gpt-4o-mini",
+        temperature=0.8,
+        messages=[
+            {"role": "system", "content": GEN_SYS_PROMPT},
+            {"role": "user", "content": json.dumps(user_prompt, ensure_ascii=False)}
+        ]
+    )
+    content = resp.choices[0].message.content.strip()
+    # JSON抽出
+    try:
+        data = json.loads(content)
+        variants = data.get("variants", [])
+    except Exception:
+        # 失敗時フォールバック: JSONのコードブロックを剥がす
+        import re
+        m = re.search(r"\{[\s\S]*\}", content)
+        if not m:
+            raise RuntimeError("モデル出力のJSON解析に失敗しました。出力:\n" + content[:500])
+        data = json.loads(m.group(0))
+        variants = data.get("variants", [])
+    # banned_words 簡易フィルタ
+    filtered = []
+    for v in variants[:n_variants]:
+        body = v.get("body", "")
+        head = v.get("headline", "")
+        cta = v.get("cta", "")
+        comp = "\n".join([head, body, cta])
+        compliance = check_compliance(comp, brand_guidelines, banned_words)
+        if compliance["ok"]:
+            filtered.append(v)
+        else:
+            # 軽微な違反（長さなど）は通す。禁止語のみ除外。
+            if not any(vi.startswith("banned_word") for vi in compliance["violations"]):
+                filtered.append(v)
+    if not filtered:
+        filtered = variants[:n_variants]
+    return filtered[:n_variants]
+
+# ---- FastAPI ルーティング（追跡/配信） ----
+fastapi_app = FastAPI()
+
+@fastapi_app.get("/")
+def root():
+    return PlainTextResponse("This Space hosts the Campaign Generator + A/B tester. Visit /gradio for UI.")
+
+@fastapi_app.get("/r")
+def rotator():
+    conn = db()
+    try:
+        vid = epsilon_greedy_pick(conn, epsilon=0.1)
+        if not vid:
+            return PlainTextResponse("No variants yet.", status_code=404)
+        return RedirectResponse(url=f"/v/{vid}")
+    finally:
+        conn.close()
+
+@fastapi_app.get("/v/{vid}")
+def serve_variant(vid: str):
+    conn = db()
+    try:
+        row = fetch_variant(conn, vid)
+        if not row:
+            return PlainTextResponse("Variant not found", status_code=404)
+        record_impression(conn, vid)
+        headline = row["headline"]
+        body = row["body"]
+        cta = row["cta"]
+        cta_url = row["cta_url"] or "/convert?vid=" + vid
+        html = f"""
+        <html>
+        <head>
+            <meta charset="utf-8" />
+            <meta name="viewport" content="width=device-width, initial-scale=1" />
+            <title>{headline}</title>
+            <style>
+              body {{ font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, 'Noto Sans JP', sans-serif; padding: 24px; line-height: 1.6; }}
+              .card {{ max-width: 720px; margin: 0 auto; border: 1px solid #eee; border-radius: 16px; padding: 24px; box-shadow: 0 2px 12px rgba(0,0,0,.04); }}
+              .headline {{ font-size: 1.8rem; font-weight: 700; margin-bottom: 12px; }}
+              .body {{ font-size: 1rem; white-space: pre-wrap; }}
+              .cta {{ display: inline-block; margin-top: 20px; padding: 12px 18px; border-radius: 999px; text-decoration: none; background: black; color: white; }}
+              .meta {{ margin-top: 16px; color: #777; font-size: .85rem; }}
+            </style>
+        </head>
+        <body>
+          <div class="card">
+            <div class="headline">{headline}</div>
+            <div class="body">{body}</div>
+            <a class="cta" href="/click?vid={vid}&to={quote(cta_url, safe='')}" rel="nofollow">{cta}</a>
+            <div class="meta">Variant ID: {vid}</div>
+          </div>
+        </body>
+        </html>
+        """
+        return HTMLResponse(html)
+    finally:
+        conn.close()
+
+@fastapi_app.get("/click")
+def click(vid: str, to: str = "/convert"):
+    conn = db()
+    try:
+        row = fetch_variant(conn, vid)
+        if not row:
+            return PlainTextResponse("Variant not found", status_code=404)
+        record_click(conn, vid)
+        # 安全のため内部相対URLのみ許可（外部リダイレクトを避ける）
+        target = unquote(to)
+        if not target.startswith("/"):
+            target = "/convert?vid=" + vid
+        return RedirectResponse(url=target)
+    finally:
+        conn.close()
+
+@fastapi_app.get("/convert")
+def convert(vid: str = ""):
+    if not vid:
+        return PlainTextResponse("Missing vid", status_code=400)
+    conn = db()
+    try:
+        row = fetch_variant(conn, vid)
+        if not row:
+            return PlainTextResponse("Variant not found", status_code=404)
+        record_conversion(conn, vid)
+        html = f"""
+        <html><head><meta charset="utf-8"><title>Thank you</title></head>
+        <body style="font-family: -apple-system,BlinkMacSystemFont,'Segoe UI',Roboto,'Noto Sans JP',sans-serif;padding:24px;">
+          <h1>Thanks!</h1>
+          <p>Conversion recorded for Variant {vid}.</p>
+          <p><a href="/v/{vid}">Back</a> | <a href="/r">Rotator</a></p>
+        </body></html>
+        """
+        return HTMLResponse(html)
+    finally:
+        conn.close()
+
+# ---- Gradio UI ----
+def create_rows_table() -> str:
+    conn = db()
+    try:
+        rows = list_variants(conn)
+        if not rows:
+            return "まだバリアントがありません。"
+        # Markdownテーブル
+        header = "|Variant|Channel|Impr|Clicks|Conv|CTR|CVR|Headline|\n|---|---:|---:|---:|---:|---:|---:|---|\n"
+        lines = []
+        for r in rows:
+            imps = r["impressions"]
+            clk = r["clicks"]
+            conv = r["conversions"]
+            ctr = (clk / imps) if imps > 0 else 0.0
+            cvr = (conv / clk) if clk > 0 else 0.0
+            link_v = f"/v/{r['id']}"
+            lines.append(f"|`{r['id']}`|{r['channel']}|{imps}|{clk}|{conv}|{ctr:.3f}|{cvr:.3f}|[{r['headline']}]({link_v})|")
+        return header + "\n".join(lines)
+    finally:
+        conn.close()
+
+def generate_and_save(goal, kpi_target, channel, product, audience, brand_guidelines, banned_words_csv, n_variants, cta_url):
+    banned_words = [w.strip() for w in (banned_words_csv or "").split(",") if w.strip()]
+    variants = generate_variants_with_openai(goal, float(kpi_target or 0), channel, product, audience, brand_guidelines, banned_words, int(n_variants or 3))
+    conn = db()
+    try:
+        for v in variants:
+            vid = gen_variant_id(v.get("name","variant"))
+            conn.execute(
+                """INSERT INTO variants(id, name, channel, target_metric, kpi_target, brand_guidelines, banned_words, product, audience, cta_url, headline, body, cta, image_prompt, created_at)
+                   VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
+                (vid, v.get("name","variant"), channel, goal, float(kpi_target or 0), brand_guidelines, ",".join(banned_words),
+                 product, audience, cta_url, v.get("headline",""), v.get("body",""), v.get("cta",""), v.get("image_prompt",""),
+                 now_ts())
+            )
+            # 初期重みを均等に
+            upsert_weight(conn, vid, 1.0)
+        table_md = create_rows_table()
+        info = {
+            "rotator_url": "/r",
+            "variant_urls": [f"/v/{row['id']}" for row in list_variants(conn)]
+        }
+        return json.dumps(info, ensure_ascii=False, indent=2), table_md
+    finally:
+        conn.commit()
+        conn.close()
+
+def reset_all():
+    conn = db()
+    try:
+        delete_all(conn)
+        return "全バリアントを削除しました。", create_rows_table()
+    finally:
+        conn.close()
+
+def metrics_report():
+    return create_rows_table()
+
+with gr.Blocks(title="キャンペーン生成器＋自動ABテスト") as gradio_app:
+    gr.Markdown("# キャンペーン生成器＋自動A/Bテスト（SNS/LP）\nOpenAIでクリエイティブ案を生成し、Space内で配信・計測（CTR/CVR）・最適化（ε-greedy）を行います。")
+    with gr.Row():
+        with gr.Column():
+            goal = gr.Dropdown(choices=["CTR","CVR"], label="目標指標", value="CTR")
+            kpi_target = gr.Number(label="KPI目標値（%表記は小数に：0.12など）", value=0.05)
+            channel = gr.Dropdown(choices=["SNS","LP"], label="チャネル", value="SNS")
+            product = gr.Textbox(label="商品・訴求ポイント", lines=3, placeholder="例：HbA1c測定アプリ。無料トライアルあり。")
+            audience = gr.Textbox(label="想定ターゲット", lines=2, placeholder="例：20-40代、健康意識の高いビジネスパーソン")
+            brand_guidelines = gr.Textbox(label="ブランドガイド（トーン/禁止表現/法務注意など）", lines=6, placeholder="例：医療効果を断定しない。誇大広告NG。丁寧で安心感のあるトーン。")
+            banned_words_csv = gr.Textbox(label="禁止ワード（カンマ区切り）", placeholder="例：無料、100%改善、絶対 など")
+            n_variants = gr.Slider(label="生成する案数", minimum=1, maximum=6, step=1, value=3)
+            cta_url = gr.Textbox(label="CTAリンク（相対パス推奨。未指定なら /convert?vid=...）", placeholder="/convert?vid=<自動付与> もしくは /thanks")
+            btn_gen = gr.Button("生成 → 反映")
+            btn_reset = gr.Button("全データ削除")
+        with gr.Column():
+            out_info = gr.Code(label="配信用URL（JSON）")
+            out_table = gr.Markdown(create_rows_table())
+
+    btn_gen.click(fn=generate_and_save, inputs=[goal, kpi_target, channel, product, audience, brand_guidelines, banned_words_csv, n_variants, cta_url], outputs=[out_info, out_table])
+    btn_reset.click(fn=reset_all, inputs=None, outputs=[out_info, out_table])
+
+    gr.Markdown("## メトリクス\n下記URLをシェアしてトラフィックを集めると、CTR/CVRが更新されます。/r はローテータ（多腕バンディット）です。")
+    rep_btn = gr.Button("最新レポートに更新")
+    rep_out = gr.Markdown(create_rows_table())
+    rep_btn.click(fn=metrics_report, inputs=None, outputs=rep_out)
+
+# ---- 起動準備 ----
+init_db()
+
+# FastAPI ←→ Gradio のマウント
+# Spacesのランタイムでは app という変数をエントリーポイントにするのが一般的。
+app = fastapi_app
+# Gradioを /gradio パスにマウント
+from gradio.routes import mount_gradio_app
+app = mount_gradio_app(app, gradio_app, path="/gradio")

requirements.txt

@@ -0,0 +1,7 @@
+gradio>=4.39.0
+fastapi>=0.115.0
+uvicorn>=0.30.0
+openai>=1.42.0
+pydantic>=2.7.0
+pandas>=2.2.2
+numpy>=1.26.4