# ui/ui_app.py from __future__ import annotations import json from typing import Dict, Any, List, Tuple import gradio as gr import plotly.graph_objects as go import pandas as pd from core.extract import parse_pdf from core.market_infer import infer_market_metrics from core.external_scoring import ( get_external_template_df, fill_missing_with_external, merge_market_into_external_df, score_external_from_df, ) from core.ai_judgement import suggest_external_with_llm, ai_evaluate # ---------- charts ---------- def _radar(title: str, cat_scores: Dict[str, float]) -> go.Figure: if not cat_scores: cat_scores = {"N/A": 0.0} labels = list(cat_scores.keys()) vals = [float(cat_scores[k] or 0.0) for k in labels] fig = go.Figure() fig.add_trace(go.Scatterpolar(r=vals + [vals[0]], theta=labels + [labels[0]], fill="toself", name=title)) fig.update_layout(polar=dict(radialaxis=dict(visible=True, range=[0, 100])), showlegend=False, height=360, margin=dict(l=30, r=30, t=40, b=30), title=title) return fig def _diff_bar(ext: Dict[str, float], ai: Dict[str, float]) -> go.Figure: ks = sorted(set(ext.keys()) | set(ai.keys())) diffs = [(float(ai.get(k, 0.0)) - float(ext.get(k, 0.0))) for k in ks] fig = go.Figure(data=[go.Bar(x=ks, y=diffs)]) fig.update_layout(title="AI評点 - 外部評価（カテゴリ差分）", height=320, margin=dict(l=30, r=30, t=40, b=30)) return fig def _fmt(x): try: f = float(x) if abs(f) >= 1e8: return f"{f/1e8:.2f}億" if abs(f) >= 1e6: return f"{f/1e6:.2f}百万円" if abs(f) >= 1e3: return f"{f/1e3:.1f}千" return f"{f:.0f}" except Exception: return str(x) if x not in (None, "") else "—" def _cards(company, meta, fin, ext_total, ai_total) -> str: bs = fin.get("balance_sheet", {}) or {}; is_ = fin.get("income_statement", {}) or {} ta = bs.get("total_assets") or 0; te = bs.get("total_equity") or 0 er = "—" try: ta = float(ta); te = float(te) er = f"{(te/ta*100):.1f}%" if ta>0 else "—" except Exception: pass period = "" if meta and isinstance(meta.get("period"), dict): period = f"{meta['period'].get('start_date','')} ~ {meta['period'].get('end_date','')}" unit = (meta.get("unit") or "円").replace("JPY","円") return f"""
企業名
{company or fin.get('company',{}).get('name','—')}
期間: {period or '—'} / 単位: {unit}
売上高
{_fmt(is_.get('sales'))}
営業利益
{_fmt(is_.get('operating_income'))}
自己資本比率
{er}
外部評価（定量）
{ext_total:.1f}
AI評点
{ai_total:.1f}
""" # ---------- market df helpers ---------- def _market_df_from_dict(d: Dict[str, Any]) -> pd.DataFrame: rows = [] order = [ "市場の年成長率(%)","市場成熟度(0-1)","競争強度(0-10)","参入障壁(0-10)","価格決定力(0-10)", "サイクル感応度(0-10)","規制リスク(0-10)","技術破壊リスク(0-10)","TAM_億円","SAM_億円","SOM_億円" ] for k in order: rows.append([k, d.get(k,"")]) return pd.DataFrame(rows, columns=["指標","値"]) def _dict_from_market_df(df: pd.DataFrame) -> Dict[str, Any]: out = {} try: for _, r in df.iterrows(): k = str(r["指標"]); v = r["値"] try: out[k] = float(v) except Exception: out[k] = None except Exception: # 列崩れ対策：空dictで返す return {} return out # ---------- flows ---------- def on_analyze(company: str, use_vision: bool, files: List[str]): """ 例外はcatchしてUIに出す。常に所定の型で返す。 """ try: if not files: raise RuntimeError("PDF をアップロードしてください。") fin, df_fin, meta, log = parse_pdf(files, company, use_vision) # 外部テンプレ & 埋め候補 ext_df = get_external_template_df() ext_df = fill_missing_with_external(ext_df, suggest_external_with_llm(fin, company)) # 市場（空の表）/ 仮スコア market_df = _market_df_from_dict({}) ext_res = score_external_from_df(ext_df) ai_res = ai_evaluate(fin, {}) cards = _cards(company, meta, fin, ext_res["external_total"], ai_res["ai_total"]) ext_fig = _radar("外部評価（カテゴリ）", ext_res.get("category_scores", {})) ai_fig = _radar("AI評点（カテゴリ）", ai_res.get("category_scores", {})) diff = _diff_bar(ext_res.get("category_scores", {}), ai_res.get("category_scores", {})) return (cards, df_fin, ext_df, market_df, float(ext_res["external_total"]), float(ai_res["ai_total"]), ext_fig, ai_fig, diff, json.dumps(fin, ensure_ascii=False, indent=2), json.dumps(ext_res, ensure_ascii=False, indent=2), json.dumps(ai_res, ensure_ascii=False, indent=2), "\n".join([str(x) for x in (log if isinstance(log,list) else [log])])) except Exception as e: # 失敗してもUIが壊れないように空プレースホルダで返す empty_df = pd.DataFrame(columns=["カテゴリー","入力項目","値"]) return ( f"
解析に失敗: {e}
", pd.DataFrame(columns=["category","item","value"]), empty_df, _market_df_from_dict({}), 0.0, 0.0, _radar("外部評価（カテゴリ）", {}), _radar("AI評点（カテゴリ）", {}), _diff_bar({}, {}), "{}", "{}","{}", f"TRACE: {type(e).__name__}: {e}" ) def on_market_infer(industry: str, products_text: str, country: str, horizon: int, ext_df: pd.DataFrame, fin_json: str): try: prods = [p.strip() for p in (products_text or "").splitlines() if p.strip()] market = infer_market_metrics(industry or "", prods, country or "JP", int(horizon or 3)) market_df = _market_df_from_dict(market) # ext_dfに市場推定を統合 ext_df2 = merge_market_into_external_df(ext_df, market, prods) # スコア更新 fin = json.loads(fin_json or "{}") ext_res = score_external_from_df(ext_df2) ext_like = { "市場の年成長率(%)": market.get("市場の年成長率(%)"), "主力商品数": len(prods), "成長中主力商品数": sum(1 for p in prods if (market.get("製品別年成長率(%)",{}).get(p,0) or 0)>10) } ai_res = ai_evaluate(fin, ext_like) ext_fig = _radar("外部評価（カテゴリ）", ext_res.get("category_scores", {})) ai_fig = _radar("AI評点（カテゴリ）", ai_res.get("category_scores", {})) diff = _diff_bar(ext_res.get("category_scores", {}), ai_res.get("category_scores", {})) return (market_df, ext_df2, float(ext_res["external_total"]), float(ai_res["ai_total"]), ext_fig, ai_fig, diff, json.dumps(ext_res, ensure_ascii=False, indent=2), json.dumps(ai_res, ensure_ascii=False, indent=2), "市場推定OK: " + "; ".join(market.get("注記", [])[:3])) except Exception as e: return (_market_df_from_dict({}), ext_df, 0.0, 0.0, _radar("外部評価（カテゴリ）", {}), _radar("AI評点（カテゴリ）", {}), _diff_bar({}, {}), "{}", "{}", f"市場推定に失敗: {e}") def on_rescore_all(ext_df: pd.DataFrame, market_df: pd.DataFrame, fin_json: str, products_text: str): try: fin = json.loads(fin_json or "{}") prods = [p.strip() for p in (products_text or "").splitlines() if p.strip()] market = _dict_from_market_df(market_df) ext_df2 = merge_market_into_external_df(ext_df, market, prods) ext_res = score_external_from_df(ext_df2) ext_like = { "市場の年成長率(%)": market.get("市場の年成長率(%)"), "主力商品数": len(prods), "成長中主力商品数": sum(1 for p in prods if (market.get("製品別年成長率(%)",{}).get(p,0) or 0)>10) } ai_res = ai_evaluate(fin, ext_like) ext_fig = _radar("外部評価（カテゴリ）", ext_res.get("category_scores", {})) ai_fig = _radar("AI評点（カテゴリ）", ai_res.get("category_scores", {})) diff = _diff_bar(ext_res.get("category_scores", {}), ai_res.get("category_scores", {})) return (ext_df2, float(ext_res["external_total"]), float(ai_res["ai_total"]), ext_fig, ai_fig, diff, json.dumps(ext_res, ensure_ascii=False, indent=2), json.dumps(ai_res, ensure_ascii=False, indent=2), "再計算完了") except Exception as e: return (ext_df, 0.0, 0.0, _radar("外部評価（カテゴリ）", {}), _radar("AI評点（カテゴリ）", {}), _diff_bar({}, {}), "{}", "{}", f"再計算に失敗: {e}") def build_ui(): with gr.Blocks(theme=gr.themes.Soft(primary_hue="indigo"), analytics_enabled=False) as demo: gr.Markdown("## 🧮 企業スコアリング：PDF抽出 × 市場推定（LLM）× 外部定量 × AI評点") with gr.Row(): with gr.Column(scale=1): company = gr.Textbox(label="企業名（任意）") use_vision = gr.Checkbox(value=True, label="OpenAI VisionでPDF表を補完") files = gr.File(label="決算書PDF（複数可）", file_count="multiple", type="filepath") run_btn = gr.Button("📄 PDFを解析", variant="primary") with gr.Column(scale=2): cards = gr.HTML(label="サマリー") with gr.Tab("入力/市場推定"): with gr.Row(): industry = gr.Textbox(label="事業領域（業界・カテゴリ）", placeholder="例）ヘルスケアIT / 産業ロボット 等") products = gr.Textbox(label="主力商品（1行1件）", lines=4, placeholder="製品A\n製品B\n…") with gr.Row(): country = gr.Dropdown(choices=["JP","US","EU","APAC","GLOBAL"], value="JP", label="対象地域") horizon = gr.Slider(1, 7, value=3, step=1, label="予測年数") infer_btn = gr.Button("🔎 市場を推定（LLM）", variant="secondary") market_df = gr.Dataframe(label="市場メトリクス（編集可）", interactive=True, wrap=True) with gr.Tab("外部入力/財務"): df_fin = gr.Dataframe(label="抽出テーブル（編集可）", interactive=True, wrap=True) ext_df = gr.Dataframe(label="外部入力（編集可）", interactive=True, wrap=True) with gr.Tab("スコア"): with gr.Row(): ext_total = gr.Number(label="外部評価 合計(0-100)", value=0.0, precision=1, interactive=False) ai_total = gr.Number(label="AI評点 合計(0-100)", value=0.0, precision=1, interactive=False) with gr.Row(): ext_plot = gr.Plot(label="外部評価（レーダー）") ai_plot = gr.Plot(label="AI評点（レーダー）") diff_plot = gr.Plot(label="差分（棒）") rescore_btn = gr.Button("🔁 すべて再計算", variant="secondary") with gr.Tab("詳細"): fin_json = gr.Code(label="抽出JSON", language="json") ext_json = gr.Code(label="外部評価JSON", language="json") ai_json = gr.Code(label="AI評点JSON", language="json") debug = gr.Textbox(label="ログ", lines=8) # state fin_state = gr.State("") # wire run_btn.click( on_analyze, inputs=[company, use_vision, files], outputs=[cards, df_fin, ext_df, market_df, ext_total, ai_total, ext_plot, ai_plot, diff_plot, fin_json, ext_json, ai_json, debug], ).then(lambda x: x, inputs=[fin_json], outputs=[fin_state]) infer_btn.click( on_market_infer, inputs=[industry, products, country, horizon, ext_df, fin_state], outputs=[market_df, ext_df, ext_total, ai_total, ext_plot, ai_plot, diff_plot, ext_json, ai_json, debug], ) rescore_btn.click( on_rescore_all, inputs=[ext_df, market_df, fin_state, products], outputs=[ext_df, ext_total, ai_total, ext_plot, ai_plot, diff_plot, ext_json, ai_json, debug], ) return demo