Update ui/ui_app.py

ui/ui_app.py

@@ -1,18 +1,82 @@
 from __future__ import annotations
-import os, io, base64, json, traceback
+import os, io, json, base64, traceback, shutil
 from typing import List, Dict, Any
 
+# ---- 起動安定化 ----
+os.environ["GRADIO_ANALYTICS_ENABLED"] = "False"
+
 import gradio as gr
 import pandas as pd
+from pdf2image import convert_from_path
+import pdfplumber
+from openai import OpenAI
 import plotly.graph_objects as go
 
-from core.ai_client import get_client, VISION_MODEL, TEXT_MODEL
-from core.pdf_utils import has_poppler, pdf_to_images, pdf_to_text
-from core.scorer import score_company
+# ====== 🔧 Gradio API 情報ページの安全パッチ ======
+# 4.44.1 + gradio-client 1.3.0 でも、一部構成で API 情報生成時に
+# additionalProperties が bool になり例外が出ることがある。
+# 例外が出たら空の API 情報を返し、UI は正常起動させる。
+try:
+    import gradio.blocks as _grb
+    _orig_get_api_info = _grb.Blocks.get_api_info
 
-# Gradio の解析情報や分析の送信を無効化（念のため）
-os.environ["GRADIO_ANALYTICS_ENABLED"] = "False"
+    def _safe_get_api_info(self, *a, **kw):
+        try:
+            return _orig_get_api_info(self, *a, **kw)
+        except Exception:
+            return {"named_endpoints": {}, "unnamed_endpoints": []}
+
+    _grb.Blocks.get_api_info = _safe_get_api_info  # type: ignore[attr-defined]
+except Exception:
+    pass
+# ===============================================
+
+# ---- モデル設定（環境変数で上書き可）----
+OPENAI_MODEL_VISION = os.environ.get("OPENAI_VISION_MODEL", "gpt-4o-mini")
+OPENAI_MODEL_TEXT   = os.environ.get("OPENAI_TEXT_MODEL",   "gpt-4o-mini")
+
+# ---- OpenAI クライアント ----
+def _client() -> OpenAI:
+    key = os.environ.get("OPENAI_API_KEY")
+    if not key:
+        raise gr.Error("OPENAI_API_KEY が未設定です。Spaces > Settings > Variables and secrets に設定してください。")
+    # proxies を渡さない（古い httpx 互換問題の回避）
+    return OpenAI(api_key=key, timeout=30)
+
+# ---- ヘルスチェック ----
+def health() -> str:
+    msgs = []
+    msgs.append("✅ OPENAI_API_KEY: " + ("検出" if os.environ.get("OPENAI_API_KEY") else "未設定"))
+    for b in ("pdftoppm", "pdftocairo"):
+        ok = bool(shutil.which(b))
+        msgs.append(("✅" if ok else "❌") + f" {b}: " + ("検出" if ok else "未検出（packages.txt に poppler-utils が必要）"))
+    msgs.append(f"ℹ️ Vision={OPENAI_MODEL_VISION} / Text={OPENAI_MODEL_TEXT}")
+    return "<br>".join(msgs)
+
+# ---- PDF -> 画像/テキスト ----
+def pdf_to_images(pdf_path: str, dpi: int = 220, max_pages: int = 6) -> List[bytes]:
+    pages = convert_from_path(pdf_path, dpi=dpi, fmt="png")
+    out: List[bytes] = []
+    for i, p in enumerate(pages):
+        if i >= max_pages:
+            break
+        buf = io.BytesIO()
+        p.save(buf, format="PNG")
+        out.append(buf.getvalue())
+    return out
+
+def pdf_to_text(pdf_path: str, max_chars: int = 15000) -> str:
+    chunks: List[str] = []
+    with pdfplumber.open(pdf_path) as pdf:
+        for i, page in enumerate(pdf.pages):
+            t = (page.extract_text() or "").strip()
+            if t:
+                chunks.append(f"[page {i+1}]\n{t}")
+            if sum(len(c) for c in chunks) > max_chars:
+                break
+    return "\n\n".join(chunks)[:max_chars]
 
+# ---- Vision / Text 抽出 ----
 SYSTEM_JSON = """あなたは有能な財務アナリストです。
 与えられた決算書（画像またはテキスト）から、次の厳密な JSON 構造のみを日本語の単位なし・半角数値で返してください。分からない項目は null。
 {
@@ -38,7 +102,7 @@ def _b64(img: bytes) -> str:
     return base64.b64encode(img).decode("utf-8")
 
 def extract_financials(images: List[bytes] | None, text_blob: str | None, company_hint: str) -> Dict[str, Any]:
-    client = get_client()
+    client = _client()
     if images:
         content = [{"type": "text", "text": SYSTEM_JSON}]
         if company_hint:
@@ -46,9 +110,9 @@ def extract_financials(images: List[bytes] | None, text_blob: str | None, compan
         for im in images:
             content.append({"type": "input_image", "image_url": f"data:image/png;base64,{_b64(im)}"})
         resp = client.chat.completions.create(
-            model=VISION_MODEL,
+            model=OPENAI_MODEL_VISION,
             messages=[
-                {"role": "system", "content": "返答は必ず有効な JSON オブジェクトのみ。説明を含めない。"},
+                {"role": "system", "content": "返答は必ず有効な JSON オブジェクトのみ。説明文は不要。"},
                 {"role": "user", "content": content},
             ],
             response_format={"type": "json_object"},
@@ -58,7 +122,7 @@ def extract_financials(images: List[bytes] | None, text_blob: str | None, compan
     else:
         prompt = f"{SYSTEM_JSON}\n\n以下は決算書のテキストです。上記の JSON だけを返してくださ��。\n\n{text_blob or ''}"
         resp = client.chat.completions.create(
-            model=TEXT_MODEL,
+            model=OPENAI_MODEL_TEXT,
             messages=[
                 {"role": "system", "content": "返答は必ず有効な JSON オブジェクトのみ。"},
                 {"role": "user", "content": prompt},
@@ -68,6 +132,33 @@ def extract_financials(images: List[bytes] | None, text_blob: str | None, compan
         )
         return json.loads(resp.choices[0].message.content)
 
+# ---- スコアリング（ダミー実装 or 既存 scorer.py を呼び出し）----
+def score_company(fin: Dict[str, Any]) -> Dict[str, Any]:
+    # もし既存の scorer.py があるなら import して差し替えてください。
+    # ここでは安定動作用の簡易版を同梱します。
+    def g(v: Any, div: float = 1.0):
+        try:
+            return float(v) / div
+        except Exception:
+            return 0.0
+
+    is_ = fin.get("income_statement") or {}
+    bs_ = fin.get("balance_sheet") or {}
+    margin = g(is_.get("operating_income")) / (g(is_.get("sales")) + 1e-9) * 100
+    equity_ratio = g(bs_.get("total_equity")) / (g(bs_.get("total_assets")) + 1e-9) * 100
+
+    details = [
+        {"metric": "売上規模", "score": min(100, g(is_.get("sales")) ** 0.5)},
+        {"metric": "営業利益率", "score": max(0, min(100, margin + 50))},
+        {"metric": "自己資本比率", "score": max(0, min(100, equity_ratio))},
+        {"metric": "利益水準", "score": min(100, max(0, g(is_.get("operating_income")) ** 0.5 + 50))},
+        {"metric": "安全性", "score": max(0, min(100, 50 + equity_ratio / 2))},
+    ]
+    total = int(sum(d["score"] for d in details) / len(details))
+    grade = "S" if total >= 85 else "A" if total >= 70 else "B" if total >= 55 else "C"
+    return {"total_score": total, "grade": grade, "details": details}
+
+# ---- DF 変換 ----
 def fin_to_df(fin: Dict[str, Any]) -> pd.DataFrame:
     rows = []
     def add(cat, d):
@@ -90,6 +181,7 @@ def df_to_fin(df: pd.DataFrame) -> Dict[str, Any]:
             out[cat][item] = parsed
     return out
 
+# ---- 可視化 ----
 def radar(score: Dict[str, Any]) -> go.Figure:
     labels = [d["metric"] for d in score["details"]]
     values = [d["score"] for d in score["details"]]
@@ -98,25 +190,21 @@ def radar(score: Dict[str, Any]) -> go.Figure:
     fig.update_layout(
         polar=dict(radialaxis=dict(visible=True, range=[0, 100])),
         showlegend=False, margin=dict(l=20, r=20, t=30, b=20), height=380,
-        title=f"総合スコア: {score['total_score']}（グレード: {score['grade']}）"
+        title=f"総合スコア: {score['total_score']}（グレード: {score['grade']}）",
     )
     return fig
 
+# ---- ハンドラ ----
 def run_analyze(company: str, use_vision: bool, files: list[str]):
     if not files:
         raise gr.Error("PDF をアップロードしてください。")
 
-    # 画像→Vision、失敗したらテキスト→Text にフォールバック（例外は UI に吸収）
+    # 1) Vision（画像化）→ 2) テキスト抽出 の順にフォールバック
     try:
-        if use_vision:
-            if not has_poppler():
-                raise RuntimeError("Poppler が見つかりません（pdf2image が使えません）")
-            all_images: List[bytes] = []
-            for p in files:
-                all_images += pdf_to_images(p, dpi=220, max_pages=6)
-            fin = extract_financials(all_images, None, company or "")
-        else:
-            raise RuntimeError("Vision 無効")
+        all_images: List[bytes] = []
+        for p in files:
+            all_images += pdf_to_images(p, dpi=220, max_pages=6)
+        fin = extract_financials(all_images if use_vision else None, None, company or "")
     except Exception:
         text_blob = ""
         for p in files:
@@ -127,9 +215,9 @@ def run_analyze(company: str, use_vision: bool, files: list[str]):
     score = score_company(fin)
     fig = radar(score)
 
-    # AI所見（失敗しても UI は返す）
+    # AI 所見（短め）
     try:
-        client = get_client()
+        client = _client()
         prompt = f"""次の財務データとスコア結果から、箇条書きで短く日本語でコメントしてください。
 - 良い点 3つ
 - 懸念点 3つ
@@ -142,7 +230,7 @@ def run_analyze(company: str, use_vision: bool, files: list[str]):
 {json.dumps(score, ensure_ascii=False)}
 """
         resp = client.chat.completions.create(
-            model=TEXT_MODEL,
+            model=OPENAI_MODEL_TEXT,
             messages=[{"role": "system", "content": "簡潔で公正な財務アナリスト。"},
                       {"role": "user", "content": prompt}],
             temperature=0.3,
@@ -156,7 +244,7 @@ def run_analyze(company: str, use_vision: bool, files: list[str]):
         df,
         json.dumps(score, ensure_ascii=False, indent=2),
         fig,
-        insight
+        insight,
     )
 
 def run_recalc(df: pd.DataFrame):
@@ -167,36 +255,40 @@ def run_recalc(df: pd.DataFrame):
         return (
             json.dumps(score, ensure_ascii=False, indent=2),
             fig,
-            json.dumps(fin, ensure_ascii=False, indent=2)
+            json.dumps(fin, ensure_ascii=False, indent=2),
         )
     except Exception as e:
         tb = traceback.format_exc(limit=6)
         raise gr.Error(f"再計算に失敗しました: {e}\n\n<pre style='white-space:pre-wrap'>{tb}</pre>")
 
-def create_demo():
-    with gr.Blocks(theme=gr.themes.Soft(primary_hue="indigo"),
-                   fill_height=True, analytics_enabled=False) as demo:
-        gr.Markdown("## 🧮 企業スコアリング（PDF解析 × OpenAI）")
+# ---- UI ----
+def main():
+    with gr.Blocks(theme=gr.themes.Soft(primary_hue="indigo"), fill_height=True, analytics_enabled=False) as demo:
+        gr.Markdown("## 🧮 企業スコアリング（PDF解析 × OpenAI Vision）")
         with gr.Row():
             with gr.Column(scale=1):
                 company = gr.Textbox(label="企業名（任意）", placeholder="例：株式会社OO")
                 use_vision = gr.Checkbox(value=True, label="OpenAIでPDFをAI解析（Vision）")
                 files = gr.File(label="決算書PDF（複数可）", file_count="multiple", type="filepath")
-                run_btn = gr.Button("📄 解析する", variant="primary")
-                recalc_btn = gr.Button("🔁 表の値で再計算")
+                run_btn = gr.Button("📄 PDFを解析してテンプレに反映", variant="primary")
+                recalc_btn = gr.Button("🔁 この表の値で再計算")
+                health_btn = gr.Button("🩺 環境チェック")
+                health_out = gr.HTML()
+                gr.Markdown("※ 画像化やVisionに失敗した場合はテキスト抽出に自動フォールバックします。")
 
             with gr.Column(scale=1):
                 fin_json = gr.Code(label="抽出JSON", language="json", interactive=False)
 
         with gr.Tabs():
             with gr.Tab("抽出結果（表で編集可）"):
-                # ★ スキーマを明示して API 情報生成で落ちないようにする
+                # JSON schema 例外の回避：列型/列数を固定し、初期値を空にする
                 df_out = gr.Dataframe(
                     headers=["category", "item", "value"],
                     datatype=["str", "str", "number"],
                     col_count=(3, "fixed"),
                     row_count=(1, "dynamic"),
-                    interactive=True
+                    value=[],
+                    interactive=True,
                 )
             with gr.Tab("スコアリング"):
                 score_json = gr.Code(label="スコア（JSON）", language="json")
@@ -204,29 +296,16 @@ def create_demo():
             with gr.Tab("AI診断（日本語）"):
                 insight_md = gr.Markdown()
 
-        run_btn.click(
-            run_analyze,
-            inputs=[company, use_vision, files],
-            outputs=[fin_json, df_out, score_json, chart, insight_md],
-            concurrency_limit=2
-        )
-        recalc_btn.click(
-            run_recalc,
-            inputs=[df_out],
-            outputs=[score_json, chart, fin_json],
-            concurrency_limit=2
-        )
-    return demo
+        run_btn.click(run_analyze, inputs=[company, use_vision, files],
+                      outputs=[fin_json, df_out, score_json, chart, insight_md],
+                      concurrency_limit=1)
+        recalc_btn.click(run_recalc, inputs=[df_out], outputs=[score_json, chart, fin_json],
+                         concurrency_limit=1)
+        health_btn.click(health, outputs=health_out, concurrency_limit=1)
 
-# Spaces/コンテナ両対応
-demo = create_demo()
-
-def main():
-    # Spaces などローカル疎通ができない環境では share=True が必須
-    share_default = os.getenv("GRADIO_SHARE", "1")  # "1"=True / "0"=False
-    share = (share_default != "0")
-    port = int(os.getenv("PORT", "7860"))
-    demo.launch(server_name="0.0.0.0", server_port=port, share=share, max_threads=8)
+    # Spaces では share=False/サーバ設定不要。show_api=False で API 情報生成も抑制。
+    demo.queue(max_size=10)
+    demo.launch(show_api=False)
 
 if __name__ == "__main__":
     main()