import requests
import json
import re
from .user import User
from .school import School
from .community import Community
from .portfolio import Portfolio, Banner
from .exceptions import sluggardException, InvalidJsonResponse
from .utils import get_value_deep_key
class Sluggard:
loads = json.loads
dumps = json.dumps
def __init__(self, csrf_token="", cookie="", tenant="", xsrf=""):
self.last_response = {}
self.url = "https://api.simulise.com/graphql"
self.tenant_url = f'https://{tenant}.simulise.com'
self.headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; rv:124.0) Gecko/20100101 Firefox/124.0",
"Accept": "application/json",
"Accept-Language": "en-US,en;q=0.5",
"Content-Type": "application/json",
"Content-Disposition": None,
"x-csrf-token": csrf_token,
"x-requested-with": "XMLHttpRequest",
"Cookie": f'simulise_session_production={cookie}; simulise_legacy_session_production={cookie}; XSRF-TOKEN={xsrf}',
"DNT": "1"
}
self.session = requests.Session()
def school_name_search(self, name):
data = {
"query": "query authenticationRealmSearch($query: String!) {\n authenticationRealmSearch(query: $query) {\n ...authenticationRealmFields __typename }\n}\n fragment authenticationRealmFields on AuthenticationRealm {\n id key sso idp {\n name logo __typename }\n logo name i18n {\n lang type __typename }\n alias permalink __typename }",
"variables": {"query": name}
}
try:
response = self.session.post(self.url, json=data)
response_json = response.json()
if 'data' in response_json and 'authenticationRealmSearch' in response_json['data']:
if not response_json['data']['authenticationRealmSearch']:
raise InvalidJsonResponse("Empty response", "Authentication Realm search returned empty array")
for info in response_json['data']['authenticationRealmSearch']:
school_info = {
'id': info['id'],
'key': info['key'],
'sso': info['sso'],
'idp_name': info['idp']['name'],
'idp_logo': info['idp']['logo'],
'school_logo': info['logo'],
'school_name': info['name'],
'school_permalink': info['permalink']
}
return School.search_from_dict(school_info)
else:
raise InvalidJsonResponse("Invalid response format", "Response does not contain expected data structure")
except requests.exceptions.RequestException as e:
raise sluggardException("Request failed") from e
def add__reflect(self, assignedassignment_id, user_number):
data = {"query":"mutation assignedAssignmentOtherReflectionInvite($input: AssignedAssignmentOtherReflectionInviteInput!) {\n response: assignedAssignmentOtherReflectionInvite(input: $input) {\n assignedAssignment {\n id\n otherReflectionUsers {\n ...assignmentAssignedUserFields\n ...userAutocompleteFieldsFragment\n __typename\n }\n reflectionIsOptionalAndUncompleted\n ...assignedAssignmentOtherReflectionInvitationsFields\n __typename\n }\n ...mutationFields\n __typename\n }\n}\n\nfragment assignmentAssignedUserFields on User {\n id\n name\n avatar\n username\n permalink\n __typename\n}\n\nfragment userAutocompleteFieldsFragment on User {\n autocompletion {\n id\n text\n icon\n value\n avatar\n caption\n __typename\n }\n __typename\n}\n\nfragment assignedAssignmentOtherReflectionInvitationsFields on AssignedAssignment {\n otherReflectionInvitations {\n id\n user {\n ...assignmentAssignedUserFields\n __typename\n }\n createdAt\n isDeclined\n declinedAt\n submittedAt\n isSubmitted\n isProgressed\n __typename\n }\n __typename\n}\n\nfragment mutationFields on MutationPayload {\n status {\n success\n errors {\n name\n messages\n __typename\n }\n __typename\n }\n __typename\n}","variables":{"input":{"users":["310397"],"assignmentAssigned":"28fd494a-1eeb-4bc3-86df-2284da036a33"}}}
try:
response = self.session.post(self.url, json=data)
response_json = response.json()
print(response_json)
print(data)
except requests.exceptions.RequestException as e:
raise sluggardException("Request failed") from e
def discover_user(self, user):
data = {"query": f"query dashboardDiscoverUser($id: String!) {{ user(id: $id) {{ id username primaryRole groupsActive {{ ...dashboardUserTileGroupsFragment __typename }} dashboardTiles {{ id key title isHidden canBeHidden __typename }} __typename }} viewer {{ id tenant {{ id settings {{ goalsLargerOnDashboard __typename }} __typename }} username primaryRole groupsActive {{ ...dashboardUserTileGroupsFragment __typename }} externalInvitationCount __typename }} }} fragment dashboardUserTileGroupsFragment on Group {{ __typename ... on Entity {{ id __typename }} ... on WorkGroup {{ id name icon permalink __typename }} ... on SchoolGroup {{ id name icon path permalink __typename }} ... on VirtualGroup {{ id name icon permalink __typename }} }}" , "variables": {"id": user}}
try:
response = self.session.post(self.url, headers=self.headers, json=data)
response_json = response.json()
return User.discover_from_dict(response_json)
except requests.exceptions.RequestException as e:
raise sluggardException("Request failed") from e
except AttributeError as e:
raise sluggardException("Error accessing 'groupsActive' attribute in user_info") from e
def get_community_recent_post(self, user):
data = {"query": "query dashboardTileCommunityActivity($id: String!) {\n activity: user(id: $id) {\n id\n portfolioRecentCommunityItems {\n id\n icon\n title\n author {\n id\n name\n avatar\n username\n portfolioPermalink\n __typename\n }\n permalink\n createdAt\n __typename\n }\n __typename\n }\n}" , "variables": {"id": user}}
try:
response = self.session.post(self.url, headers=self.headers, json=data)
response_json = response.json()
return Community.from_dict(response_json)
except requests.exceptions.RequestException as e:
raise sluggardException("Request failed") from e
def get_community_posts(self):
response = self.session.get(self.tenant_url + "/community/stream/filter?combine=false", headers=self.headers)
response_json = response.json()
pattern_activity_id = r'data-activity-id="(\d+)"'
pattern_real_id = r'data-activity-real-id="([a-f0-9\-]+)"'
pattern_title = r'\s*(.*?)'
pattern_title2 = r''
pattern_author_name = r'\s*(.*?)\s*'
pattern_timestamp = r'