Update ui/ui_app.py

ui/ui_app.py

@@ -1,234 +1,251 @@
 # ui/ui_app.py
 from __future__ import annotations
-import os, io, json, base64, traceback
+import os, json, io
+from typing import Any, Dict, List, Tuple
+
 import gradio as gr
 import pandas as pd
 import plotly.graph_objects as go
 
-from core.extract import parse_pdf, ExtractError
-from core.scoring import score_company
-from core.external_scoring import score_external_from_df
-from core.ai_judgement import make_ai_memo
-
-# ================= 共通ユーティリティ =================
-
-def _b64(img_bytes):
-    return base64.b64encode(img_bytes).decode("utf-8")
-
-def fin_to_df(fin):
-    rows = []
-    def add(cat, d):
-        for k, v in (d or {}).items():
-            rows.append({"category": cat, "item": k, "value": v})
-    add("balance_sheet", fin.get("balance_sheet"))
-    add("income_statement", fin.get("income_statement"))
-    add("cash_flows", fin.get("cash_flows"))
-    return pd.DataFrame(rows, columns=["category", "item", "value"])
-
-def df_to_fin(df):
-    out = {"balance_sheet": {}, "income_statement": {}, "cash_flows": {}}
-    for _, r in df.iterrows():
-        cat, item, val = str(r["category"]), str(r["item"]), r["value"]
-        try:
-            parsed = None if val in (None, "", "null") else float(str(val).replace(",",""))
-        except Exception:
-            parsed = None
-        if cat in out:
-            out[cat][item] = parsed
-    return out
-
-def radar(score):
-    labels = [d["metric"] for d in score["details"]]
-    values = [d["score"] for d in score["details"]]
-    fig = go.Figure()
-    fig.add_trace(go.Scatterpolar(r=values + values[:1], theta=labels + labels[:1], fill="toself"))
-    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']}）"
-    )
-    return fig
-
-# ================ OpenAI 抽出（Vision / Text） =================
-
-OPENAI_MODEL_VISION = os.environ.get("OPENAI_VISION_MODEL", "gpt-4o-mini")
-OPENAI_MODEL_TEXT   = os.environ.get("OPENAI_TEXT_MODEL",   "gpt-4o-mini")
-
-SYSTEM_JSON = """あなたは有能な財務アナリストです。
-与えられた決算書（画像またはテキスト）から、次の厳密な JSON 構造のみを日本語の単位なし・半角数値で返してください。分からない項目は null。
-{
-  "company": {"name": null},
-  "period": {"start_date": null, "end_date": null},
-  "balance_sheet": {
-    "total_assets": null, "total_liabilities": null, "total_equity": null,
-    "current_assets": null, "fixed_assets": null,
-    "current_liabilities": null, "long_term_liabilities": null
-  },
-  "income_statement": {
-    "sales": null, "cost_of_sales": null, "gross_profit": null,
-    "operating_expenses": null, "operating_income": null,
-    "ordinary_income": null, "net_income": null
-  },
-  "cash_flows": {
-    "operating_cash_flow": null, "investing_cash_flow": null, "financing_cash_flow": null
-  }
-}
-"""
-
-def _openai_client():
-    # openai==1.x の公式クライアント。proxies を渡さない（互換性エラー回避）。
-    from openai import OpenAI
-    key = os.environ.get("OPENAI_API_KEY")
-    if not key:
-        raise gr.Error("OPENAI_API_KEY が未設定です。Spaces → Settings → **Variables and secrets** に `OPENAI_API_KEY` を追加してください。")
-    return OpenAI(api_key=key, timeout=30)
-
-def extract_financials(images, text_blob, company_hint):
-    client = _openai_client()
-    if images:
-        content = [{"type": "text", "text": SYSTEM_JSON}]
-        if company_hint:
-            content.append({"type": "text", "text": f"会社名の候補: {company_hint}"})
-        for im in images:
-            content.append({"type": "input_image", "image_url": f"data:image/png;base64,{_b64(im)}"})
-        resp = client.chat.completions.create(
-            model=OPENAI_MODEL_VISION,
-            messages=[
-                {"role": "system", "content": "返答は必ず有効な JSON オブジェクトのみ。説明を含めない。"},
-                {"role": "user", "content": content},
-            ],
-            response_format={"type": "json_object"},
-            temperature=0.1,
-        )
-        return json.loads(resp.choices[0].message.content)
-    else:
-        prompt = f"{SYSTEM_JSON}\n\n以下は決算書のテキストです。上記の JSON だけを返してください。\n\n{text_blob or ''}"
-        resp = client.chat.completions.create(
-            model=OPENAI_MODEL_TEXT,
-            messages=[
-                {"role": "system", "content": "返答は必ず有効な JSON オブジェクトのみ。"},
-                {"role": "user", "content": prompt},
-            ],
-            response_format={"type": "json_object"},
-            temperature=0.1,
-        )
-        return json.loads(resp.choices[0].message.content)
-
-# ================== ハンドラ（型ヒントなしで安定化） ==================
-
-def run_analyze(company, use_vision, files, force_ocr):
-    if not files:
-        raise gr.Error("PDF をアップロードしてください。")
-
-    # 1) PDF抽出（テキスト→足りなければ画像化）
+# 既存スコア計算（無ければ簡易版でフォールバック）
+try:
+    from core.scoring import score_company   # 既存の社内スコア（レーダー）
+except Exception:
+    def score_company(fin: Dict[str,Any]) -> Dict[str,Any]:
+        bs = fin.get("balance_sheet",{}) or {}
+        is_ = fin.get("income_statement",{}) or {}
+        safe = lambda x: float(x) if x not in (None,"") else 0.0
+        kpis = {
+            "総資産": safe(bs.get("total_assets")),
+            "売上高": safe(is_.get("sales")),
+            "営業利益": safe(is_.get("operating_income")),
+            "純利益": safe(is_.get("net_income")),
+        }
+        details = []
+        for k,v in kpis.items():
+            val = max(0.0, min(100.0, (v/(kpis["売上高"]+1e-9))*30 if k!="売上高" else 50))
+            details.append({"metric": k, "score": round(val,1)})
+        total = round(sum(d["score"] for d in details)/len(details),1) if details else 0.0
+        grade = "A" if total>=85 else "B" if total>=70 else "C" if total>=55 else "D"
+        return {"total_score": total, "grade": grade, "details": details}
+
+try:
+    from core.external_scoring import score_external  # あなたの外部評価（定量化）
+except Exception:
+    def score_external(df: pd.DataFrame) -> Dict[str,Any]:
+        return {"name":"外部評価（簡易）","external_total": 60.0, "items":[],"notes":"モジュール未検出のため簡易"}
+
+try:
+    from core.ai_judgement import make_ai_memo       # AI所見（中立）
+except Exception:
+    def make_ai_memo(fin: Dict[str,Any], score: Dict[str,Any], ext: Dict[str,Any]) -> str:
+        return "（AI所見モジュール未検出のため簡易）\n- 財務の整合性と収益性を総合的に確認してください。"
+
+from core.extract import parse_pdf  # PDF→(fin, df, meta, log)
+
+UNITS = [("自動",""), ("円","円"), ("千円","千円"), ("万円","万円"),
+         ("百万円","百万円"), ("千万円","千万円"), ("億円","億円")]
+
+def _fmt_yen(x: Any) -> str:
     try:
-        images, raw_text, business_text, dbg = parse_pdf(files, force_ocr=force_ocr)
-    except ExtractError as e:
-        raise gr.Error(f"PDF読み込みに失敗: {e}")
-
-    # 2) Vision 優先 → 失敗ならテキスト
-    try:
-        if use_vision and images:
-            fin = extract_financials(images, None, company or "")
-        else:
-            fin = extract_financials(None, raw_text, company or "")
+        f = float(x)
+        sign = "-" if f < 0 else ""
+        f = abs(f)
+        return f"{sign}{f:,.0f} 円"
     except Exception:
-        try:
-            fin = extract_financials(None, raw_text, company or "")
-        except Exception as e:
-            raise gr.Error(f"AI抽出に失敗: {e}")
+        return "—"
 
-    df = fin_to_df(fin)
-    score_int = score_company(fin)
-    fig = radar(score_int)
-
-    # 3) 外部評価（定量化）
-    try:
-        score_ext = score_external_from_df(df)
-    except Exception as e:
-        score_ext = {"name": "外部評価（失敗）", "external_total": None, "items": [], "notes": str(e)}
-
-    # 4) AI 所見（中立）
-    try:
-        memo = make_ai_memo(
-            company=company or "",
-            fin=fin,
-            score_internal=score_int,
-            score_external=score_ext,
-            business_text=business_text
-        )
-    except Exception as e:
-        memo = f"AI所見の生成に失敗: {e}"
-
-    return (
-        json.dumps(fin, ensure_ascii=False, indent=2),
-        df,
-        json.dumps(score_int, ensure_ascii=False, indent=2),
-        fig,
-        memo,
-        json.dumps(score_ext, ensure_ascii=False, indent=2),
-        dbg
-    )
-
-def run_recalc(df):
-    try:
-        fin = df_to_fin(df)
-        score_int = score_company(fin)
-        fig = radar(score_int)
-        return (
-            json.dumps(score_int, ensure_ascii=False, indent=2),
-            fig,
-            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 _radar(score: Dict[str, Any]) -> go.Figure:
+    labels = [d["metric"] for d in score.get("details",[])]
+    values = [d["score"] for d in score.get("details",[])]
+    if not labels:
+        labels, values = ["データ不足"], [0]
+    fig = go.Figure()
+    fig.add_trace(go.Scatterpolar(r=values + values[:1], theta=labels + labels[:1], fill="toself"))
+    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.get('total_score',0)}（グレード: {score.get('grade','-')}）")
+    return fig
 
-# ================== UI 組み立て ==================
+def _summary_cards(fin: Dict[str,Any]) -> str:
+    bs = fin.get("balance_sheet",{}) or {}
+    is_ = fin.get("income_statement",{}) or {}
+    html = f"""
+    <style>
+    .cards {{display:grid;grid-template-columns:repeat(4,minmax(0,1fr));gap:12px}}
+    .card {{border:1px solid #eee;border-radius:12px;padding:12px;background:#fff;box-shadow:0 1px 2px rgba(0,0,0,.04)}}
+    .k {{font-size:.8rem;color:#666}}
+    .v {{font-size:1.1rem;font-weight:700;margin-top:6px}}
+    @media (max-width: 980px) {{ .cards {{grid-template-columns:repeat(2,minmax(0,1fr));}}}}
+    </style>
+    <div class="cards">
+      <div class="card"><div class="k">総資産</div><div class="v">{_fmt_yen(bs.get('total_assets'))}</div></div>
+      <div class="card"><div class="k">売上高</div><div class="v">{_fmt_yen(is_.get('sales'))}</div></div>
+      <div class="card"><div class="k">営業利益</div><div class="v">{_fmt_yen(is_.get('operating_income'))}</div></div>
+      <div class="card"><div class="k">当期純利益</div><div class="v">{_fmt_yen(is_.get('net_income'))}</div></div>
+    </div>
+    """
+    return html
 
 def build_ui():
     with gr.Blocks(theme=gr.themes.Soft(primary_hue="indigo"), fill_height=True, analytics_enabled=False) as demo:
-        gr.Markdown("## 🧮 企業スコアリング（PDF解析 × OpenAI Vision）")
+        gr.Markdown("## 🧮 企業スコアリング（PDF解析 × 単位自動判定 × UI改善）")
+
+        state_fin = gr.State({})
+        state_df  = gr.State(pd.DataFrame(columns=["category","item","value"]))
+        state_unit_detected = gr.State({"label":"円","scale":1.0})
+        state_unit_current  = gr.State({"label":"円","scale":1.0})
 
         with gr.Row():
-            with gr.Column(scale=1):
+            with gr.Column(scale=1, min_width=320):
                 company = gr.Textbox(label="企業名（任意）", placeholder="例：株式会社OO")
-                use_vision = gr.Checkbox(value=True, label="OpenAIでPDFをAI解析（Vision）")
-                force_ocr = gr.Checkbox(value=False, label="OCRを強制（スキャンPDF向け）")
+                use_vision = gr.Checkbox(value=True, label="OpenAI Visionで解析")
                 files = gr.File(label="決算書PDF（複数可）", file_count="multiple", type="filepath")
-                run_btn = gr.Button("📄 PDFを解析してテンプレに反映", variant="primary")
-                recalc_btn = gr.Button("🔁 この表の値で再計算")
-                gr.Markdown("※ 画像化やVisionに失敗した場合はテキスト抽出に自動フォールバックします。")
-
-            with gr.Column(scale=1):
-                fin_json = gr.Code(label="抽出JSON（編集不可）", language="json", interactive=False)
-
-        with gr.Tabs():
-            with gr.Tab("抽出結果（表で編集可）"):
-                df_out = gr.Dataframe(headers=["category", "item", "value"], interactive=True, wrap=True)
-            with gr.Tab("スコアリング（内部ルール）"):
-                score_json = gr.Code(label="スコア（JSON）", language="json")
-                chart = gr.Plot(label="スコアレーダー")
-            with gr.Tab("AI診断（中立・日本語）"):
-                insight_md = gr.Markdown()
-            with gr.Tab("外部評価（定量化）"):
-                ext_json = gr.Code(label="外部評価JSON", language="json")
-            with gr.Tab("抽出ログ/デバッグ"):
-                debug_out = gr.Textbox(label="ログ",lines=12, interactive=False)
+
+                with gr.Group():
+                    unit_manual = gr.Dropdown(choices=[u for u,_ in UNITS], value="自動", label="単位の上書き（任意）")
+                    apply_unit_btn = gr.Button("↻ 単位を反映（換算）")
+
+                run_btn = gr.Button("📄 PDFを解析", variant="primary")
+                dl_csv = gr.File(label="ダウンロード（CSV）", interactive=False)
+                dl_json = gr.File(label="ダウンロード（JSON）", interactive=False)
+
+                unit_info = gr.HTML()
+                debug_out = gr.Textbox(label="ログ", lines=10, interactive=False)
+
+            with gr.Column(scale=2):
+                summary = gr.HTML()
+                with gr.Tabs():
+                    with gr.Tab("編集可能テーブル"):
+                        df_out = gr.Dataframe(headers=["category","item","value"], interactive=True, wrap=True)
+                        recalc_btn = gr.Button("🔁 表の値で再計算")
+                    with gr.Tab("スコア"):
+                        score_json = gr.JSON(label="スコア（内部JSON）", visible=False)
+                        chart = gr.Plot(label="スコアレーダー")
+                    with gr.Tab("外部評価"):
+                        ext_json = gr.JSON(label="外部評価（内部JSON）", visible=False)
+                        ext_md = gr.Markdown()
+                    with gr.Tab("AI所見"):
+                        insight_md = gr.Markdown()
+
+        # -------- ハンドラ --------
+        def on_analyze(company: str, use_vision: bool, files: List[str]):
+            try:
+                fin, df, meta, log = parse_pdf(files, company, use_vision)
+                # 状態に反映
+                state_fin_v = fin
+                state_df_v = df
+                det = {"label": meta["unit_label"], "scale": meta["unit_scale"]}
+                cur = det.copy()
+
+                # スコア
+                sc = score_company(fin)
+                fig = _radar(sc)
+
+                # 外部評価（数値中心）
+                ext = score_external(df)
+                ext_md_text = f"**外部評価合計:** {ext.get('external_total','—')}\n\n" \
+                              f"{ext.get('notes','')}"
+
+                # 所見
+                memo = make_ai_memo(fin, sc, ext)
+
+                # ダウンロード用ファイル作成
+                csv_path = "/tmp/result.csv"
+                json_path = "/tmp/result.json"
+                df.to_csv(csv_path, index=False)
+                with open(json_path, "w", encoding="utf-8") as f:
+                    json.dump(fin, f, ensure_ascii=False, indent=2)
+
+                unit_badge = f"<b>単位（自動）:</b> {det['label']}  ×{det['scale']:,.0f}"
+                if meta.get("warnings"):
+                    unit_badge += "<br>" + " / ".join(f"⚠️ {w}" for w in meta["warnings"])
+
+                return (
+                    fin,              # state_fin
+                    df,               # state_df
+                    det,              # detected
+                    cur,              # current
+                    _summary_cards(fin),
+                    df, fig, sc,
+                    ext, ext_md_text,
+                    memo,
+                    unit_badge,
+                    log,
+                    csv_path,
+                    json_path
+                )
+            except Exception as e:
+                import traceback
+                tb = traceback.format_exc(limit=4)
+                raise gr.Error(f"解析に失敗しました: {e}\n\n{tb}")
 
         run_btn.click(
-            run_analyze,
-            inputs=[company, use_vision, files, force_ocr],
-            outputs=[fin_json, df_out, score_json, chart, insight_md, ext_json, debug_out],
-            concurrency_limit=1
+            on_analyze,
+            inputs=[company, use_vision, files],
+            outputs=[
+                state_fin, state_df, state_unit_detected, state_unit_current,
+                summary,
+                df_out, chart, score_json,
+                ext_json, ext_md,
+                insight_md,
+                unit_info,
+                debug_out,
+                dl_csv, dl_json
+            ],
         )
+
+        def on_apply_unit(unit_label: str, df_cur: pd.DataFrame,
+                          det: Dict[str,Any], cur: Dict[str,Any], fin_cur: Dict[str,Any]):
+            from core.unit_utils import UNIT_SCALE, apply_unit_scale
+            # 新しい係数
+            if unit_label in ("", "自動"):
+                new = det
+            else:
+                new = {"label": unit_label, "scale": UNIT_SCALE.get(unit_label, 1.0)}
+            # 係数比で再換算（現在→新）
+            ratio = (new["scale"] / max(cur.get("scale",1.0), 1e-12))
+
+            def _scale_df(df: pd.DataFrame) -> pd.DataFrame:
+                df2 = df.copy()
+                for i in df2.index:
+                    try:
+                        v = df2.at[i,"value"]
+                        df2.at[i,"value"] = float(v) * ratio if v not in (None,"") else v
+                    except Exception:
+                        pass
+                return df2
+
+            df2 = _scale_df(df_cur)
+            fin2 = apply_unit_scale(fin_cur, ratio)  # 比率で再拡大/縮小
+
+            sc2 = score_company(fin2)
+            fig2 = _radar(sc2)
+            unit_badge = f"<b>単位（現在）:</b> {new['label']} ×{new['scale']:,.0f}"
+            return fin2, df2, new, _summary_cards(fin2), df2, fig2, sc2, unit_badge
+
+        apply_unit_btn.click(
+            on_apply_unit,
+            inputs=[unit_manual, df_out, state_unit_detected, state_unit_current, state_fin],
+            outputs=[state_fin, state_df, state_unit_current, summary, df_out, chart, score_json, unit_info]
+        )
+
+        def on_recalc(df_cur: pd.DataFrame, fin_cur: Dict[str,Any]):
+            # ユーザー編集を反映して再計算
+            fin2 = {"balance_sheet":{}, "income_statement":{}, "cash_flows":{}}
+            for _, r in df_cur.iterrows():
+                cat, item, val = str(r["category"]), str(r["item"]), r["value"]
+                try: v = None if val in (None,"","null") else float(val)
+                except Exception: v = None
+                if cat in fin2: fin2[cat][item] = v
+            sc2 = score_company(fin2)
+            fig2 = _radar(sc2)
+            return fin2, sc2, fig2, _summary_cards(fin2)
+
         recalc_btn.click(
-            run_recalc,
-            inputs=[df_out],
-            outputs=[score_json, chart, fin_json],
-            concurrency_limit=1
+            on_recalc,
+            inputs=[df_out, state_fin],
+            outputs=[state_fin, score_json, chart, summary]
         )
 
     return demo