(
source_text: str,
*,
proposed_name: Optional[str] = None,
proposed_name_cn: Optional[str] = None,
confidence: float = 0.0,
user_id: Optional[str] = None,
reason: str = "",
)
| 741 | |
| 742 | |
| 743 | def upsert_pending_direction_candidate( |
| 744 | source_text: str, |
| 745 | *, |
| 746 | proposed_name: Optional[str] = None, |
| 747 | proposed_name_cn: Optional[str] = None, |
| 748 | confidence: float = 0.0, |
| 749 | user_id: Optional[str] = None, |
| 750 | reason: str = "", |
| 751 | ) -> Dict[str, Any]: |
| 752 | raw_source = str(source_text or "").strip() |
| 753 | if not raw_source: |
| 754 | raw_source = str(proposed_name_cn or proposed_name or "").strip() |
| 755 | candidate_key = normalize_direction_key(proposed_name or raw_source) |
| 756 | if not candidate_key: |
| 757 | candidate_key = normalize_direction_key(proposed_name_cn or raw_source) |
| 758 | if not candidate_key: |
| 759 | raise ValueError("pending direction candidate requires a non-empty source text") |
| 760 | |
| 761 | store = _load_pending_store() |
| 762 | existing = copy.deepcopy(store.get(candidate_key, {})) |
| 763 | created_at = existing.get("created_at") or _now_iso() |
| 764 | |
| 765 | source_texts = _dedupe_strings([*(existing.get("source_texts", []) or []), raw_source]) |
| 766 | user_ids = _dedupe_strings([*(existing.get("user_ids", []) or []), str(user_id or "").strip()]) |
| 767 | |
| 768 | candidate = { |
| 769 | "candidate_key": candidate_key, |
| 770 | "canonical_name": candidate_key, |
| 771 | "name": str(proposed_name or existing.get("name") or candidate_key.replace("-", " ").title()).strip(), |
| 772 | "name_cn": str(proposed_name_cn or existing.get("name_cn") or raw_source or candidate_key.replace("-", " ").title()).strip(), |
| 773 | "source_texts": source_texts, |
| 774 | "aliases": source_texts[:], |
| 775 | "paper_terms": source_texts[:], |
| 776 | "confidence": round(max(float(existing.get("confidence", 0.0)), float(confidence or 0.0)), 4), |
| 777 | "reason": str(reason or existing.get("reason") or "").strip(), |
| 778 | "created_at": created_at, |
| 779 | "last_seen_at": _now_iso(), |
| 780 | "seen_count": int(existing.get("seen_count", 0)) + 1, |
| 781 | "user_ids": user_ids, |
| 782 | "status": "pending", |
| 783 | } |
| 784 | |
| 785 | store[candidate_key] = candidate |
| 786 | _save_pending_store(store) |
| 787 | return candidate |
| 788 | |
| 789 | |
| 790 | def find_pending_direction_candidate(query: str, *, user_id: Optional[str] = None) -> Optional[Dict[str, Any]]: |
no test coverage detected