Collapse duplicate direction aliases into canonical keys while preserving unknown topics.
(raw_weights: Any)
| 187 | |
| 188 | |
| 189 | def _normalize_direction_weight_map(raw_weights: Any) -> Dict[str, float]: |
| 190 | """Collapse duplicate direction aliases into canonical keys while preserving unknown topics.""" |
| 191 | if not isinstance(raw_weights, dict): |
| 192 | return {} |
| 193 | |
| 194 | try: |
| 195 | direction_lexicon = importlib.import_module("config.direction_lexicon") |
| 196 | resolve_canonical_direction = getattr(direction_lexicon, "resolve_canonical_direction", None) |
| 197 | except Exception: |
| 198 | resolve_canonical_direction = None |
| 199 | |
| 200 | normalized: Dict[str, float] = {} |
| 201 | for raw_key, raw_weight in raw_weights.items(): |
| 202 | key_text = _collapse_whitespace(raw_key) |
| 203 | if not key_text: |
| 204 | continue |
| 205 | try: |
| 206 | weight = round(float(raw_weight or 0.0), 4) |
| 207 | except (TypeError, ValueError): |
| 208 | continue |
| 209 | |
| 210 | canonical_key = key_text |
| 211 | if callable(resolve_canonical_direction): |
| 212 | try: |
| 213 | resolved = resolve_canonical_direction(key_text, include_paper_terms=True) |
| 214 | except Exception: |
| 215 | resolved = None |
| 216 | canonical_key = str((resolved or {}).get("canonical_name") or key_text).strip() or key_text |
| 217 | |
| 218 | normalized[canonical_key] = round(max(float(normalized.get(canonical_key, 0.0)), weight), 4) |
| 219 | |
| 220 | return normalized |
| 221 | |
| 222 | |
| 223 | def _resolve_role_meta_path() -> Path: |
no test coverage detected