diff --git a/frontend/src/app/page.tsx b/frontend/src/app/page.tsx index ee16cfc3..b7d83293 100644 --- a/frontend/src/app/page.tsx +++ b/frontend/src/app/page.tsx @@ -1445,12 +1445,13 @@ export default function Home() { profile: nextProfile, updatedAt: timestamp(), }; + let cloudSaved = false; try { record = await saveCloudChartProfile(record); - } catch (caught) { - setProfileNotice(""); - setAccountError(friendlyError(caught instanceof Error ? caught.message : "云端星盘保存失败，请稍后重试。")); - return; + cloudSaved = true; + } catch { + setProfileNotice("已保存到本地星盘库；云端同步失败，稍后会继续使用本地记录。"); + setAccountError(""); } setChartLibrary((current) => { const next = [...upsertSelfChart(current, profile), record]; @@ -1458,8 +1459,10 @@ export default function Home() { return next; }); setOtherProfileDraft(emptyProfile); - setAccountError(""); - setProfileNotice("已保存到云端星盘库。"); + if (cloudSaved) { + setAccountError(""); + setProfileNotice("已保存到云端星盘库。"); + } } async function deleteOtherChart(recordId: string) { diff --git a/frontend/tests/chart-library-other-profile.test.ts b/frontend/tests/chart-library-other-profile.test.ts index 0f0130fb..e10e2d2c 100644 --- a/frontend/tests/chart-library-other-profile.test.ts +++ b/frontend/tests/chart-library-other-profile.test.ts @@ -10,9 +10,12 @@ test("other chart saves do not require the owner's rectification state", () => { assert.doesNotMatch(source, /saveOtherChart[\s\S]{0,500}missingProfileStep\(nextProfile\)/); }); -test("other chart mutations acknowledge only confirmed cloud writes", () => { - assert.match(source, /await saveCloudChartProfile\(record\);[\s\S]{0,500}已保存到云端星盘库/); - assert.doesNotMatch(source, /saveOtherChart[\s\S]{0,500}catch\s*\{[\s\S]{0,500}已添加到星盘库/); +test("other chart save falls back to local library when cloud sync fails", () => { + assert.match(source, /let cloudSaved = false/); + assert.match(source, /record = await saveCloudChartProfile\(record\);[\s\S]{0,120}cloudSaved = true/); + assert.match(source, /catch\s*\{[\s\S]{0,300}已保存到本地星盘库；云端同步失败/); + assert.match(source, /localStorage\.setItem\(chartLibraryStorageKey\(accountId\), JSON\.stringify\(next\)\)/); + assert.match(source, /if \(cloudSaved\)[\s\S]{0,180}已保存到云端星盘库/); assert.match(source, /async function deleteOtherChart[\s\S]{0,500}await deleteCloudChartProfile\(recordId\)/); assert.doesNotMatch(source, /deleteOtherChart[\s\S]{0,500}void deleteCloudChartProfile/); }); diff --git a/scripts/answer_quality_audit.py b/scripts/answer_quality_audit.py index 97b3d5d5..e1f00c63 100644 --- a/scripts/answer_quality_audit.py +++ b/scripts/answer_quality_audit.py @@ -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()) diff --git a/tests/test_answer_quality_audit.py b/tests/test_answer_quality_audit.py new file mode 100644 index 00000000..176a7c4a --- /dev/null +++ b/tests/test_answer_quality_audit.py @@ -0,0 +1,35 @@ +import json +import subprocess +from pathlib import Path + +from scripts.answer_quality_audit import audit_answer + + +ROOT = Path(__file__).resolve().parents[1] + + +def test_answer_quality_audit_blocks_absolute_claims() -> None: + result = audit_answer("你一定发生婚姻，并且保证发财。") + assert result["status"] == "fail" + assert "一定发生" in result["forbidden_hits"] + + +def test_answer_quality_audit_requires_timing_health_case_and_method_boundaries() -> None: + assert audit_answer("什么时候结婚？今年几月。")["timing_boundary_missing"] is True + assert audit_answer("健康看这里。")["health_boundary_missing"] is True + assert audit_answer("相似案例说明这个预测正确。")["case_boundary_missing"] is True + assert audit_answer("Shadbala 和 Ashtakavarga 结果不同。")["method_boundary_missing"] is True + + +def test_answer_quality_audit_passes_when_boundaries_are_present(tmp_path: Path) -> None: + rows = [ + { + "answer": "应期只能给候选窗口，claim_status=exploratory_unvalidated；健康为非医疗表达；相似案例只是参考，不是证明；Shadbala 是流派/方法差异。" + } + ] + path = tmp_path / "answers.json" + path.write_text(json.dumps(rows, ensure_ascii=False), encoding="utf-8") + data = json.loads(subprocess.check_output(["python3", "scripts/answer_quality_audit.py", str(path)], cwd=ROOT, text=True)) + assert data["scope"] == "answer_quality_audit" + assert data["status"] == "pass" + assert data["answer_count"] == 1