Create unit_utils.py

core/unit_utils.py

@@ -0,0 +1,101 @@
+# core/unit_utils.py
+from __future__ import annotations
+import re
+from typing import Dict, Any, Tuple, List
+
+# 単位→換算（円基準）
+UNIT_SCALE = {
+    "円": 1.0,
+    "千円": 1e3,
+    "万円": 1e4,
+    "百万円": 1e6,
+    "千万円": 1e7,
+    "億円": 1e8,
+    "十億円": 1e10,
+    # 英語表現
+    "thousands of yen": 1e3,
+    "thousand yen": 1e3,
+    "in thousands of yen": 1e3,
+    "millions of yen": 1e6,
+    "in millions of yen": 1e6,
+    "jpy in thousands": 1e3,
+    "jpy in millions": 1e6,
+}
+
+# よく使うパターンを包括的に
+_PATTERNS = [
+    r"単位[：:\s]*?(千円|百万円|万円|円|千万円|億円|十億円)",
+    r"(?:Amounts?|Figures?)\s+in\s+(thousands|millions)\s+of\s+yen",
+    r"(?:JPY|YEN)\s+in\s+(thousands|millions)",
+    r"(?:in\s+)?(thousands|millions)\s+of\s+yen",
+]
+
+def detect_unit_scale(text: str) -> Tuple[float, str, List[str]]:
+    """
+    テキストから単位を推定し (scale, label, hits) を返す。
+    見つからない場合は (1.0, '円', [])。
+    複数ヒット時は多数決→先勝で決定。
+    """
+    hits: List[str] = []
+    t = text.lower()
+    for pat in _PATTERNS:
+        for m in re.finditer(pat, text, flags=re.IGNORECASE):
+            g = m.group(1).lower()
+            if g in ("千円", "百万円", "万円", "円", "千万円", "億円", "十億円"):
+                hits.append(g)
+            elif g in ("thousands", "thousand"):
+                hits.append("thousands of yen")
+            elif g in ("millions", "million"):
+                hits.append("millions of yen")
+
+    if not hits:
+        return 1.0, "円", []
+
+    # 多数決
+    from collections import Counter
+    cnt = Counter(hits)
+    label = cnt.most_common(1)[0][0]
+
+    # ラベル正規化（日本語優先）
+    if label in ("thousands of yen", "thousand yen", "in thousands of yen", "jpy in thousands"):
+        label = "千円"
+    if label in ("millions of yen", "in millions of yen", "jpy in millions"):
+        label = "百万円"
+
+    scale = UNIT_SCALE.get(label, 1.0)
+    return scale, label, hits
+
+def _scale_number(v: Any, scale: float) -> Any:
+    if v is None:
+        return None
+    try:
+        f = float(v)
+    except Exception:
+        return v
+    return f * scale
+
+def apply_unit_scale(fin: Dict[str, Any], scale: float) -> Dict[str, Any]:
+    """
+    抽出JSON（balance_sheet / income_statement / cash_flows の数値）に換算を適用。
+    比率などが混ざるケースを想定し、-1～1の狭い値はそのまま残す (誤変換防止)。
+    """
+    def maybe_scale(x):
+        if x is None:
+            return None
+        try:
+            f = float(x)
+        except Exception:
+            return x
+        # 典型的な比率やマージンらしき小さい値はスキップ
+        if -1.0 <= f <= 1.0 and scale > 1:
+            return f
+        return f * scale
+
+    out = {k: (v.copy() if isinstance(v, dict) else v) for k, v in fin.items()}
+    for sec in ("balance_sheet", "income_statement", "cash_flows"):
+        d = out.get(sec, {})
+        if isinstance(d, dict):
+            for k, v in d.items():
+                d[k] = maybe_scale(v)
+            out[sec] = d
+    return out