app.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import os, re, json, io, tempfile
from datetime import datetime
from typing import List, Optional, Dict, Any, Tuple

import gradio as gr
import yaml
import matplotlib.pyplot as plt
from reportlab.lib.pagesizes import A4
from reportlab.lib import colors
from reportlab.lib.styles import ParagraphStyle, getSampleStyleSheet
from reportlab.platypus import SimpleDocTemplate, Paragraph, Spacer, Table, TableStyle, Image as RLImage
from reportlab.pdfbase import pdfmetrics
from reportlab.pdfbase.cidfonts import UnicodeCIDFont

from schemas import FinancialExtract, ExtractedPeriod, MultipleSuggestion, MarketOutlook
from finance_core import (
    compute_ratios, credit_decision, loan_decision, investment_decision, build_report_dict,
)
from llm_extract import (
    get_client, upload_file_to_openai, extract_financials_from_files,
    suggest_multiples_with_llm, suggest_market_outlook_with_llm,
)

VISION_MODEL = os.environ.get("OPENAI_VISION_MODEL", "gpt-4o-mini")
TEXT_MODEL   = os.environ.get("OPENAI_TEXT_MODEL",   "gpt-4o-mini")
BASE_RATE    = float(os.environ.get("BASE_RATE", "2.0"))  # % p.a.

def _load_policies() -> dict:
    default_yaml = """\
liquidity:
  current_ratio_good: 2.0
  current_ratio_ok: 1.5
leverage:
  debt_to_equity_good: 0.5
  debt_to_equity_ok: 1.0
coverage:
  interest_coverage_good: 6.0
  interest_coverage_ok: 3.0
profitability:
  net_margin_good: 0.1
  net_margin_ok: 0.03
growth:
  revenue_growth_high: 0.2
  revenue_growth_ok: 0.05
"""
    cfg_dir = os.path.join(os.path.dirname(__file__), "config")
    cfg_path = os.path.join(cfg_dir, "risk_policies.yaml")
    default = yaml.safe_load(default_yaml)
    try:
        with open(cfg_path, "r", encoding="utf-8") as f:
            data = yaml.safe_load(f) or {}
            if isinstance(data, dict):
                default.update(data)
            return default
    except FileNotFoundError:
        os.makedirs(cfg_dir, exist_ok=True)
        with open(cfg_path, "w", encoding="utf-8") as f:
            f.write(default_yaml)
        return default

POLICIES = _load_policies()

# ---------- ファイル入力吸収 ----------
def _read_file_input(f):
    if isinstance(f, (str, bytes)) or hasattr(f, "__fspath__"):
        path = os.fspath(f)
        with open(path, "rb") as fh:
            return os.path.basename(path), fh.read()
    if hasattr(f, "read"):
        data = f.read()
        name = getattr(f, "name", "uploaded")
        try: base = os.path.basename(name)
        except Exception: base = "uploaded"
        return base, data
    if isinstance(f, dict):
        path = f.get("path") or f.get("name")
        if path and os.path.exists(path):
            with open(path, "rb") as fh:
                return os.path.basename(path), fh.read()
        data = f.get("data")
        if data is not None:
            name = f.get("orig_name") or "uploaded"
            if isinstance(data, str): data = data.encode("utf-8")
            return name, data
    try:
        path = str(f)
        if os.path.exists(path):
            with open(path, "rb") as fh:
                return os.path.basename(path), fh.read()
    except Exception:
        pass
    raise ValueError(f"Unsupported file input type: {type(f)}")

# ---------- 単位検出＆換算 ----------
def _concat_pdf_text(paths: List[str], max_chars: int = 180_000) -> str:
    try:
        from pypdf import PdfReader
    except Exception:
        return ""
    out, total = [], 0
    for p in paths:
        try:
            r = PdfReader(p)
            for page in r.pages:
                t = page.extract_text() or ""
                if t:
                    out.append(t); total += len(t)
                    if total > max_chars: break
        except Exception:
            continue
        if total > max_chars: break
    return "\n\n".join(out)[:max_chars]

def detect_unit_multiplier_from_paths(paths: List[str]) -> Tuple[float, str]:
    text = _concat_pdf_text(paths)
    if not text: return 1.0, "不明"
    lower = text.lower()
    if re.search(r"単位[：:]\s*百万円", text) or re.search(r"（百万円）", text): return 1_000_000.0, "百万円"
    if re.search(r"単位[：:]\s*千円", text) or re.search(r"（千円）", text):   return 1_000.0,     "千円"
    if re.search(r"単位[：:]\s*万円", text) or re.search(r"（万円）", text):   return 10_000.0,    "万円"
    if re.search(r"単位[：:]\s*円", text) or re.search(r"（円）", text):       return 1.0,         "円"
    if re.search(r"in\s+millions\s+of\s+(yen|jpy|usd|dollars?)", lower) or re.search(r"\b(jpy|¥|\$|usd)\s*\(\s*millions?\s*\)", lower):
        return 1_000_000.0, "millions"
    if re.search(r"in\s+thousands\s+of\s+(yen|jpy|usd|dollars?)", lower) or re.search(r"\b(jpy|¥|\$|usd)\s*\(\s*thousands?\s*\)", lower):
        return 1_000.0, "thousands"
    if re.search(r"百万円", text): return 1_000_000.0, "百万円"
    return 1.0, "不明"

_NUM_FIELDS = [
    "revenue","cogs","ebit","depreciation","ebitda","net_income",
    "cash_and_equivalents","accounts_receivable","inventory","accounts_payable",
    "current_assets","current_liabilities","total_assets","total_equity",
    "total_debt","interest_expense",
]
def scale_extract_inplace(extract: FinancialExtract, multiplier: float) -> None:
    if not multiplier or multiplier == 1: return
    for period in extract.periods:
        for k in _NUM_FIELDS:
            v = getattr(period, k)
            if v is not None:
                try: setattr(period, k, float(v) * float(multiplier))
                except Exception: pass

# ---------- PDF 生成 ----------
def _register_jp_fonts():
    # 組込みCIDフォント（埋め込み不要・日本語可）
    try:
        pdfmetrics.registerFont(UnicodeCIDFont('HeiseiKakuGo-W5'))
        pdfmetrics.registerFont(UnicodeCIDFont('HeiseiMin-W3'))
        return "HeiseiKakuGo-W5"
    except Exception:
        return "Helvetica"

def _make_metrics_chart_png(ratios: Dict[str, Any]) -> str:
    # 英数字のみのシンプルグラフ（日本語フォントが無くてもOK）
    vals = []
    labels = []
    for key,label in [("revenue","Revenue"),("ebitda","EBITDA"),("net_income","NetIncome")]:
        v = ratios.get(key)
        if isinstance(v, str):
            try:
                # 末尾通貨文字を除外して数値化（ざっくり）
                n = float(v.replace("—","").replace(",","").replace("円","").replace("百万円","e6").replace("億円","e8").split()[0] or 0)
            except Exception:
                n = 0.0
        else:
            n = float(v or 0.0)
        labels.append(label); vals.append(max(0.0, n))
    fig, ax = plt.subplots(figsize=(4,2.2), dpi=160)
    ax.bar(labels, vals)
    ax.set_title("Key Metrics")
    ax.ticklabel_format(style='plain', axis='y')
    tmp_png = os.path.join(tempfile.gettempdir(), f"chart_{os.getpid()}.png")
    fig.tight_layout()
    fig.savefig(tmp_png)
    plt.close(fig)
    return tmp_png

def export_pdf(report: Dict[str, Any], pdf_path: str) -> None:
    font_name = _register_jp_fonts()
    styles = getSampleStyleSheet()
    styles.add(ParagraphStyle(name="JPTitle", fontName=font_name, fontSize=16, leading=20, spaceAfter=10))
    styles.add(ParagraphStyle(name="JPHead",  fontName=font_name, fontSize=12, leading=16, spaceBefore=6, spaceAfter=6))
    styles.add(ParagraphStyle(name="JPBody",  fontName=font_name, fontSize=10, leading=14))
    doc = SimpleDocTemplate(pdf_path, pagesize=A4, rightMargin=36, leftMargin=36, topMargin=36, bottomMargin=36)
    flow = []
    md = report["metadata"]; dec = report["decisions"]; ratios = report["ratios"]
    mo = report.get("market_outlook")
    unit = report.get("unit_detection", {})

    title = f"評価レポート：{md.get('company_name') or '不明'}"
    flow += [Paragraph(title, styles["JPTitle"]), Spacer(1,6)]

    # バッジ風サマリ
    badges = []
    if "credit" in dec: badges.append(f"与信: <b>{dec['credit']['rating']}</b>")
    if "investment" in dec: badges.append(f"投資魅力度: <b>{dec['investment']['attractiveness']}/5</b>")
    if mo: badges.append(f"市場期待: <b>{mo.get('expectation_label')}</b>（CAGR {mo.get('expected_market_cagr',0):.1f}%）")
    if ratios.get("revenue_growth_pct"): badges.append(f"売上成長率: <b>{ratios['revenue_growth_pct']}</b>")
    flow += [Paragraph(" / ".join(badges), styles["JPBody"]), Spacer(1,6)]

    # 主要指標テーブル
    tbl_data = [
        ["売上高", ratios.get("revenue"), "EBITDA", ratios.get("ebitda"), "純利益", ratios.get("net_income")],
        ["流動比率", ratios.get("current_ratio"), "D/E", ratios.get("debt_to_equity"), "ICR", ratios.get("interest_coverage")],
        ["ROE", ratios.get("roe_pct"), "ROA", ratios.get("roa_pct"), "キャッシュ比率", ratios.get("cash_ratio")],
        ["DSO", ratios.get("dso_days"), "DIO", ratios.get("dio_days"), "DPO", ratios.get("dpo_days")],
        ["CCC", ratios.get("ccc_days"), "売上成長率", ratios.get("revenue_growth_pct"), "", ""],
    ]
    tbl = Table(tbl_data, colWidths=[70,110,70,110,70,110])
    tbl.setStyle(TableStyle([
        ("FONTNAME",(0,0),(-1,-1), font_name),
        ("FONTSIZE",(0,0),(-1,-1),9),
        ("GRID",(0,0),(-1,-1),0.25, colors.grey),
        ("BACKGROUND",(0,0),(-1,0), colors.whitesmoke),
        ("VALIGN",(0,0),(-1,-1),"MIDDLE"),
    ]))
    flow += [Paragraph("主要指標", styles["JPHead"]), tbl, Spacer(1,6)]

    # 与信
    if "credit" in dec:
        c = dec["credit"]
        flow += [Paragraph("与信判断（提案）", styles["JPHead"])]
        flow += [Paragraph(f"与信ランク: {c['rating']}（スコア {c['risk_score']}/100） / 取引サイト: {c['site_days']}日 / 上限: {c['transaction_limit_display']} / 見直し: {c['review_cycle']}", styles["JPBody"])]

    # 融資
    if "loan" in dec:
        l = dec["loan"]
        flow += [Paragraph("融資判断（提案）", styles["JPHead"])]
        flow += [Paragraph(f"上限額: {l['max_principal_display']} / 期間: {l['term_years']}年 / 金利: {l['interest_rate_pct']}% / 目標DSCR: {l['target_dscr']}", styles["JPBody"])]

    # 投資＋市場期待＋シナリオ
    if "investment" in dec:
        inv = dec["investment"]
        flow += [Paragraph("投資判断（提案）", styles["JPHead"])]
        flow += [Paragraph(f"EV: {inv['ev_display']} / 時価総額: {inv['market_cap_display']} / 想定投資レンジ: {inv['recommended_check_size_display']} / 魅力度: {inv['attractiveness']}/5（成長性: {inv['growth_label']}） / 市場期待補正: x{inv['market_factor']:.2f}", styles["JPBody"])]
        # シナリオ表
        sc_rows = [["シナリオ","倍率","投資レンジ"]]
        for r in inv.get("scenarios", []):
            sc_rows.append([r["label"], f"x{r['factor']:.2f}", r["check_size_display"]])
        sc_tbl = Table(sc_rows, colWidths=[80,80,160])
        sc_tbl.setStyle(TableStyle([
            ("FONTNAME",(0,0),(-1,-1), font_name),
            ("FONTSIZE",(0,0),(-1,-1),9),
            ("GRID",(0,0),(-1,-1),0.25, colors.grey),
            ("BACKGROUND",(0,0),(-1,0), colors.whitesmoke),
        ]))
        flow += [sc_tbl, Spacer(1,6)]

    # 単位
    unit = report.get("unit_detection", {})
    flow += [Paragraph("単位（検出）", styles["JPHead"]),
             Paragraph(f"ソース表記: {unit.get('source_label','不明')} / 乗数: x{unit.get('multiplier',1):,}", styles["JPBody"]),
             Spacer(1,6)]

    # グラフを挿入（英数字のみ）
    try:
        png = _make_metrics_chart_png(ratios)
        flow += [Paragraph("Key Metrics Chart", styles["JPHead"]), RLImage(png, width=360, height=200), Spacer(1,6)]
    except Exception:
        pass

    # 根拠要約
    if mo:
        flow += [Paragraph("市場規模拡大の期待（根拠）", styles["JPHead"]),
                 Paragraph(mo.get("rationale","—"), styles["JPBody"])]

    flow += [Spacer(1,12), Paragraph("免責: 本レポートは参考情報です。最終判断は自己責任で行ってください。", styles["JPBody"])]
    doc.build(flow)

# ---------- サマリUI ----------
def build_human_summary(extract: FinancialExtract, ratios: Dict[str, Any], decisions: Dict[str, Any],
                        unit_info: Dict[str, Any], market: Optional[MarketOutlook]) -> str:
    p = []
    p.append(f"### 評価サマリ")
    badges = []
    if "credit" in decisions: badges.append(f"与信: **{decisions['credit']['rating']}**")
    if "investment" in decisions: badges.append(f"投資魅力度: **{decisions['investment']['attractiveness']}/5**")
    if market: badges.append(f"市場期待: **{market.expectation_label}**（{market.expected_market_cagr:.1f}%）")
    if ratios.get("revenue_growth_pct"): badges.append(f"売上成長率: **{ratios['revenue_growth_pct']}**")
    p.append(" / ".join(badges))
    p.append("\n### 主要KPI\n- 売上高: {revenue}\n- EBITDA: {ebitda}\n- 純利益: {net_income}\n- 流動比率: {cr}\n- D/E: {de}\n- ICR: {icr}\n- ROE: {roe}\n- ROA: {roa}\n- 売上成長率: {rg}".format(
        revenue=ratios.get("revenue"), ebitda=ratios.get("ebitda"), net_income=ratios.get("net_income"),
        cr=ratios.get("current_ratio"), de=ratios.get("debt_to_equity"), icr=ratios.get("interest_coverage"),
        roe=ratios.get("roe_pct"), roa=ratios.get("roa_pct"), rg=ratios.get("revenue_growth_pct"),
    ))
    if "credit" in decisions:
        c = decisions["credit"]
        p.append("### 与信判断（提案）\n- ランク: **{r}**（{s}/100） / サイト: **{d}日** / 上限: **{lim}** / 見直し: **{rev}**".format(
            r=c["rating"], s=c["risk_score"], d=c["site_days"], lim=c["transaction_limit_display"], rev=c["review_cycle"]
        ))
    if "loan" in decisions:
        l = decisions["loan"]
        p.append("### 融資判断（提案）\n- 上限: **{p}** / 期間: **{y}年** / 金利: **{i}%** / 目標DSCR: **{d}**".format(
            p=l["max_principal_display"], y=l["term_years"], i=l["interest_rate_pct"], d=l["target_dscr"]
        ))
    if "investment" in decisions:
        inv = decisions["investment"]
        p.append("### 投資判断（提案）\n- EV: **{ev}** / 時価総額: **{mc}** / 投資レンジ: **{chk}** / 魅力度: **{att}/5**（成長性: {g}） / 市場補正: x{mf:.2f}".format(
            ev=inv["ev_display"], mc=inv["market_cap_display"], chk=inv["recommended_check_size_display"],
            att=inv["attractiveness"], g=inv["growth_label"], mf=inv["market_factor"]
        ))
        # シナリオ抜粋
        if inv.get("scenarios"):
            line = " / ".join([f"{r['label']}: {r['check_size_display']}" for r in inv["scenarios"]])
            p.append(f"- シナリオ別投資レンジ: {line}")
    if market:
        p.append("### 市場根拠（日本語）\n" + (market.rationale or "—"))
    p.append("### 単位（検出）\n- ソース表記: {lab} / 乗数: x{mul:,}".format(lab=unit_info["source_label"], mul=unit_info["multiplier"]))
    return "\n".join(p)

# ---------- 解析本体 ----------
def analyze(files: List, company_name: str, industry_hint: str, currency_hint: str, market_notes: str,
            base_rate: float, want_credit: bool, want_loan: bool, want_invest: bool, debug: bool):
    try:
        client = get_client()
    except Exception as e:
        raise gr.Error(str(e))

    if not files: raise gr.Error("決算書ファイル（PDF/画像）を1つ以上アップロードしてください。")

    try:
        file_ids = []
        for f in files:
            fname, fbytes = _read_file_input(f)
            file_ids.append(upload_file_to_openai(client, fname, fbytes))
    except Exception as e:
        raise gr.Error(f"ファイルのアップロードに失敗しました: {e}")

    local_paths = [os.fspath(f) for f in files if isinstance(f, (str, bytes)) or hasattr(f, "__fspath__")]

    try:
        try:
            extract = extract_financials_from_files(client, file_ids, company_name or None, currency_hint or None,
                                                    model=VISION_MODEL, debug=debug, local_paths=local_paths)
        except TypeError:
            extract = extract_financials_from_files(client, file_ids, company_name or None, currency_hint or None,
                                                    model=VISION_MODEL, debug=debug)
    except Exception as e:
        raise gr.Error(f"LLM抽出に失敗しました: {e}")

    if company_name:  extract.company_name = company_name
    if industry_hint: extract.industry     = industry_hint

    unit_info = {"source_label": "不明", "multiplier": 1}
    try:
        if local_paths:
            mult, label = detect_unit_multiplier_from_paths(local_paths)
            unit_info = {"source_label": label, "multiplier": int(mult)}
            if mult and mult != 1: scale_extract_inplace(extract, mult)
    except Exception as e:
        if debug: print(f"[unit-detect] warning: {e}")

    ratios = compute_ratios(extract)

    decisions: Dict[str, Any] = {}
    if want_credit: decisions["credit"] = credit_decision(extract, ratios, POLICIES)
    if want_loan:   decisions["loan"]   = loan_decision(extract, ratios, base_rate or BASE_RATE, POLICIES)

    market_outlook: Optional[MarketOutlook] = None
    if want_invest:
        multiples: Optional[MultipleSuggestion] = None
        try:
            multiples = suggest_multiples_with_llm(client, TEXT_MODEL, extract.industry or industry_hint or "", "JP", debug)
        except Exception: multiples = None
        try:
            market_outlook = suggest_market_outlook_with_llm(client, TEXT_MODEL, extract.industry or industry_hint or "", market_notes or "", "JP", debug)
        except Exception: market_outlook = None
        decisions["investment"] = investment_decision(extract, ratios, POLICIES, multiples, market_outlook=market_outlook)

    report = build_report_dict(extract, ratios, decisions, unit_info=unit_info, market_outlook=market_outlook)
    report_json = json.dumps(report, ensure_ascii=False, indent=2)

    ts = datetime.utcnow().strftime("%Y%m%d-%H%M%S")
    data_dir = os.environ.get("HF_DATA_DIR", "/tmp")
    os.makedirs(data_dir, exist_ok=True)
    json_path = os.path.join(data_dir, f"report-{ts}.json")
    pdf_path  = os.path.join(data_dir, f"report-{ts}.pdf")
    with open(json_path, "w", encoding="utf-8") as f: f.write(report_json)
    export_pdf(report, pdf_path)

    summary_md = build_human_summary(extract, ratios, decisions, unit_info, market_outlook)
    return summary_md, json.loads(report_json), json_path, pdf_path

# ---------- UI ----------
def build_ui():
    with gr.Blocks(theme=gr.themes.Soft(), css="footer {visibility: hidden}") as demo:
        gr.Markdown("## 決算書→与信・融資・投資判断（HF + OpenAI）")
        with gr.Row():
            with gr.Column(scale=1):
                files = gr.File(label="決算書（PDF/JPG/PNG, 複数可）", file_types=[".pdf",".png",".jpg",".jpeg"], file_count="multiple", type="filepath")
                company_name = gr.Textbox(label="会社名（任意）")
                industry_hint = gr.Textbox(label="業種（任意）", placeholder="例：ソフトウェア、食品、物流 等")
                currency_hint = gr.Textbox(label="通貨（任意, 例: JPY, USD）")
                market_notes = gr.Textbox(label="市場拡大の期待（自由記述）", placeholder="例：勤怠管理×WFMの普及、BPO連携、新製品 等（日本語で）")
                base_rate = gr.Number(label="ベース金利（%/年）", value=BASE_RATE)
                with gr.Row():
                    want_credit = gr.Checkbox(label="与信判断", value=True)
                    want_loan = gr.Checkbox(label="融資判断", value=True)
                    want_invest = gr.Checkbox(label="投資判断", value=True)
                debug = gr.Checkbox(label="デバッグモード（厳格JSON/ログ）", value=False)
                run_btn = gr.Button("分析する", variant="primary")
            with gr.Column(scale=1):
                with gr.Tabs():
                    with gr.TabItem("概要（評価サマリ）"):
                        summary = gr.Markdown()
                    with gr.TabItem("詳細JSON"):
                        report = gr.JSON(label="詳細レポート（JSON）")
                    with gr.TabItem("ダウンロード"):
                        dl_json = gr.File(label="レポート（JSON）")
                        dl_pdf  = gr.File(label="レポート（PDF）")

        run_btn.click(
            analyze,
            inputs=[files, company_name, industry_hint, currency_hint, market_notes, base_rate, want_credit, want_loan, want_invest, debug],
            outputs=[summary, report, dl_json, dl_pdf],
        )
    return demo

if __name__ == "__main__":
    demo = build_ui()
    demo.launch(allowed_paths=["/tmp", "/mnt/data"])