(
user_metadata: Dict[str, Dict[str, Any]],
roles: Dict[str, Dict[str, Any]],
episode_rows: Sequence[Dict[str, Any]],
)
| 601 | |
| 602 | |
| 603 | def build_user_states( |
| 604 | user_metadata: Dict[str, Dict[str, Any]], |
| 605 | roles: Dict[str, Dict[str, Any]], |
| 606 | episode_rows: Sequence[Dict[str, Any]], |
| 607 | ) -> Dict[str, UserKnowledgeEntityState]: |
| 608 | user_ids = sorted( |
| 609 | { |
| 610 | str(row.get("user_id") or "").strip() |
| 611 | for row in episode_rows |
| 612 | if str(row.get("user_id") or "").strip() |
| 613 | } |
| 614 | | set(user_metadata.keys()) |
| 615 | ) |
| 616 | states: Dict[str, UserKnowledgeEntityState] = {} |
| 617 | for user_id in user_ids: |
| 618 | meta = user_metadata.get(user_id, {"user_id": user_id}) |
| 619 | role_name = str(meta.get("role_name") or user_id.replace("user_", "")).strip() |
| 620 | profile = build_initial_profile(meta, roles.get(role_name, {})) |
| 621 | states[user_id] = UserKnowledgeEntityState( |
| 622 | profile_text=str(profile.get("profile_text") or ""), |
| 623 | profile_entities=list(profile.get("entity_terms") or []), |
| 624 | ) |
| 625 | return states |
| 626 | |
| 627 | |
| 628 | def clone_output_row( |
no test coverage detected