Replace stored profiles with fresh role-specific seed profiles.
(
conn: sqlite3.Connection,
roles_meta: Dict[str, Any],
roles: Optional[Iterable[str]] = None,
)
| 177 | |
| 178 | |
| 179 | def reset_user_profiles( |
| 180 | conn: sqlite3.Connection, |
| 181 | roles_meta: Dict[str, Any], |
| 182 | roles: Optional[Iterable[str]] = None, |
| 183 | ) -> Dict[str, Dict[str, float]]: |
| 184 | """Replace stored profiles with fresh role-specific seed profiles.""" |
| 185 | all_roles = roles_meta.get("roles", {}) |
| 186 | target_roles = list(roles) if roles else list(all_roles.keys()) |
| 187 | |
| 188 | missing_roles = [role_name for role_name in target_roles if role_name not in all_roles] |
| 189 | if missing_roles: |
| 190 | raise ValueError(f"Unknown roles: {', '.join(missing_roles)}") |
| 191 | |
| 192 | cursor = conn.cursor() |
| 193 | cursor.execute("DELETE FROM profiles") |
| 194 | |
| 195 | seeded_profiles: Dict[str, Dict[str, float]] = {} |
| 196 | for role_name in target_roles: |
| 197 | role_data = all_roles[role_name] |
| 198 | profile = build_seed_profile(role_name, role_data) |
| 199 | cursor.execute( |
| 200 | """ |
| 201 | INSERT INTO profiles (user_id, profile_json, version, updated_at) |
| 202 | VALUES (?, ?, ?, CURRENT_TIMESTAMP) |
| 203 | """, |
| 204 | ( |
| 205 | profile["user_id"], |
| 206 | json.dumps(profile, ensure_ascii=False), |
| 207 | profile.get("version", "0.1"), |
| 208 | ), |
| 209 | ) |
| 210 | seeded_profiles[role_name] = dict(profile.get("core_directions") or {}) |
| 211 | |
| 212 | return seeded_profiles |
| 213 | |
| 214 | |
| 215 | def clear_tables(conn: sqlite3.Connection, table_names: Iterable[str]) -> None: |
no test coverage detected