(query: str, *, user_id: Optional[str] = None)
| 788 | |
| 789 | |
| 790 | def find_pending_direction_candidate(query: str, *, user_id: Optional[str] = None) -> Optional[Dict[str, Any]]: |
| 791 | normalized_query = _normalize_lookup_key(query) |
| 792 | if not normalized_query: |
| 793 | return None |
| 794 | |
| 795 | store = _load_pending_store() |
| 796 | for candidate in store.values(): |
| 797 | user_ids = candidate.get("user_ids", []) or [] |
| 798 | if user_id and user_ids and str(user_id) not in user_ids: |
| 799 | continue |
| 800 | candidate_terms = [ |
| 801 | candidate.get("candidate_key"), |
| 802 | candidate.get("name"), |
| 803 | candidate.get("name_cn"), |
| 804 | *(candidate.get("source_texts", []) or []), |
| 805 | ] |
| 806 | for term in candidate_terms: |
| 807 | if _normalize_lookup_key(term) == normalized_query: |
| 808 | return copy.deepcopy(candidate) |
| 809 | return None |
| 810 | |
| 811 | |
| 812 | def confirm_pending_direction_candidate(query: str, *, user_id: Optional[str] = None) -> Optional[Dict[str, Any]]: |
no test coverage detected