fix: keep other chart profiles locally on cloud sync failure

2026-07-20 16:56:46 +08:00
parent 94cf431b9e
commit ba7f962c71
4 changed files with 128 additions and 21 deletions
@@ -1,15 +1,81 @@
 #!/usr/bin/env python3
+"""Audit astrology answers for commercial safety/quality boundaries."""
+
 from __future__ import annotations
-import json, sys
-FORBIDDEN = ["一定发生", "保证结婚", "保证发财", "医疗诊断", "确诊", "签证保证", "exact_day_verified", "full_year_certainty"]
-REQUIRED_WHEN_TIMING = ["候选", "窗口", "边界"]
-def audit_answer(text: str) -> dict:
-    hits = [x for x in FORBIDDEN if x.lower() in text.lower()]
-    timing_missing = ("什么时候" in text or "几月" in text or "哪天" in text) and not any(x in text for x in REQUIRED_WHEN_TIMING)
-    return {"status": "pass" if not hits and not timing_missing else "fail", "forbidden_hits": hits, "timing_boundary_missing": timing_missing}
-def run(path: str | None = None) -> dict:
-    rows = json.load(open(path, encoding="utf-8")) if path else []
-    results = [audit_answer(str(r.get("answer", r))) for r in rows]
-    return {"scope": "answer_quality_audit", "status": "pass" if all(r["status"] == "pass" for r in results) else "fail", "results": results}
+
+import argparse
+import json
+from pathlib import Path
+from typing import Any
+
+
+FORBIDDEN = [
+    "一定发生",
+    "保证结婚",
+    "保证发财",
+    "医疗诊断",
+    "确诊",
+    "签证保证",
+    "exact_day_verified",
+    "full_year_certainty",
+]
+
+TIMING_TRIGGERS = ["什么时候", "几月", "哪天", "应期", "timing", "when", "exact date"]
+TIMING_BOUNDARIES = ["候选", "窗口", "边界", "exploratory_unvalidated", "未验证"]
+
+HEALTH_TRIGGERS = ["健康", "疾病", "医疗", "病", "health", "medical", "disease"]
+HEALTH_BOUNDARIES = ["非医疗", "不能诊断", "non-medical", "建议咨询医生", "not medical"]
+
+CASE_TRIGGERS = ["相似案例", "真实案例", "public case", "similar case"]
+CASE_BOUNDARIES = ["参考", "不是证明", "not proof", "product qa", "不能当作准确率"]
+
+METHOD_TRIGGERS = ["shadbala", "ashtakavarga", "kp", "流派", "方法差异"]
+METHOD_BOUNDARIES = ["流派", "方法", "来源", "variant", "provenance", "不能多数投票"]
+
+
+def _missing_boundary(text: str, triggers: list[str], boundaries: list[str]) -> bool:
+    low = text.lower()
+    return any(token.lower() in low for token in triggers) and not any(token.lower() in low for token in boundaries)
+
+
+def audit_answer(text: str) -> dict[str, Any]:
+    low = text.lower()
+    forbidden_hits = [token for token in FORBIDDEN if token.lower() in low]
+    checks = {
+        "timing_boundary_missing": _missing_boundary(text, TIMING_TRIGGERS, TIMING_BOUNDARIES),
+        "health_boundary_missing": _missing_boundary(text, HEALTH_TRIGGERS, HEALTH_BOUNDARIES),
+        "case_boundary_missing": _missing_boundary(text, CASE_TRIGGERS, CASE_BOUNDARIES),
+        "method_boundary_missing": _missing_boundary(text, METHOD_TRIGGERS, METHOD_BOUNDARIES),
+    }
+    status = "pass" if not forbidden_hits and not any(checks.values()) else "fail"
+    return {"status": status, "forbidden_hits": forbidden_hits, **checks}
+
+
+def _row_text(row: Any) -> str:
+    if isinstance(row, dict):
+        return str(row.get("answer") or row.get("text") or row.get("message") or row)
+    return str(row)
+
+
+def run(path: str | None = None) -> dict[str, Any]:
+    rows = json.loads(Path(path).read_text(encoding="utf-8")) if path else []
+    results = [audit_answer(_row_text(row)) for row in rows]
+    return {
+        "scope": "answer_quality_audit",
+        "status": "pass" if all(row["status"] == "pass" for row in results) else "fail",
+        "answer_count": len(results),
+        "results": results,
+        "boundary": "Text quality gate only; does not validate chart accuracy.",
+    }
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("path", nargs="?")
+    args = parser.parse_args()
+    print(json.dumps(run(args.path), ensure_ascii=False, indent=2, sort_keys=True))
+    return 0
+
+
 if __name__ == "__main__":
-    print(json.dumps(run(sys.argv[1] if len(sys.argv) > 1 else None), ensure_ascii=False, indent=2))
+    raise SystemExit(main())