import os import json import math from typing import Any, Dict, List, Tuple import gradio as gr # OpenAI SDK（v1系） try: from openai import OpenAI except Exception: OpenAI = None # ---------------------------- # 設定 # ---------------------------- DEFAULT_MODEL = os.getenv("OPENAI_MODEL", "gpt-4o-mini") # 必要ならSpacesのEnvで差し替え DB_PATH = os.getenv("EVENTS_DB_PATH", "events_db.json") # ---------------------------- # プロンプト（prompts.py があれば優先） # ---------------------------- PROMPT_PRE_DEFAULT = """\ あなたはイベント運営の専門コンサルです。 ユーザーのイベント要件と、参照用の類似イベント情報を根拠として、 「従来より失敗しにくい設計案」を短時間で作るための提案を作成してください。 # ユーザー要件 {requirements_json} # 類似イベント（参照データ） {similar_events_json} # 出力要件（JSONのみ） 次のJSONスキーマに厳密に従って出力してください。余計な文章は禁止です。 {{ "risks": [ {{"title": "失敗しやすいポイント（短い）", "why": "理由（具体的）", "mitigation": "対策（具体的）"}} ], "budget_plan": [ {{"item": "費目", "amount_yen": 0, "reason": "根拠（類似イベント/要件に基づく）"}} ], "schedule": [ {{"phase": "フェーズ名", "when": "時期/当日時間帯", "owner": "担当(例: 主催/警備/受付)", "notes": "注意点"}} ], "assumptions": [ {{"key": "前提", "value": "値"}} ] }} 制約: - risksは最低5件 - budget_planは最低6費目（人件費/警備/設備/広報/予備費/その他など） - scheduleは「準備（数週間前〜前日）」「当日（時間帯）」「撤収」の流れがわかるように最低10行 - amount_yenは整数 """ PROMPT_POST_DEFAULT = """\ あなたはイベント運営の専門コンサルです。 事前計画と事後実績の差分から、改善点と成果物（成果レポート/引継ぎ資料）を作成してください。 # 事前計画（JSON） {pre_plan_json} # 事後実績（ユーザー入力） {post_actual_json} # 出力要件（JSONのみ） 次のJSONスキーマに厳密に従って出力してください。余計な文章は禁止です。 {{ "diff": [ {{"kpi": "指標名", "plan": "計画値/想定", "actual": "実績", "impact": "影響（短く）"}} ], "improvements": [ {{"issue": "課題", "cause": "原因", "next_action": "次回アクション（具体）"}} ], "report": {{ "summary": "200〜400字で要約", "highlights": ["良かった点1", "良かった点2", "良かった点3"], "numbers": [ {{"name": "来場者数", "value": "xxx"}}, {{"name": "ピーク時間帯", "value": "xx:xx-xx:xx"}} ], "risks_observed": ["当日発生したリスク1", "リスク2"] }}, "handover": {{ "checklist": ["次回チェック項目1", "2", "3", "4", "5"], "next_time": ["次回こうする1", "2", "3"] }} }} 制約: - diffは最低4件 - improvementsは最低5件 - report.highlightsは最低3件 - handover.checklistは最低7件 """ try: # 同一ディレクトリに prompts.py がある場合は優先 from prompts import PROMPT_PRE as PROMPT_PRE_FILE, PROMPT_POST as PROMPT_POST_FILE # type: ignore PROMPT_PRE = PROMPT_PRE_FILE PROMPT_POST = PROMPT_POST_FILE except Exception: PROMPT_PRE = PROMPT_PRE_DEFAULT PROMPT_POST = PROMPT_POST_DEFAULT # ---------------------------- # ユーティリティ # ---------------------------- def load_db() -> List[Dict[str, Any]]: with open(DB_PATH, "r", encoding="utf-8") as f: return json.load(f) def safe_json(obj: Any) -> str: return json.dumps(obj, ensure_ascii=False, indent=2) def score_similarity(req: Dict[str, Any], ev: Dict[str, Any]) -> float: """ 最短用：ルール＋簡易スコア """ score = 0.0 # 一致系 if req["venue_type"] == ev.get("venue_type"): score += 3.0 if req["event_type"] == ev.get("event_type"): score += 3.0 # 来場者差（小さいほど加点） rv = req["visitors"] evv = ev.get("visitors", 0) if rv and evv: diff = abs(rv - evv) score += max(0.0, 5.0 - (diff / 500.0)) # 500人差で-1点相当 # 予算差（小さいほど加点） rb = req["budget_yen"] eb = ev.get("budget_yen", 0) if rb and eb: diff = abs(rb - eb) score += max(0.0, 5.0 - (diff / 200000.0)) # 20万円差で-1点相当 # 目的（重なり） goals = set(req.get("goals", [])) ev_goals = set(ev.get("goals", [])) if goals and ev_goals: score += 0.8 * len(goals.intersection(ev_goals)) return score def top_k_similar(req: Dict[str, Any], db: List[Dict[str, Any]], k: int = 3) -> List[Dict[str, Any]]: scored = [(score_similarity(req, ev), ev) for ev in db] scored.sort(key=lambda x: x[0], reverse=True) return [ev for _, ev in scored[:k]] def resolve_api_key(api_key_ui: str = "") -> str: """ 1) Secrets（OPENAI_API_KEY）優先 2) 空ならUI入力を使う """ k = (os.getenv("OPENAI_API_KEY", "") or "").strip() if k: return k return (api_key_ui or "").strip() def call_openai_json(prompt: str, api_key_ui: str = "") -> Tuple[bool, str]: """ OpenAIに投げて「JSON文字列」を返す（失敗時はテキスト） """ api_key = resolve_api_key(api_key_ui=api_key_ui) if not api_key: return False, "OPENAI_API_KEY が設定されていません。SpacesのSecretsに追加するか、画面上のAPIキー欄に入力してください。" if OpenAI is None: return False, "openai パッケージが見つかりません。requirements.txt を確認してください。" client = OpenAI(api_key=api_key) # Responses API優先。失敗したらchat.completionsにフォールバック try: resp = client.responses.create( model=DEFAULT_MODEL, input=prompt, ) text = resp.output_text return True, text except Exception: try: resp = client.chat.completions.create( model=DEFAULT_MODEL, messages=[ {"role": "system", "content": "You must output ONLY valid JSON."}, {"role": "user", "content": prompt}, ], ) text = resp.choices[0].message.content return True, text except Exception as e: return False, f"OpenAI呼び出しでエラー: {e}" def extract_json(text: str) -> Tuple[bool, Any]: """ モデルが余計な文章を混ぜても、最初の{...}を拾ってJSONとして解釈する簡易処理 """ if not text: return False, None t = text.strip() # すでにJSONっぽい場合 try: return True, json.loads(t) except Exception: pass # 先頭の { から最後の } までを抜く start = t.find("{") end = t.rfind("}") if start >= 0 and end > start: candidate = t[start : end + 1] try: return True, json.loads(candidate) except Exception: return False, candidate return False, t def yen_from_range(budget_range: str) -> int: mapping = { "〜50万円": 500000, "〜100万円": 1000000, "〜200万円": 2000000, "〜300万円": 3000000, "〜500万円": 5000000, "500万円以上": 7000000, } return mapping.get(budget_range, 1000000) # ---------------------------- # 事前：生成 # ---------------------------- def pre_generate(event_type, venue_type, visitors_range, budget_range, goals, constraints_text, api_key_ui, state: Dict[str, Any]): db = load_db() # 来場者レンジ → 代表値 vr_map = { "〜300人": 300, "〜1,000人": 1000, "〜3,000人": 3000, "〜5,000人": 5000, "〜10,000人": 10000, "10,000人以上": 15000, } visitors = vr_map.get(visitors_range, 1000) budget_yen = yen_from_range(budget_range) req = { "event_type": event_type, "venue_type": venue_type, "visitors": visitors, "budget_yen": budget_yen, "goals": goals or [], "constraints": constraints_text.strip() if constraints_text else "", } similars = top_k_similar(req, db, k=3) prompt = PROMPT_PRE.format( requirements_json=safe_json(req), similar_events_json=safe_json(similars), ) ok, text = call_openai_json(prompt, api_key_ui=api_key_ui) if not ok: return safe_json(similars), "", f"エラー: {text}", state ok2, parsed = extract_json(text) if not ok2: # JSONとして読めない場合は生のテキストを返す plan_json_str = text plan_view = f"JSONとして解釈できませんでした。生出力を表示します。\n\n{plan_json_str}" else: plan_json_str = safe_json(parsed) plan_view = plan_json_str # stateに保持（同一セッション内で事後に渡す） state = state or {} state["pre_requirements"] = req state["pre_similars"] = similars state["pre_plan_raw"] = plan_json_str return safe_json(similars), plan_json_str, plan_view, state # ---------------------------- # 事後：比較＆生成 # ---------------------------- def post_generate(actual_visitors, peak_time, troubles, satisfaction, notes, pre_plan_json_text, api_key_ui, state: Dict[str, Any]): # 事前計画：優先順 # 1) ユーザーがテキストに貼ったJSON # 2) stateに保存された最新の事前計画 pre_plan_text = (pre_plan_json_text or "").strip() if not pre_plan_text: pre_plan_text = (state or {}).get("pre_plan_raw", "").strip() if not pre_plan_text: return "", "事前計画がありません。先に「事前」タブで生成するか、事前計画JSONを貼り付けてください。", state # 事後実績 post_actual = { "actual_visitors": int(actual_visitors) if actual_visitors else None, "peak_time": peak_time.strip() if peak_time else "", "troubles": troubles or [], "satisfaction_1to5": satisfaction, "notes": (notes or "").strip(), } prompt = PROMPT_POST.format( pre_plan_json=pre_plan_text, post_actual_json=safe_json(post_actual), ) ok, text = call_openai_json(prompt, api_key_ui=api_key_ui) if not ok: return "", f"エラー: {text}", state ok2, parsed = extract_json(text) if not ok2: out = f"JSONとして解釈できませんでした。生出力を表示します。\n\n{text}" return "", out, state out_json = safe_json(parsed) return out_json, out_json, state # ---------------------------- # UI # ---------------------------- with gr.Blocks(title="EventX 最短デモ（事前/事後）") as demo: gr.Markdown( "## EventX 最短デモ（事前/事後 切り替え）\n" "- 可能ならAPIキーはSpacesのSecrets（OPENAI_API_KEY）に設定してください。\n" "- **フォールバック**：Secretsが効かない場合のみ、下の欄にAPIキーを入力して動かせます（公開Spaceでは注意）。" ) # フォールバック用：UIでAPIキーを入力（値は表示されない） api_key_input = gr.Textbox( label="OpenAI API Key（フォールバック用）", type="password", placeholder="sk-...（Secretsが効かないときだけ使用）", ) state = gr.State({}) with gr.Tabs(): # ---------------------------- # 事前タブ # ---------------------------- with gr.TabItem("事前：設計案生成"): gr.Markdown("### 1) 要件入力 → 類似イベント → 失敗ポイント/予算/スケジュール生成") with gr.Row(): event_type = gr.Dropdown( label="イベント種別", choices=["マルシェ", "学園祭", "地域祭り", "フードフェス", "展示会", "スポーツ大会", "その他"], value="マルシェ", ) venue_type = gr.Dropdown( label="場所タイプ", choices=["屋外", "屋内", "複数拠点", "駐車場分散"], value="屋外", ) with gr.Row(): visitors_range = gr.Dropdown( label="想定来場者数", choices=["〜300人", "〜1,000人", "〜3,000人", "〜5,000人", "〜10,000人", "10,000人以上"], value="〜1,000人", ) budget_range = gr.Dropdown( label="予算レンジ", choices=["〜50万円", "〜100万円", "〜200万円", "〜300万円", "〜500万円", "500万円以上"], value="〜100万円", ) goals = gr.CheckboxGroup( label="目的（複数選択）", choices=["集客", "売上", "地域PR", "安全重視", "回遊促進", "満足度向上"], value=["集客", "安全重視"], ) constraints_text = gr.Textbox( label="制約・前提（任意）", placeholder="例：雨天時は体育館へ移動、音出しNG、駐車場200台、受付は1箇所…", lines=3, ) run_pre = gr.Button("設計案を生成", variant="primary") gr.Markdown("### 類似イベント（上位3件）") similar_view = gr.Code(language="json") gr.Markdown("### 事前計画JSON（事後タブにそのまま貼れます）") pre_plan_json_text = gr.Textbox(lines=8, label="事前計画JSON（保存用）") gr.Markdown("### 生成結果（表示用）") pre_plan_view = gr.Code(language="json") run_pre.click( fn=pre_generate, inputs=[event_type, venue_type, visitors_range, budget_range, goals, constraints_text, api_key_input, state], outputs=[similar_view, pre_plan_json_text, pre_plan_view, state], ) # ---------------------------- # 事後タブ # ---------------------------- with gr.TabItem("事後：振り返り・改善"): gr.Markdown("### 2) 実績入力 → 計画との差分 → 改善点 → 成果レポート/引継ぎ") gr.Markdown("#### 事前計画JSON（空なら、事前タブで生成した最新を自動で使います）") pre_plan_paste = gr.Textbox( lines=6, label="事前計画JSON（貼り付け任意）", placeholder="事前タブのJSONを貼ると、別セッションでも動きます", ) with gr.Row(): actual_visitors = gr.Number(label="実来場者数（任意）", value=850, precision=0) peak_time = gr.Textbox(label="ピーク時間帯（例 12:00-13:00）", value="12:00-13:00") troubles = gr.CheckboxGroup( label="発生トラブル（複数選択）", choices=[ "受付混雑", "駐車場待ち", "行列過多", "雨天対応", "迷子/案内不足", "回線不調", "備品不足", "スタッフ不足", "クレーム対応", ], value=["受付混雑"], ) satisfaction = gr.Slider(label="主観評価（1〜5）", minimum=1, maximum=5, step=1, value=3) notes = gr.Textbox(label="自由メモ（任意）", lines=3, placeholder="例：12時に急に人が増えた、誘導が足りなかった、など") run_post = gr.Button("差分＆レポートを生成", variant="primary") gr.Markdown("### 生成結果（JSON）") post_out_json = gr.Code(language="json") post_msg = gr.Textbox(label="メッセージ", lines=2) run_post.click( fn=post_generate, inputs=[actual_visitors, peak_time, troubles, satisfaction, notes, pre_plan_paste, api_key_input, state], outputs=[post_out_json, post_msg, state], ) demo.launch()