#!/usr/bin/env python3 """ Yoga 覆盖率 benchmark：对比当前 skill JSON 规则库与 PyJHora yoga.py。 目标： 1. 统计当前 references/yoga_rules.json 规则数量和分类分布 2. 若本机存在 PyJHora 安装，提取 jhora.horoscope.chart.yoga.py 中的唯一 Yoga 名称 3. 用名称归一化 + alias 表做近似覆盖匹配 4. 输出覆盖率、疑似缺失项和疑似重复项，作为后续补规则依据 说明： - 这是覆盖 benchmark，不判断每条规则的数学条件是否与 PyJHora 完全等价。 - PyJHora 大量函数是同一 Yoga 的编号变体；本脚本会把函数名归一为唯一 Yoga 名称。 """ from __future__ import annotations import argparse import json import os import re import sys from collections import Counter, defaultdict from pathlib import Path from typing import Iterable SKILL_ROOT = Path(__file__).resolve().parents[1] RULES_FILE = SKILL_ROOT / "references" / "yoga_rules.json" # 常见拼写/转写差异。左侧为 PyJHora normalized key，右侧为 skill 中的可能 normalized key。 ALIASES = { "sasa": {"shasha", "sasha"}, "maalavya": {"malavya"}, "maalaa": {"mala"}, "maala": {"mala"}, "gadaa": {"gada"}, "kedaara": {"kedara"}, "naukaa": {"nauka"}, "chaapa": {"chapa"}, "kaahala": {"kahala"}, "kalaanidhi": {"kalanidhi"}, "sreenaatha": {"sreenatha", "srinatha"}, "vasumathi": {"vasumati"}, "subha": {"shubha"}, "dharidhra": {"daridra"}, "budha_aditya": {"budhaditya", "nipuna"}, "chandra_mangala": {"chandra_mangal"}, "gaja_kesari": {"gajakesari"}, "harihara_brahma": {"harihara_brahma", "hari_hara_brahma"}, "siva": {"shiva"}, "vesi": {"vesi"}, "vosi": {"vosi"}, "sunaphaa": {"sunapha"}, "anaphaa": {"anapha"}, "ubhayachara": {"ubhayachari"}, "duradhara": {"duradhara"}, } SKIP_FUNCTIONS = { "get_yoga_resources", "get_yoga_details", "get_yoga_details_for_all_charts", "grihanasa_yoga_planet_positions", "are_lords_exchanged", } def normalize_name(name: str) -> str: """Normalize Yoga names for approximate matching.""" s = name.lower() s = re.sub(r"\([^)]*\)", " ", s) s = s.replace("&", " and ") s = re.sub(r"[^a-z0-9]+", "_", s) s = re.sub(r"_+", "_", s).strip("_") for suffix in ("_yoga", "_graha", "_classic", "_calculation", "_calc"): if s.endswith(suffix): s = s[: -len(suffix)] s = re.sub(r"_\d+$", "", s) return s def extract_skill_names(rules: list[dict]) -> tuple[set[str], dict[str, list[str]]]: keys: set[str] = set() reverse: dict[str, list[str]] = defaultdict(list) for rule in rules: raw_names = [rule.get("name", ""), rule.get("name_cn", ""), rule.get("id", "")] for raw in raw_names: key = normalize_name(raw) if not key: continue keys.add(key) reverse[key].append(rule.get("id", "?")) # 对英文名进一步拆出核心 token 组合，处理 “Gaja Kesari Yoga (Classic)” 这类名称 en = rule.get("name", "") key = normalize_name(en) if key: parts = [p for p in key.split("_") if p not in {"yoga", "classic", "from", "lord", "lords", "combination"}] if parts: compact = "_".join(parts[:3]) keys.add(compact) reverse[compact].append(rule.get("id", "?")) return keys, reverse def extract_external_benchmark_names(external_benchmark_yoga_file: Path) -> set[str]: content = external_benchmark_yoga_file.read_text(encoding="utf-8", errors="ignore") funcs = re.findall(r"^def ([a-zA-Z_][a-zA-Z0-9_]*)\(", content, re.MULTILINE) names: set[str] = set() for fn in funcs: if fn in SKIP_FUNCTIONS: continue if fn.startswith("_"): # PyJHora 的内部 calculation 函数大多也是 Yoga；保留 *_yoga*_calculation if "yoga" not in fn: continue if "_from_" in fn or "_get_" in fn or "_is_" in fn: continue base = fn base = re.sub(r"^_+", "", base) base = base.replace("_calculation", "").replace("_calc", "") base = re.sub(r"_\d+$", "", base) if base.endswith("_yoga"): base = base[:-5] key = normalize_name(base) if key and key not in SKIP_FUNCTIONS: names.add(key) return names def find_external_benchmark_yoga_file(explicit: str | None = None) -> Path | None: if explicit: p = Path(explicit).expanduser().resolve() return p if p.exists() else None candidates: list[Path] = [] for root in sys.path: if not root: continue p = Path(root) / "jhora" / "horoscope" / "chart" / "yoga.py" if p.exists(): candidates.append(p) # 常见 WorkBuddy isolated venv 位置 home = Path.home() candidates.extend(home.glob(".workbuddy/binaries/python/envs/*/lib/python*/site-packages/jhora/horoscope/chart/yoga.py")) return candidates[0] if candidates else None def covered(py_key: str, skill_keys: set[str]) -> bool: if py_key in skill_keys: return True compact_py = py_key.replace("_", "") compact_skill = {k.replace("_", "") for k in skill_keys} if compact_py in compact_skill: return True for alias in ALIASES.get(py_key, set()): if alias in skill_keys or alias.replace("_", "") in compact_skill: return True return False def main() -> int: parser = argparse.ArgumentParser(description="Benchmark Yoga coverage against PyJHora yoga.py") parser.add_argument("--pyjhora-yoga-file", help="Path to PyJHora jhora/horoscope/chart/yoga.py") parser.add_argument("--json", action="store_true", help="Output machine-readable JSON") parser.add_argument("--show-missing", type=int, default=60, help="How many missing items to display") args = parser.parse_args() data = json.loads(RULES_FILE.read_text(encoding="utf-8")) all_rules = data.get("rules", []) rules = [r for r in all_rules if r.get("enabled", True)] ids = [r.get("id") for r in all_rules] dup_ids = sorted([i for i, c in Counter(ids).items() if c > 1]) categories = Counter(r.get("category", "unknown") for r in rules) strength_values = Counter(r.get("strength", "?") for r in rules) skill_keys, skill_reverse = extract_skill_names(rules) external_benchmark_file = find_external_benchmark_yoga_file(args.external_benchmark_yoga_file) py_names: set[str] = set() missing: list[str] = [] coverage_pct = None if external_benchmark_file: py_names = extract_external_benchmark_names(external_benchmark_file) missing = sorted([name for name in py_names if not covered(name, skill_keys)]) coverage_pct = round((len(py_names) - len(missing)) / len(py_names) * 100, 2) if py_names else None result = { "rules_file": str(RULES_FILE), "total_rules": len(all_rules), "enabled_rules": len(rules), "disabled_rules": len(all_rules) - len(rules), "declared_total_rules": data.get("total_rules"), "declared_total_enabled_rules": data.get("total_enabled_rules"), "schema_version": data.get("schema_version"), "duplicate_ids": dup_ids, "categories": dict(categories.most_common()), "strength_values": dict(strength_values), "skill_normalized_name_keys": len(skill_keys), "external_benchmark_yoga_file": str(external_benchmark_file) if external_benchmark_file else None, "external_benchmark_unique_yoga_names": len(py_names) if external_benchmark_file else None, "matched_unique_yoga_names": (len(py_names) - len(missing)) if external_benchmark_file else None, "coverage_pct": coverage_pct, "missing_count": len(missing) if external_benchmark_file else None, "missing": missing, } if args.json: print(json.dumps(result, ensure_ascii=False, indent=2)) return 0 print("🧘 Yoga Coverage Benchmark") print("=" * 60) print(f"规则文件: {RULES_FILE}") print(f"Schema: {data.get('schema_version')}") print(f"规则总数: {len(all_rules)} (enabled={len(rules)}, disabled={len(all_rules) - len(rules)}, declared={data.get('total_rules')})") print(f"重复 ID: {len(dup_ids)}" + (f" → {dup_ids}" if dup_ids else " ✅")) print(f"Strength 值: {dict(strength_values)}") print("\n分类统计:") for cat, count in categories.most_common(): print(f" {cat:16s} {count:3d}") print("\nPyJHora 对比:") if not external_benchmark_file: print(" 未找到 PyJHora yoga.py；仅完成本地 JSON 统计。") print(" 可用 --pyjhora-yoga-file 指定路径。") else: print(f" yoga.py: {external_benchmark_file}") print(f" PyJHora 唯一 Yoga 名称: {len(py_names)}") print(f" 已匹配: {len(py_names) - len(missing)}") print(f" 疑似缺失: {len(missing)}") print(f" 名称覆盖率: {coverage_pct}%") if missing: print(f"\n疑似缺失 Top {min(args.show_missing, len(missing))}:") for i, name in enumerate(missing[: args.show_missing], 1): print(f" {i:3d}. {name}") if len(missing) > args.show_missing: print(f" ... 还有 {len(missing) - args.show_missing} 条") print("\n说明: 此 benchmark 为名称覆盖率，不等同于逐条数学逻辑等价验证。") return 0 if __name__ == "__main__": raise SystemExit(main())