|
| 1 | +""" |
| 2 | +Webex Teams management tools. |
| 3 | +""" |
| 4 | +from typing import Optional, Dict, Any |
| 5 | +from .common import get_webex_api, create_error_response, create_success_response, WebexErrorCodes, WebexTokenMissingError |
| 6 | + |
| 7 | + |
| 8 | +def _map_exception_to_error(e: Exception) -> Dict[str, Any]: |
| 9 | + if isinstance(e, WebexTokenMissingError): |
| 10 | + return create_error_response(WebexErrorCodes.UNAUTHORIZED, str(e)) |
| 11 | + error_str = str(e).lower() |
| 12 | + if 'unauthorized' in error_str or 'invalid token' in error_str: |
| 13 | + return create_error_response(WebexErrorCodes.UNAUTHORIZED, |
| 14 | + "Invalid or expired bot token.") |
| 15 | + if 'not found' in error_str or '404' in error_str: |
| 16 | + return create_error_response(WebexErrorCodes.NOT_FOUND, |
| 17 | + "Team or membership not found.") |
| 18 | + if 'forbidden' in error_str or '403' in error_str: |
| 19 | + return create_error_response(WebexErrorCodes.FORBIDDEN, |
| 20 | + "Bot lacks permission for this operation. " |
| 21 | + "Ensure the bot has moderator privileges.") |
| 22 | + if 'rate limit' in error_str or 'too many requests' in error_str: |
| 23 | + return create_error_response(WebexErrorCodes.RATE_LIMITED, |
| 24 | + "API rate limit exceeded. Please retry after delay.", |
| 25 | + temporary=True, retry_after_seconds=60) |
| 26 | + if 'network' in error_str or 'connection' in error_str: |
| 27 | + return create_error_response(WebexErrorCodes.NETWORK_ERROR, |
| 28 | + "Network connectivity issue. Please retry.", |
| 29 | + temporary=True, retry_after_seconds=30) |
| 30 | + return create_error_response(WebexErrorCodes.WEBEX_API_ERROR, |
| 31 | + f"Webex API error: {e}", |
| 32 | + temporary=True, details={'original_error': str(e)}) |
| 33 | + |
| 34 | + |
| 35 | +def _team_to_dict(t) -> Dict[str, Any]: |
| 36 | + d: Dict[str, Any] = { |
| 37 | + 'id': t.id, |
| 38 | + 'name': t.name, |
| 39 | + 'creatorId': t.creatorId, |
| 40 | + 'created': t.created, |
| 41 | + } |
| 42 | + if getattr(t, 'description', None): |
| 43 | + d['description'] = t.description |
| 44 | + return d |
| 45 | + |
| 46 | + |
| 47 | +def _team_membership_to_dict(m) -> Dict[str, Any]: |
| 48 | + d: Dict[str, Any] = { |
| 49 | + 'id': m.id, |
| 50 | + 'teamId': m.teamId, |
| 51 | + 'personId': m.personId, |
| 52 | + 'personEmail': m.personEmail, |
| 53 | + 'personDisplayName': m.personDisplayName, |
| 54 | + 'personOrgId': m.personOrgId, |
| 55 | + 'isModerator': m.isModerator, |
| 56 | + 'created': m.created, |
| 57 | + } |
| 58 | + return d |
| 59 | + |
| 60 | + |
| 61 | +def list_webex_teams( |
| 62 | + display_name: Optional[str] = None, |
| 63 | + max_results: Optional[int] = None |
| 64 | +) -> Dict[str, Any]: |
| 65 | + """ |
| 66 | + List Webex teams the authenticated bot belongs to. |
| 67 | +
|
| 68 | + Args: |
| 69 | + display_name: Filter teams by display name (optional) |
| 70 | + max_results: Maximum number of teams to return (default 100, max 1000) |
| 71 | +
|
| 72 | + Returns: |
| 73 | + Standardized response dictionary with success/error information |
| 74 | + """ |
| 75 | + try: |
| 76 | + params: Dict[str, Any] = {} |
| 77 | + if display_name: |
| 78 | + params['displayName'] = display_name |
| 79 | + if max_results: |
| 80 | + params['max'] = max_results |
| 81 | + |
| 82 | + teams = [_team_to_dict(t) for t in get_webex_api().teams.list(**params)] |
| 83 | + return create_success_response( |
| 84 | + data={'teams': teams}, |
| 85 | + metadata={'count': len(teams), 'filters_applied': params} |
| 86 | + ) |
| 87 | + except Exception as e: |
| 88 | + return _map_exception_to_error(e) |
| 89 | + |
| 90 | + |
| 91 | +def create_webex_team( |
| 92 | + name: str, |
| 93 | + description: Optional[str] = None |
| 94 | +) -> Dict[str, Any]: |
| 95 | + """ |
| 96 | + Create a new Webex team. |
| 97 | +
|
| 98 | + Args: |
| 99 | + name: Name of the team (required) |
| 100 | + description: Description of the team (optional) |
| 101 | +
|
| 102 | + Returns: |
| 103 | + Standardized response dictionary with success/error information |
| 104 | + """ |
| 105 | + try: |
| 106 | + params: Dict[str, Any] = {'name': name} |
| 107 | + if description: |
| 108 | + params['description'] = description |
| 109 | + |
| 110 | + team = get_webex_api().teams.create(**params) |
| 111 | + return create_success_response( |
| 112 | + data={'team': _team_to_dict(team)}, |
| 113 | + metadata={'operation': 'create_team', 'parameters_used': params} |
| 114 | + ) |
| 115 | + except Exception as e: |
| 116 | + return _map_exception_to_error(e) |
| 117 | + |
| 118 | + |
| 119 | +def get_webex_team(team_id: str) -> Dict[str, Any]: |
| 120 | + """ |
| 121 | + Get detailed information about a specific Webex team. |
| 122 | +
|
| 123 | + Args: |
| 124 | + team_id: Team ID to get details for (required) |
| 125 | +
|
| 126 | + Returns: |
| 127 | + Standardized response dictionary with success/error information |
| 128 | + """ |
| 129 | + try: |
| 130 | + team = get_webex_api().teams.get(teamId=team_id) |
| 131 | + return create_success_response( |
| 132 | + data={'team': _team_to_dict(team)}, |
| 133 | + metadata={'team_id': team_id} |
| 134 | + ) |
| 135 | + except Exception as e: |
| 136 | + return _map_exception_to_error(e) |
| 137 | + |
| 138 | + |
| 139 | +def update_webex_team( |
| 140 | + team_id: str, |
| 141 | + name: Optional[str] = None, |
| 142 | + description: Optional[str] = None |
| 143 | +) -> Dict[str, Any]: |
| 144 | + """ |
| 145 | + Update an existing Webex team's name and/or description. |
| 146 | +
|
| 147 | + Args: |
| 148 | + team_id: Team ID to update (required) |
| 149 | + name: New name for the team (optional) |
| 150 | + description: New description for the team (optional) |
| 151 | +
|
| 152 | + Returns: |
| 153 | + Standardized response dictionary with success/error information |
| 154 | + """ |
| 155 | + try: |
| 156 | + if name is None and description is None: |
| 157 | + return create_error_response( |
| 158 | + error_code=WebexErrorCodes.INVALID_ARGUMENTS, |
| 159 | + message="Must specify at least one field to update (name or description)" |
| 160 | + ) |
| 161 | + |
| 162 | + # The SDK requires name; fetch current value if not provided. |
| 163 | + if name is None: |
| 164 | + current = get_webex_api().teams.get(teamId=team_id) |
| 165 | + name = current.name |
| 166 | + |
| 167 | + params: Dict[str, Any] = {'name': name} |
| 168 | + if description is not None: |
| 169 | + params['description'] = description |
| 170 | + |
| 171 | + team = get_webex_api().teams.update(teamId=team_id, **params) |
| 172 | + return create_success_response( |
| 173 | + data={'team': _team_to_dict(team)}, |
| 174 | + metadata={'operation': 'update_team', 'team_id': team_id, |
| 175 | + 'parameters_used': params} |
| 176 | + ) |
| 177 | + except Exception as e: |
| 178 | + return _map_exception_to_error(e) |
| 179 | + |
| 180 | + |
| 181 | +def delete_webex_team(team_id: str) -> Dict[str, Any]: |
| 182 | + """ |
| 183 | + Delete a Webex team. |
| 184 | +
|
| 185 | + Args: |
| 186 | + team_id: Team ID to delete (required) |
| 187 | +
|
| 188 | + Returns: |
| 189 | + Standardized response dictionary with success/error information |
| 190 | + """ |
| 191 | + try: |
| 192 | + if not team_id: |
| 193 | + return create_error_response( |
| 194 | + error_code=WebexErrorCodes.INVALID_ARGUMENTS, |
| 195 | + message="team_id is required" |
| 196 | + ) |
| 197 | + get_webex_api().teams.delete(teamId=team_id) |
| 198 | + return create_success_response( |
| 199 | + data={'deleted': True, 'team_id': team_id}, |
| 200 | + metadata={'operation': 'delete_team'} |
| 201 | + ) |
| 202 | + except Exception as e: |
| 203 | + return _map_exception_to_error(e) |
| 204 | + |
| 205 | + |
| 206 | +def list_webex_team_memberships( |
| 207 | + team_id: str, |
| 208 | + max_results: Optional[int] = None |
| 209 | +) -> Dict[str, Any]: |
| 210 | + """ |
| 211 | + List members of a Webex team. |
| 212 | +
|
| 213 | + Args: |
| 214 | + team_id: Team ID to list members for (required) |
| 215 | + max_results: Maximum number of memberships to return (default 100, max 1000) |
| 216 | +
|
| 217 | + Returns: |
| 218 | + Standardized response dictionary with success/error information |
| 219 | + """ |
| 220 | + try: |
| 221 | + params: Dict[str, Any] = {'teamId': team_id} |
| 222 | + if max_results: |
| 223 | + params['max'] = max_results |
| 224 | + |
| 225 | + memberships = [_team_membership_to_dict(m) for m in get_webex_api().team_memberships.list(**params)] |
| 226 | + return create_success_response( |
| 227 | + data={'memberships': memberships}, |
| 228 | + metadata={'count': len(memberships), 'team_id': team_id} |
| 229 | + ) |
| 230 | + except Exception as e: |
| 231 | + return _map_exception_to_error(e) |
| 232 | + |
| 233 | + |
| 234 | +def add_webex_team_membership( |
| 235 | + team_id: str, |
| 236 | + person_id: Optional[str] = None, |
| 237 | + person_email: Optional[str] = None, |
| 238 | + is_moderator: Optional[bool] = None |
| 239 | +) -> Dict[str, Any]: |
| 240 | + """ |
| 241 | + Add a person to a Webex team. |
| 242 | +
|
| 243 | + Args: |
| 244 | + team_id: Team ID to add person to (required) |
| 245 | + person_id: Person ID to add (use this OR person_email) |
| 246 | + person_email: Person email to add (use this OR person_id) |
| 247 | + is_moderator: Whether to make the person a team moderator (optional) |
| 248 | +
|
| 249 | + Returns: |
| 250 | + Standardized response dictionary with success/error information |
| 251 | + """ |
| 252 | + try: |
| 253 | + if not (person_id or person_email): |
| 254 | + return create_error_response( |
| 255 | + error_code=WebexErrorCodes.INVALID_ARGUMENTS, |
| 256 | + message="Must specify either person_id or person_email", |
| 257 | + details={"required_one_of": ["person_id", "person_email"]} |
| 258 | + ) |
| 259 | + |
| 260 | + params: Dict[str, Any] = {'teamId': team_id} |
| 261 | + if person_id: |
| 262 | + params['personId'] = person_id |
| 263 | + elif person_email: |
| 264 | + params['personEmail'] = person_email |
| 265 | + if is_moderator is not None: |
| 266 | + params['isModerator'] = is_moderator |
| 267 | + |
| 268 | + membership = get_webex_api().team_memberships.create(**params) |
| 269 | + return create_success_response( |
| 270 | + data={'membership': _team_membership_to_dict(membership)}, |
| 271 | + metadata={'operation': 'add_team_membership', 'parameters_used': params} |
| 272 | + ) |
| 273 | + except Exception as e: |
| 274 | + return _map_exception_to_error(e) |
| 275 | + |
| 276 | + |
| 277 | +def delete_webex_team_membership(membership_id: str) -> Dict[str, Any]: |
| 278 | + """ |
| 279 | + Remove a person from a Webex team by deleting their team membership. |
| 280 | +
|
| 281 | + Args: |
| 282 | + membership_id: Team membership ID to delete (required). Use |
| 283 | + list_webex_team_memberships to find the membership ID. |
| 284 | +
|
| 285 | + Returns: |
| 286 | + Standardized response dictionary with success/error information |
| 287 | + """ |
| 288 | + try: |
| 289 | + if not membership_id: |
| 290 | + return create_error_response( |
| 291 | + error_code=WebexErrorCodes.INVALID_ARGUMENTS, |
| 292 | + message="membership_id is required" |
| 293 | + ) |
| 294 | + get_webex_api().team_memberships.delete(membershipId=membership_id) |
| 295 | + return create_success_response( |
| 296 | + data={'deleted': True, 'membership_id': membership_id}, |
| 297 | + metadata={'operation': 'delete_team_membership'} |
| 298 | + ) |
| 299 | + except Exception as e: |
| 300 | + return _map_exception_to_error(e) |
0 commit comments