commit 0422e09400443de0636c02a7e236f8646a6baba4 Author: Marcin Kowalicki Date: Thu Oct 22 17:40:30 2020 +0200 initial commit diff --git a/__init__.py b/__init__.py new file mode 100644 index 0000000..f1e542c --- /dev/null +++ b/__init__.py @@ -0,0 +1,69 @@ +from librus_tricks import exceptions +from librus_tricks.auth import authorizer, load_json as __load_json +from librus_tricks.classes import * +from librus_tricks.core import SynergiaClient + +__name__ = 'librus_tricks' +__title__ = 'librus_tricks' +__author__ = 'Krystian Postek' +__version__ = '0.8.1' + + +def create_session(email, password, fetch_first=True, **kwargs): + """ + Używaj tego tylko kiedy hasło do Portal Librus jest takie samo jako do Synergii. + + :param email: str + :param password: str + :param fetch_first: bool or int + :rtype: librus_tricks.core.SynergiaClient + :return: obiekt lub listę obiektów z sesjami + """ + if fetch_first is True: + user = authorizer(email, password)[0] + session = SynergiaClient(user, **kwargs) + elif fetch_first is False: + users = authorizer(email, password) + sessions = [SynergiaClient(user, **kwargs) for user in users] + return sessions + else: + user = authorizer(email, password)[fetch_first] + session = SynergiaClient(user, **kwargs) + + return session + + +def use_json(file=None, **kwargs): + if file is None: + from glob import glob + jsons = glob('*.json') + + if jsons.__len__() == 0: + raise FileNotFoundError('Nie znaleziono zapisanych sesji') + if jsons.__len__() > 1: + raise FileExistsError('Zaleziono za dużo zapisanych sesji') + + user = __load_json(open(jsons[0], 'r')) + else: + user = __load_json(file) + session = SynergiaClient(user, **kwargs) + session.get('Me') + return session + + +def minified_login(email, password, **kwargs): + import logging + import warnings + + warnings.warn(exceptions.SecurityWarning('Using minified_login in production environment is REAL SECURITY ISSUE!')) + try: + logging.debug('Trying to use json file to create session') + session = use_json(**kwargs) + logging.debug('Created session using json file') + except Exception: + logging.debug('Switching to regular http auth') + session = create_session(email, password, **kwargs) + logging.debug('Created session using http auth') + + session.user.dump_credentials() + return session diff --git a/__pycache__/__init__.cpython-36.pyc b/__pycache__/__init__.cpython-36.pyc new file mode 100644 index 0000000..26a2ca3 Binary files /dev/null and b/__pycache__/__init__.cpython-36.pyc differ diff --git a/__pycache__/auth.cpython-36.pyc b/__pycache__/auth.cpython-36.pyc new file mode 100644 index 0000000..c8fb67a Binary files /dev/null and b/__pycache__/auth.cpython-36.pyc differ diff --git a/__pycache__/cache.cpython-36.pyc b/__pycache__/cache.cpython-36.pyc new file mode 100644 index 0000000..05e5358 Binary files /dev/null and b/__pycache__/cache.cpython-36.pyc differ diff --git a/__pycache__/classes.cpython-36.pyc b/__pycache__/classes.cpython-36.pyc new file mode 100644 index 0000000..d639707 Binary files /dev/null and b/__pycache__/classes.cpython-36.pyc differ diff --git a/__pycache__/core.cpython-36.pyc b/__pycache__/core.cpython-36.pyc new file mode 100644 index 0000000..9aa435f Binary files /dev/null and b/__pycache__/core.cpython-36.pyc differ diff --git a/__pycache__/exceptions.cpython-36.pyc b/__pycache__/exceptions.cpython-36.pyc new file mode 100644 index 0000000..91ade1f Binary files /dev/null and b/__pycache__/exceptions.cpython-36.pyc differ diff --git a/__pycache__/messages.cpython-36.pyc b/__pycache__/messages.cpython-36.pyc new file mode 100644 index 0000000..1df52ba Binary files /dev/null and b/__pycache__/messages.cpython-36.pyc differ diff --git a/__pycache__/tools.cpython-36.pyc b/__pycache__/tools.cpython-36.pyc new file mode 100644 index 0000000..c6d2584 Binary files /dev/null and b/__pycache__/tools.cpython-36.pyc differ diff --git a/auth.py b/auth.py new file mode 100644 index 0000000..a6d4696 --- /dev/null +++ b/auth.py @@ -0,0 +1,210 @@ +import logging +from datetime import datetime, timedelta + +import requests +from bs4 import BeautifulSoup + +from .exceptions import * + +# Some globals +REDIRURL = 'http://localhost/bar' +LOGINURL = 'https://portal.librus.pl/rodzina/login/action' +OAUTHURL = 'https://portal.librus.pl/oauth2/access_token' +SYNERGIAAUTHURL = 'https://portal.librus.pl/api/v2/SynergiaAccounts' +FRESHURL = 'https://portal.librus.pl/api/v2/SynergiaAccounts/fresh/{login}' +CLIENTID = '6XPsKf10LPz1nxgHQLcvZ1KM48DYzlBAhxipaXY8' +LIBRUSLOGINURL = f'https://portal.librus.pl/oauth2/authorize?client_id={CLIENTID}&redirect_uri={REDIRURL}&response_type=code' +# User agents +XIAOMI_USERAGENT = 'Mozilla/5.0 (Linux; Android 9; Mi A1 Build/PQ3B.190801.002; wv) AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/79.0.3921.2 Mobile Safari/537.36LibrusMobileApp' +IPHONE_USERAGENT = 'Mozilla/5.0 (iPhone; CPU iPhone OS 13_1 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) CriOS/77.0.3865.103 Mobile/15E148 Safari/605.1LibrusMobileApp' +GOOGLEBOT_USERAGENT = 'Googlebot/2.1 (+http://www.google.com/bot.html)LibrusMobileApp' + +class SynergiaUser: + """ + Obiekt zawierający dane do tworzenia sesji + """ + + def __init__(self, user_dict, root_token, revalidation_token, exp_in): + self.token = user_dict['accessToken'] + self.refresh_token = revalidation_token + self.root_token = root_token + self.name, self.last_name = user_dict['studentName'].split(' ', maxsplit=1) + self.login = user_dict['login'] + self.uid = user_dict['id'] + self.expires_in = datetime.now() + timedelta(seconds=exp_in) + + def __repr__(self): + return f'' + + def __str__(self): + return f'{self.name} {self.last_name}' + + def revalidate_root(self): + """ + Aktualizuje token do Portalu Librus. + """ + auth_session = requests.session() + new_tokens = auth_session.post( + OAUTHURL, + data={ + 'grant_type': 'refresh_token', + 'refresh_token': self.refresh_token, + 'client_id': CLIENTID + } + ) + logging.debug('%s response %s', new_tokens.status_code, new_tokens.json()) + try: + self.root_token = new_tokens.json()['access_token'] + self.refresh_token = new_tokens.json()['refresh_token'] + except KeyError: + raise LibrusTricksAuthException('Invalid payload recived', new_tokens.json()) + + def revalidate_user(self): + """ + Aktualizuje token dostępu do Synergii, który wygasa po 24h. + """ + def do_revalidation(): + auth_session = requests.session() + new_token = auth_session.get( + FRESHURL.format(login=self.login), + headers={'Authorization': f'Bearer {self.root_token}'} + ) + logging.debug('%s response %s', new_token.status_code, new_token.json()) + return new_token + + new_token = do_revalidation() + if new_token.json().get('error') == 'access_denied': + logging.info('Obtaing new token failed! Refreshing root token') + self.revalidate_root() + new_token = do_revalidation() # again... + + try: + self.token = new_token.json()['accessToken'] + except KeyError: + raise LibrusTricksAuthException('Invalid response received', new_token.json()) + + def check_is_expired(self, use_clock=True, use_query=True): + """ + :param bool use_clock: Sprawdza na podstawie czasu + :param bool use_query: Sprawdza poprzez zapytanie http GET na ``/Me`` + :return: krotka z wynikami + :rtype: tuple[bool] + """ + clock_resp = None + query_resp = None + + if use_clock: + if datetime.now() > self.expires_in: + clock_resp = False + else: + clock_resp = True + if use_query: + test = requests.get('https://api.librus.pl/2.0/Me', headers={'Authorization': f'Bearer {self.token}'}) + if test.status_code == 401: + query_resp = False + else: + query_resp = True + + return clock_resp, query_resp + + @property + def is_valid(self): + """ + Umożliwia sprawdzenie czy konto ma jeszcze aktualny token. + + :return: ``False`` - trzeba wyrobić nowy token + :rtype: bool + """ + return self.check_is_expired(use_clock=False)[1] + + def dump_credentials(self, cred_file=None): + import json + if cred_file is None: + cred_file = open(f'{self.login}.json', 'w') + json.dump({ + 'user_dict': { + 'accessToken': self.token, + 'studentName': f'{self.name} {self.last_name}', + 'id': self.uid, + 'login': self.login, + }, + 'root_token': self.root_token, + 'revalidation_token': self.refresh_token, + 'exp_in': int(self.expires_in.timestamp()) + }, cred_file) + + def dict_credentials(self): + return { + 'user_dict': { + 'accessToken': self.token, + 'studentName': f'{self.name} {self.last_name}', + 'id': self.uid, + 'login': self.login, + }, + 'root_token': self.root_token, + 'revalidation_token': self.refresh_token, + 'exp_in': int(self.expires_in.timestamp()) + } + + +def load_json(cred_file): + import json + return SynergiaUser(**json.load(cred_file)) + + +def authorizer(email, password, user_agent=None): + """ + Zwraca listę użytkowników dostępnych dla danego konta Librus Portal + + :param str email: Email do Portalu Librus + :param str password: Hasło do Portalu Librus + :return: Listę z użytkownikami połączonymi do konta Librus Synergia + :rtype: list[librus_tricks.auth.SynergiaUser] + """ + if user_agent is None: + from random import choice + user_agent = choice([XIAOMI_USERAGENT, IPHONE_USERAGENT]) + logging.debug('No user-agent specified, using %s', user_agent) + + auth_session = requests.session() + auth_session.headers.update({'User-Agent': user_agent, 'X-Requested-With': 'pl.librus.synergiaDru2'}) + site = auth_session.get(LIBRUSLOGINURL) + soup = BeautifulSoup(site.text, 'html.parser') + csrf = soup.find('meta', attrs={'name': 'csrf-token'})['content'] + login_response_redirection = auth_session.post( + LOGINURL, json={'email': email, 'password': password}, + headers={'X-CSRF-TOKEN': csrf, 'Content-Type': 'application/json'} + ) + + if login_response_redirection.status_code != 200: + if login_response_redirection.status_code == 403: + if 'g-recaptcha-response' in login_response_redirection.json()['errors']: + raise CaptchaRequired(login_response_redirection.json()) + raise LibrusPortalInvalidPasswordError(login_response_redirection.json()) + raise LibrusLoginError(login_response_redirection.text) + + redirection_addr = login_response_redirection.json()['redirect'] + redirection_response = auth_session.get(redirection_addr, allow_redirects=False) + oauth_code = redirection_response.headers['location'].replace('http://localhost/bar?code=', '') + + synergia_root_response = auth_session.post( + OAUTHURL, + data={ + 'grant_type': 'authorization_code', + 'code': oauth_code, + 'client_id': CLIENTID, + 'redirect_uri': REDIRURL + } + ) + synergia_root_login_token = synergia_root_response.json()['access_token'] + synergia_root_revalidation_token = synergia_root_response.json()['refresh_token'] + synergia_root_expiration = synergia_root_response.json()['expires_in'] + + synergia_users_response = auth_session.get(SYNERGIAAUTHURL, + headers={'Authorization': f'Bearer {synergia_root_login_token}'}) + synergia_users_raw = synergia_users_response.json()['accounts'] + synergia_users = [ + SynergiaUser(user_data, synergia_root_login_token, synergia_root_revalidation_token, synergia_root_expiration) + for user_data in synergia_users_raw] + return synergia_users diff --git a/cache.py b/cache.py new file mode 100644 index 0000000..c9d3352 --- /dev/null +++ b/cache.py @@ -0,0 +1,134 @@ +from datetime import datetime + +from sqlalchemy import create_engine, String, JSON, Column, DateTime, Integer +from sqlalchemy.ext.declarative import declarative_base +from sqlalchemy.orm import sessionmaker +from sqlalchemy.pool import StaticPool + + +class CacheBase: + def add_object(self, uid, cls, resource): + pass + + def get_object(self, uid, cls): + raise NotImplementedError('get_object require implementation') + + def del_object(self, uid): + raise NotImplementedError('del_object require implementation') + + def clear_objects(self): + raise NotImplementedError('clear_objects require implementation') + + def count_object(self): + raise NotImplementedError('count_object require implementation') + + def add_query(self, uri, response, user_id): + pass + + def get_query(self, uri, user_id): + raise NotImplementedError('get_query require implementation') + + def del_query(self, uri, user_id): + raise NotImplementedError('del_query require implementation') + + def clear_queries(self): + raise NotImplementedError('clear_queries require implementation') + + def count_queries(self): + raise NotImplementedError('count_queries require implementation') + + def about_backend(self): + raise NotImplementedError('required providing info about cache provider') + + +class DumbCache(CacheBase): + def get_object(self, uid, cls): + return + + def get_query(self, uri, user_id): + return + + def __repr__(self): + return f'' + + +class AlchemyCache(CacheBase): + Base = declarative_base() + + def __init__(self, engine_uri='sqlite:///:memory:'): + engine = create_engine(engine_uri, connect_args={'check_same_thread': False}, poolclass=StaticPool) + + db_session = sessionmaker(bind=engine) + db_session.configure(bind=engine) + self.session = db_session() + self.Base.metadata.create_all(engine) + self.syn_session = None + + class APIQueryCache(Base): + __tablename__ = 'uri_cache' + + pk = Column(Integer(), primary_key=True) + uri = Column(String(length=512)) + owner = Column(String(length=16)) + response = Column(JSON()) + last_load = Column(DateTime()) + + class ObjectLoadCache(Base): + __tablename__ = 'object_cache' + + uid = Column(Integer(), primary_key=True) + name = Column(String(length=64)) + resource = Column(JSON()) + last_load = Column(DateTime()) + + def add_object(self, uid, cls, resource): + self.session.add( + self.ObjectLoadCache(uid=uid, name=cls.__name__, resource=resource, last_load=datetime.now()) + ) + self.session.commit() + + def get_object(self, uid, cls): + """ + + :rtype: AlchemyCache.ObjectLoadCache + """ + response = self.session.query(self.ObjectLoadCache).filter_by(uid=uid, name=cls.__name__).first() + if response is None: + return None + return cls.assembly(response.resource, self.syn_session) + + def add_query(self, uri, response, user_id): + self.session.add( + self.APIQueryCache(uri=uri, response=response, last_load=datetime.now(), owner=user_id) + ) + self.session.commit() + + def get_query(self, uri, user_id): + """ + + :rtype: AlchemyCache.APIQueryCache + """ + return self.session.query(self.APIQueryCache).filter_by(uri=uri, owner=user_id).first() + + def del_query(self, uri, user_id): + self.session.query(self.APIQueryCache).filter_by(uri=uri, owner=user_id).delete() + self.session.commit() + + def del_object(self, uid): + self.session.query(self.ObjectLoadCache).filter_by(uid=uid).delete() + self.session.commit() + + def clear_queries(self): + self.session.query(self.APIQueryCache).delete() + + def clear_objects(self): + self.session.query(self.ObjectLoadCache).delete() + + def count_object(self): + return self.session.query(self.ObjectLoadCache).count() + + def about_backend(self): + return f'SQLAlchemy ORM {self.session.bind.dialect.name} {self.session.bind.dialect.driver}' + + def __repr__(self): + return f'<{self.__class__.__name__} with {self.session.bind.dialect.name} backend using {self.session.bind.dialect.driver} driver ({self.session.bind.url})>' diff --git a/classes.py b/classes.py new file mode 100644 index 0000000..057a6a1 --- /dev/null +++ b/classes.py @@ -0,0 +1,1080 @@ +from datetime import datetime, timedelta + +from librus_tricks.exceptions import SessionRequired, APIPathIsEmpty + + +class _RemoteObjectsUIDManager: + """ + Menadżer obiektów, które dopiero zostaną utworzone. + """ + + def __init__(self, session, parent): + """ + + :param librus_tricks.core.SynergiaClient session: Obiekt sesji + """ + self.__storage = dict() + self._session = session + self.__parent = parent + + def set_object(self, attr, uid, cls): + """ + Zapisuje dane przyszłego obiektu. + + :param str attr: Nazwa przyszłego property + :param int uid: Id obiektu + :param cls: Klasa obiektu + """ + self.__storage[attr] = uid, cls + # self.__parent.__setattr__(attr, cls.create(uid=uid, session=self.__session)) + return self + + def set_value(self, attr, val): + """ + Ustawia obiekt. + + :param str attr: Nazwa obiektu + :param val: Obiekt + """ + self.__storage[attr] = val + return self + + def assembly(self, attr): + """ + Pobiera wcześniej zapisany obiekt. + + :param str attr: Nazwa property + :return: Żądany obiekt + """ + uid, cls = self.__storage[attr] + return cls.create(uid=uid, session=self._session) + + def return_id(self, attr): + """ + Zwraca id obiektu. + + :param str attr: Nazwa property + :rtype: int + :return: Id obiektu + """ + return self.__storage[attr][0] + + +class SynergiaGenericClass: + """ + Klasa macierzysta dla obiektów dziennika Synergia. + """ + + def __init__(self, uid, resource, session): + """ + + :param str uid: Id żądanego obiektu + :param librus_tricks.core.SynergiaClient session: Obiekt sesji + :param resource: ścieżka do źródła danych + :type resource: iterable of str + :param str extraction_key: str zawierający klucz do wyjęcia danych + :param dict resource: dict zawierający gotowe dane (np. załadowane z cache) + """ + + self._session = session + self.uid = uid + self.objects = _RemoteObjectsUIDManager(self._session, self) + self._json_resource = resource + + # Of course i can comment it out, but for code completion props will be better + # def __getattr__(self, name): + # return self.objects_ids.assembly(name) + + @classmethod + def assembly(cls, resource, session): + """ + Umożliwia stworzenie obiektu posiadając dict i obiekt sesji. + + :param dict resource: Gotowe dane do stworzenia obiektu + :param librus_tricks.core.SynergiaClient session: Obiekt sesji + :return: Nowy obiekt + """ + self = cls(resource['Id'], resource, session) + return self + + @classmethod + def create(cls, uid=None, path=('',), session=None, extraction_key=None, expire=timedelta(seconds=1)): + """ + Pobiera i składa nowy obiekt. + + :param int uid: Id obiektu + :param tuple of str path: Niezłożona ścieżka API + :param librus_tricks.core.SynergiaClient session: Obiekt sesji + :param str extraction_key: Klucz do wyciągnięcia danych + :return: Pobrany obiekt + """ + import logging + + if uid is None or session is None: + raise SessionRequired() + + maybe_response = session.cache.get_object(uid, cls) + if not maybe_response is None: + logging.debug('Returning %s %s from object cache', maybe_response, uid) + return maybe_response + + if path == ('',): + raise APIPathIsEmpty(f'Path for {cls.__name__} class is empty!') + + response = session.get_cached_response(*path, uid, max_lifetime=expire) + logging.debug('Returning %s %s object from response cache', cls.__name__, uid) + + if extraction_key is None: + extraction_key = SynergiaGenericClass.auto_extract(response) + + resource = response[extraction_key] + self = cls(resource['Id'], resource, session) + return self + + @staticmethod + def auto_extract(payload): + """ + Próbuje automatycznie wydobyć klucz. + + :param dict payload: + :return: Wydobyty klucz + :rtype: str + """ + for key in payload.keys(): + if key not in ('Resources', 'Url'): + return key + return + + def export_resource(self): + return self._json_resource.copy() + + def __repr__(self): + return f'<{self.__class__.__name__} {self.uid} at {hex(id(self))}>' + + def __hash__(self): + return hash(self.uid) + + def __eq__(self, other): + if isinstance(other, self.__class__): + return self.uid == other.uid + raise TypeError(f'Object is not instance of {self.__class__.__name__}') + + def __ne__(self, other): + return not self.__eq__(other) + + def _is_compatible(self, other): + if not isinstance(other, self.__class__): + raise TypeError() + return True + + +class SynergiaTeacher(SynergiaGenericClass): + def __init__(self, uid, resource, session): + super().__init__(uid, resource, session) + self.name = self._json_resource['FirstName'] + self.last_name = self._json_resource['LastName'] + + @classmethod + def create(cls, uid=None, path=('Users',), session=None, extraction_key='User', expire=timedelta(days=31)): + return super().create(uid, path, session, extraction_key, expire) + + def __repr__(self): + return f'<{self.__class__.__name__} {self.name} {self.last_name}>' + + def __str__(self): + return f'{self.name} {self.last_name}' + + +class SynergiaStudent(SynergiaTeacher): + pass + + +class SynergiaGlobalClass(SynergiaGenericClass): + """Klasa reprezentująca klasę (np. 1C)""" + + def __init__(self, uid, resource, session): + """ + Tworzy obiekt reprezentujący klasę (jako zbiór uczniów) + + :param str uid: id klasy + :param librus_tricks.core.SynergiaClient session: obiekt sesji z API Synergii + :param dict resource: dane z json'a + """ + super().__init__(uid, resource, session) + + self.alias = f'{self._json_resource["Number"]}{self._json_resource["Symbol"]}' + self.begin_date = datetime.strptime(self._json_resource['BeginSchoolYear'], '%Y-%m-%d').date() + self.end_date = datetime.strptime(self._json_resource['EndSchoolYear'], '%Y-%m-%d').date() + self.objects.set_object( + 'tutor', self._json_resource['ClassTutor']['Id'], SynergiaTeacher + ) + + @property + def tutor(self) -> SynergiaTeacher: + return self.objects.assembly('tutor') + + def __repr__(self): + return f'<{self.__class__.__name__} {self.alias}>' + + +class SynergiaVirtualClass(SynergiaGenericClass): + def __init__(self, uid, resource, session): + """ + Tworzy obiekt reprezentujący grupę uczniów + + :param str uid: id klasy + :param librus_tricks.core.SynergiaClient session: obiekt sesji z API Synergii + :param dict resource: dane z json'a + """ + super().__init__(uid, resource, session) + + self.name = self._json_resource['Name'] + self.number = self._json_resource['Number'] + self.symbol = self._json_resource['Symbol'] + self.objects.set_object( + 'teacher', self._json_resource['Teacher']['Id'], SynergiaTeacher + ).set_object( + 'subject', self._json_resource['Subject']['Id'], SynergiaSubject + ) + + def __repr__(self): + return f'<{self.__class__.__name__} {self.name}>' + + @property + def teacher(self) -> SynergiaTeacher: + return self.objects.assembly('teacher') + + @property + def subject(self): + return self.objects.assembly('subject') + + +class SynergiaSubject(SynergiaGenericClass): + def __init__(self, uid, resource, session): + super().__init__(uid, resource, session) + self.name = self._json_resource['Name'] + self.short_name = self._json_resource['Short'] + + @classmethod + def create(cls, uid=None, path=('Subjects',), session=None, extraction_key='Subject', expire=timedelta(days=31)): + return super().create(uid, path, session, extraction_key, expire) + + def __repr__(self): + return f'<{self.__class__.__name__} {self.name}>' + + def __str__(self): + return self.name + + +class SynergiaLesson(SynergiaGenericClass): + def __init__(self, uid, resource, session): + """ + Klasa reprezentująca jednostkową lekcję + + :type session: librus_tricks.core.SynergiaClient + """ + super().__init__(uid, resource, session) + + self.objects.set_object( + 'teacher', self._json_resource['Teacher']['Id'], SynergiaTeacher + ).set_object( + 'subject', self._json_resource['Subject']['Id'], SynergiaSubject + ) + + @classmethod + def create(cls, uid=None, path=('Lessons',), session=None, extraction_key='Lesson', expire=timedelta(minutes=5)): + return super().create(uid, path, session, extraction_key, expire) + + @property + def teacher(self) -> SynergiaTeacher: + return self.objects.assembly('teacher') + + @property + def subject(self) -> SynergiaSubject: + return self.objects.assembly('subject') + + +class SynergiaGradeCategory(SynergiaGenericClass): + def __init__(self, uid, resource, session): + super().__init__(uid, resource, session) + + def __try_to_extract(payload, extraction_key, false_return=None): + if extraction_key in payload.keys(): + return payload[extraction_key] + return false_return + + self.count_to_the_average = self._json_resource['CountToTheAverage'] + self.name = self._json_resource['Name'] + self.obligation_to_perform = self._json_resource['ObligationToPerform'] + self.standard = self._json_resource['Standard'] + self.weight = __try_to_extract(self._json_resource, 'Weight', false_return=0) + + if 'Teacher' in self._json_resource.keys(): + self.objects.set_object( + 'teacher', self._json_resource['Id'], SynergiaTeacher + ) + + @classmethod + def create(cls, uid=None, path=('Grades', 'Categories'), session=None, extraction_key='Category', + expire=timedelta(days=31)): + return super().create(uid, path, session, extraction_key, expire) + + @property + def teacher(self): + try: + return self.objects.assembly('teacher') + except KeyError: + return + + def __repr__(self): + return f'<{self.__class__.__name__} {self.name}>' + + +class SynergiaGradeComment(SynergiaGenericClass): + def __init__(self, uid, resource, session): + super().__init__(uid, resource, session) + + self.text = self._json_resource['Text'] + self.objects.set_object( + 'teacher', self._json_resource['AddedBy']['Id'], SynergiaTeacher + ).set_object( + 'bind', self._json_resource['Grade']['Id'], SynergiaGrade + ) + + def __str__(self): + return self.text + + @classmethod + def create(cls, uid=None, path=('Grades', 'Comments'), session=None, extraction_key='Comment', + expire=timedelta(days=31)): + return super().create(uid, path, session, extraction_key, expire) + + @property + def teacher(self) -> SynergiaTeacher: + return self.objects.assembly('teacher') + + @property + def grade_bind(self): + return self.objects.assembly('bind') + + def __repr__(self): + return f'<{self.__class__.__name__} {self.text}>' + + +class SynergiaBaseTextGrade(SynergiaGenericClass): + def __init__(self, uid, resource, session): + super().__init__(uid, resource, session) + + self.add_date = datetime.strptime(self._json_resource['AddDate'], '%Y-%m-%d %H:%M:%S') + self.date = datetime.strptime(self._json_resource['Date'], '%Y-%m-%d').date() + self.grade = self._json_resource['Grade'] + self.semester = self._json_resource['Semester'] + self.visible = self._json_resource['ShowInGradesView'] + self.objects.set_object( + 'teacher', self._json_resource['AddedBy']['Id'], SynergiaTeacher + ).set_object( + 'subject', self._json_resource['Subject']['Id'], SynergiaSubject + ).set_object( + 'student', self._json_resource['Student']['Id'], SynergiaStudent + ) + + @classmethod + def create(cls, uid=None, path=('BaseTextGrades',), session=None, extraction_key='BaseTextGrades', + expire=timedelta(minutes=5)): + return super().create(uid, path, session, extraction_key, expire) + + @property + def teacher(self) -> SynergiaTeacher: + return self.objects.assembly('teacher') + + @property + def subject(self) -> SynergiaSubject: + return self.objects.assembly('subject') + + @property + def student(self): + return self.objects.assembly('student') + + +class SynergiaGrade(SynergiaGenericClass): + def __init__(self, uid, resource, session): + super().__init__(uid, resource, session) + + class GradeMetadata: + def __init__(self, is_c, is_s, is_sp, is_f, is_fp): + self.is_constituent = is_c + self.is_semester_grade = is_s + self.is_semester_grade_proposition = is_sp + self.is_final_grade = is_f + self.is_final_grade_proposition = is_fp + + self.add_date = datetime.strptime(self._json_resource['AddDate'], '%Y-%m-%d %H:%M:%S') + self.date = datetime.strptime(self._json_resource['Date'], '%Y-%m-%d').date() + self.grade = self._json_resource['Grade'] + self.is_constituent = self._json_resource['IsConstituent'] + self.semester = self._json_resource['Semester'] + self.metadata = GradeMetadata( + self._json_resource['IsConstituent'], + self._json_resource['IsSemester'], + self._json_resource['IsSemesterProposition'], + self._json_resource['IsFinal'], + self._json_resource['IsFinalProposition'] + ) + + self.objects.set_object( + 'teacher', self._json_resource['AddedBy']['Id'], SynergiaTeacher + ).set_object( + 'subject', self._json_resource['Subject']['Id'], SynergiaSubject + ).set_object( + 'category', self._json_resource['Category']['Id'], SynergiaGradeCategory + ) + + @property + def is_special(self): + m = self.metadata + if m.is_final_grade or m.is_final_grade_proposition or m.is_semester_grade or m.is_semester_grade_proposition: + return True + return False + + def __repr__(self): + return f'<{self.__class__.__name__} {self.grade} from SynergiaSubject with id {self.objects.return_id("subject")} ' \ + f'added {self.add_date.strftime("%Y-%m-%d %H:%M:%S")}>' + + def __str__(self): + return self.grade + + @property + def teacher(self) -> SynergiaTeacher: + return self.objects.assembly('teacher') + + @property + def subject(self) -> SynergiaSubject: + return self.objects.assembly('subject') + + @property + def category(self) -> SynergiaGradeCategory: + return self.objects.assembly('category') + + @property + def comments(self): + """ + + :rtype: list of SynergiaGradeComment + """ + if self._json_resource.get('Comments') is not None: + return [ + SynergiaGradeComment.create( + uid=com.get('Id'), session=self._session + ) for com in self._json_resource.get('Comments') + ] + return tuple() + + @property + def real_value(self): + return { + '1': 1, + '1+': 1.5, + '2-': 1.75, + '2': 2, + '2+': 2.5, + '3-': 2.75, + '3': 3, + '3+': 3.5, + '4-': 3.75, + '4': 4, + '4+': 4.5, + '5-': 4.75, + '5': 5, + '5+': 5.5, + '6-': 5.75, + '6': 6 + }.get(self.grade) + + @classmethod + def create(cls, uid=None, path=('Grades',), session=None, extraction_key='Grade', expire=timedelta(minutes=45)): + return super().create(uid, path, session, extraction_key, expire) + + def __eq__(self, other): + self._is_compatible(other) + return self.real_value == other.real_value and self.category.weight + + def __gt__(self, other): + self._is_compatible(other) + return self.real_value > other.real_value and self.category.weight >= other.category.weight + + def __ge__(self, other): + self._is_compatible(other) + return self.real_value >= other.real_value and self.category.weight >= other.category.weight + + def __lt__(self, other): + self._is_compatible(other) + return self.real_value < other.real_value and self.category.weight <= other.category.weight + + def __le__(self, other): + self._is_compatible(other) + return self.real_value <= other.real_value and self.category.weight <= other.category.weight + + +class SynergiaAttendanceType(SynergiaGenericClass): + def __init__(self, uid, resource, session): + super().__init__(uid, resource, session) + self.color = self._json_resource['ColorRGB'] + self.is_presence_kind = self._json_resource['IsPresenceKind'] + self.name = self._json_resource['Name'] + self.short_name = self._json_resource['Short'] + + @classmethod + def create(cls, uid=None, path=('Attendances', 'Types'), session=None, extraction_key='Type', + expire=timedelta(days=31)): + return super().create(uid, path, session, extraction_key, expire) + + def __repr__(self): + return f'<{self.__class__.__name__} {self.short_name}>' + + def __str__(self): + return self.name + + +class SynergiaAttendance(SynergiaGenericClass): + def __init__(self, uid, resource, session): + super().__init__(uid, resource, session) + self.add_date = datetime.strptime(self._json_resource['AddDate'], '%Y-%m-%d %H:%M:%S') + self.date = datetime.strptime(self._json_resource['Date'], '%Y-%m-%d').date() + self.lesson_no = int(self._json_resource['LessonNo']) + self.objects.set_object( + 'teacher', self._json_resource['AddedBy']['Id'], SynergiaTeacher + ).set_object( + 'student', self._json_resource['Student']['Id'], SynergiaStudent + ).set_object( + 'type', self._json_resource['Type']['Id'], SynergiaAttendanceType + ).set_object( + 'lesson', resource['Lesson']['Id'], SynergiaLesson + ) + + @classmethod + def create(cls, uid=None, path=('Attendances',), session=None, extraction_key='Attendance', + expire=timedelta(minutes=10)): + return super().create(uid, path, session, extraction_key, expire) + + @property + def teacher(self): + """ + :rtype: SynergiaTeacher + """ + return self.objects.assembly('teacher') + + @property + def student(self): + """ + :rtype: SynergiaStudent + """ + return self.objects.assembly('student') + + @property + def type(self): + """ + :rtype: SynergiaAttendanceType + """ + return self.objects.assembly('type') + + @property + def lesson(self): + """ + :rtype: librus_tricks.classes.SynergiaLesson + """ + return self.objects.assembly('lesson') + + def __repr__(self): + return f'' + + def __str__(self): + return self.type.__str__() + + +class SynergiaExamCategory(SynergiaGenericClass): + def __init__(self, uid, resource, session): + super().__init__(uid, resource, session) + + self.name = self._json_resource['Name'] + self.objects.set_object('color', self._json_resource['Color']['Id'], SynergiaColor) + + @classmethod + def create(cls, uid=None, path=('HomeWorks', 'Categories'), session=None, extraction_key='Category', + expire=timedelta(days=31)): + return super().create(uid, path, session, extraction_key, expire) + + @property + def color(self): + """ + + :rtype: SynergiaColor + """ + return self.objects.assembly('color') + + def __str__(self): + return self.name + + +class SynergiaExam(SynergiaGenericClass): + def __init__(self, uid, resource, session): + + super().__init__(uid, resource, session) + + self.add_date = datetime.strptime(self._json_resource['AddDate'], '%Y-%m-%d %H:%M:%S') + self.content = self._json_resource['Content'] + self.date = datetime.strptime(self._json_resource['Date'], '%Y-%m-%d').date() + self.lesson = self._json_resource['LessonNo'] + if self._json_resource['TimeFrom'] is None: + self.time_start = None + else: + self.time_start = datetime.strptime(self._json_resource['TimeFrom'], '%H:%M:%S').time() + if self._json_resource['TimeTo'] is None: + self.time_end = None + else: + self.time_end = datetime.strptime(self._json_resource['TimeTo'], '%H:%M:%S').time() + + self.objects.set_object( + 'teacher', self._json_resource['CreatedBy']['Id'], SynergiaTeacher + ).set_object( + 'category', self._json_resource['Category']['Id'], SynergiaExamCategory + ) + if 'Subject' in self._json_resource: + self.objects.set_object('subject', self._json_resource['Subject']['Id'], SynergiaSubject) + self.__subject_present = True + else: + self.__subject_present = False + + @classmethod + def create(cls, uid=None, path=('HomeWorks',), session=None, extraction_key='HomeWork', expire=timedelta(days=3)): + return super().create(uid, path, session, extraction_key, expire) + + def __repr__(self): + return f'<{self.__class__.__name__} ' \ + f'{self.date.strftime("%Y-%m-%d")} for subject {self.subject}>' + + @property + def teacher(self) -> SynergiaTeacher: + return self.objects.assembly('teacher') + + # @property + # def group(self): + # """ + # + # :rtype: SynergiaGlobalClass + # :rtype: SynergiaVirtualClass + # """ + # if self.objects_ids.group_type is SynergiaGlobalClass: + # return SynergiaGlobalClass(self.objects_ids.group, self._session) + # else: + # return SynergiaVirtualClass(self.objects_ids.group, self._session) + + @property + def subject(self): + if self.__subject_present: + return self.objects.assembly('subject') + return None + + @property + def category(self): + return self.objects.assembly('category') + + +class SynergiaColor(SynergiaGenericClass): + def __init__(self, uid, resource, session): + super().__init__(uid, resource, session) + self.name = self._json_resource['Name'] + self.hex_rgb = self._json_resource['RGB'] + + @classmethod + def create(cls, uid=None, path=('Colors',), session=None, extraction_key='Color', expire=timedelta(days=31)): + return super().create(uid, path, session, extraction_key, expire) + + def __repr__(self): + return f'<{self.__class__.__name__} {self.hex_rgb}>' + + def __str__(self): + return self.hex_rgb + + +class SynergiaClassroom(SynergiaGenericClass): + def __init__(self, uid, resource, session): + super().__init__(uid, resource, session) + self.name = self._json_resource['Name'] + self.symbol = self._json_resource['Symbol'] + + @classmethod + def create(cls, uid=None, path=('Classrooms',), session=None, extraction_key=None, expire=timedelta(days=31)): + return super().create(uid, path, session, extraction_key, expire) + + def __repr__(self): + return f'' + + def __str__(self): + return self.name + + +class SynergiaTeacherFreeDaysTypes(SynergiaGenericClass): + def __init__(self, uid, resource, session): + super().__init__(uid, resource, session) + self.name = self._json_resource[0]['Name'] + + +class SynergiaTeacherFreeDays(SynergiaGenericClass): + def __init__(self, uid, resource, session): + super().__init__(uid, resource, session) + + self.starts = datetime.strptime(self._json_resource['DateFrom'], '%Y-%m-%d').date() + self.ends = datetime.strptime(self._json_resource['DateTo'], '%Y-%m-%d').date() + self.objects.set_object( + 'teacher', self._json_resource['Teacher']['Id'], SynergiaTeacher + ) + + if self._json_resource.get('TimeTo') is not None: + self.time_begin = datetime.strptime(self._json_resource['TimeFrom'], '%H:%M:%S').time() + self.time_ends = datetime.strptime(self._json_resource['TimeTo'], '%H:%M:%S').time() + else: + self.time_begin = None + self.time_ends = None + + @property + def period_length(self): + return self.ends - self.starts + + @property + def teacher(self) -> SynergiaTeacher: + """ + + :rtype: SynergiaTeacher + """ + return self.objects.assembly('teacher') + + def __repr__(self): + return f'' + + +class SynergiaSchoolFreeDays(SynergiaGenericClass): + def __init__(self, uid, resource, session): + super().__init__(uid, resource, session) + self.starts = datetime.strptime(self._json_resource['DateFrom'], '%Y-%m-%d').date() + self.ends = datetime.strptime(self._json_resource['DateTo'], '%Y-%m-%d').date() + self.name = self._json_resource['Name'] + + @classmethod + def create(cls, uid=None, path=('Calendars', 'SchoolFreeDays'), session=None, extraction_key='SchoolFreeDays', + expire=timedelta(minutes=5)): + return super().create(uid, path, session, extraction_key, expire) + + def __repr__(self): + return f'<{self.__class__.__name__} {self.starts.isoformat()} - {self.ends.isoformat()}>' + + +class SynergiaTimetableEntry(SynergiaGenericClass): + def __init__(self, uid, resource, session): + super().__init__(uid, resource, session) + self.available = datetime.strptime(resource['DateFrom'], '%Y-%m-%d').date(), \ + datetime.strptime(resource['DateTo'], '%Y-%m-%d').date() + + @classmethod + def create(cls, uid=None, path=('TimetableEntries',), session=None, extraction_key='TimetableEntry', + expire=timedelta(seconds=15)): + return super().create(uid, path, session, extraction_key, expire) + + +class SynergiaTimetableEvent: + def __init__(self, resource, session): + self.lesson_no = int(resource['LessonNo']) #: int: numer lekcji + self.start = datetime.strptime(resource['HourFrom'], '%H:%M').time() #: time: początek lekcji + self.end = datetime.strptime(resource['HourTo'], '%H:%M').time() #: time: koniec lekcji + self.is_cancelled = resource['IsCanceled'] #: bool: czy lekcja jest odwołana + self.is_sub = resource['IsSubstitutionClass'] #: bool: czy lekcja jest zastępstwem + self.preloaded = { + 'subject_title': resource['Subject']['Name'], + 'teacher': f'{resource["Teacher"]["FirstName"]} {resource["Teacher"]["LastName"]}' + } + self.objects = _RemoteObjectsUIDManager(session, self) + self.objects.set_object( + 'subject', resource['Subject']['Id'], SynergiaSubject + ).set_object( + 'teacher', resource['Teacher']['Id'], SynergiaTeacher + ) + + self.__set_classroom(resource) + + def __set_classroom(self, resource): + """ + + :param dict resource: + :return: + """ + if 'Classroom' in resource.keys(): + self.objects.set_object( + 'classroom', resource['Classroom']['Id'], SynergiaClassroom + ) + elif 'OrgClassroom' in resource.keys(): + self.objects.set_object( + 'classroom', resource['OrgClassroom']['Id'], SynergiaClassroom + ) + else: + self.objects.set_value('classroom', None) + + @property + def lesson_status(self): + if self.is_cancelled: + return 'Cancelled' + elif self.is_sub: + return 'Changed' + return 'Planned' + + @property + def subject(self): + """ + :rtype: librus_tricks.classes.SynergiaSubject + """ + return self.objects.assembly('subject') + + @property + def teacher(self): + """ + :rtype: librus_tricks.classes.SynergiaTeacher + """ + return self.objects.assembly('teacher') + + @property + def classroom(self): + """ + :rtype: librus_tricks.classes.SynergiaClassroom + """ + return self.objects.assembly('classroom') + + @property + def human_readable_time_range(self): + return f'{self.start.strftime("%H:%M")} - {self.end.strftime("%H:%M")}' + + def __repr__(self): + return f'' + + def __str__(self): + return self.preloaded["subject_title"] + + +class SynergiaTimetableDay: + def __init__(self, lessons): + self.lessons = tuple(lessons) #: tuple[SynergiaTimetableEvent]: krotka z lekcjami + if self.lessons.__len__() != 0: + self.day_start = self.lessons[0].start + self.day_end = self.lessons[-1].end + else: + self.day_start = None + self.day_end = None + + def __repr__(self): + return f'<{self.__class__.__name__} with {self.lessons.__len__()} lessons between {self.day_start} and {self.day_end}>' + + +class SynergiaTimetable(SynergiaGenericClass): + """ + Obiekt zawierający cały tydzień w planie lekcji + """ + + def __init__(self, uid, resource, session): + super().__init__(uid, resource, session) + self.days = self.convert_parsed_timetable( + self.parse_timetable(resource) + ) #: list[SynergiaTimetableDay]: lista z dniami tygodnia + + @property + def today_timetable(self): + """ + + :rtype: list of SynergiaTimetableEvent + """ + return self.days[datetime.now().date()] + + @classmethod + def assembly(cls, resource, session): + pseudo_id = int(datetime.now().timestamp()).__str__() + self = cls(pseudo_id, resource, session) + return self + + @classmethod + def create(cls, uid=None, path=('Timetables',), session=None, extraction_key='Timetable', + expire=timedelta(seconds=15)): + response = session.get_cached_response(*path) + + if extraction_key is None: + extraction_key = SynergiaGenericClass.auto_extract(response) + + resource = response[extraction_key] + self = cls.assembly(resource, session) + return self + + @staticmethod + def parse_timetable(resource): + root = {} + + for day in resource.keys(): + day_date = datetime.strptime(day, '%Y-%m-%d').date() + root[day_date] = [] + for period in resource[day]: + if period.__len__() != 0: + root[day_date].append(period[0]) + return root + + def convert_parsed_timetable(self, timetable): + for day in timetable: + for event_index in range(len(timetable[day])): + if timetable[day][event_index].keys().__len__() != 0: + timetable[day][event_index] = SynergiaTimetableEvent(timetable[day][event_index], self._session) + + for day in timetable.keys(): + timetable[day] = SynergiaTimetableDay(timetable[day]) + + return timetable + + def __repr__(self): + return f'<{self.__class__.__name__} for {self.days.keys()}>' + + def __str__(self): + o_str = '' + for day_key in self.days.keys(): + o_str += f'{day_key}\n' + for event in self.days[day_key]: + if event != {}: + o_str += f' {event.__str__()}\n' + return o_str + + +class SynergiaNativeMessageAuthor(SynergiaGenericClass): + def __init__(self, uid, resource, session): + super().__init__(uid, resource, session) + self.name = resource.get('Name') + + @classmethod + def create(cls, uid=None, path=('Messages', 'User'), session=None, extraction_key='User', + expire=timedelta(days=60)): + return super().create(uid, path, session, extraction_key, expire) + + def __repr__(self): + return f'<{self.__class__.__name__} {self.name}>' + + @property + def matching_indentity(self): + teachers = self._session.return_objects('Users', cls=SynergiaTeacher, extraction_key='Users') + for teacher in teachers: + if str(teacher.name) in self.name and str(teacher.last_name) in self.name: + return teacher + return + + +class SynergiaNativeMessage(SynergiaGenericClass): + def __init__(self, uid, resource, session): + super().__init__(uid, resource, session) + self.body = self._json_resource['Body'] #: str: wiadomość + self.topic = self._json_resource['Subject'] #: str: temat + self.send_date = datetime.fromtimestamp(self._json_resource['SendDate']) #: datetime: data wysłania + self.objects.set_object('sender', self._json_resource['Sender']['Id'], SynergiaNativeMessageAuthor) + + @property + def sender(self) -> SynergiaNativeMessageAuthor: + return self.objects.assembly('sender') + + @classmethod + def create(cls, uid=None, path=('Messages',), session=None, extraction_key='Message', expire=timedelta(days=31)): + return super().create(uid, path, session, extraction_key, expire) + + +class SynergiaNews(SynergiaGenericClass): + """ + Obiekt reprezentujący ogłoszenie szkolne + """ + + def __init__(self, uid, resource, session): + super().__init__(uid, resource, session) + self.content = self._json_resource['Content'] #: str: wiadomość ogłoszenia + self.created = datetime.strptime(self._json_resource['CreationDate'], + '%Y-%m-%d %H:%M:%S') #: datetime: data utworzenia + self.unique_id = self._json_resource['Id'] #: int: id ogłoszenia + self.topic = self._json_resource['Subject'] #: str: temat + self.was_read = self._json_resource['WasRead'] #: bool: status odczytania? + self.starts = datetime.strptime(self._json_resource['StartDate'], '%Y-%m-%d') #: date: ?? + self.ends = datetime.strptime(self._json_resource['EndDate'], '%Y-%m-%d') #: date: ?? + self.objects.set_object( + 'teacher', self._json_resource['AddedBy']['Id'], SynergiaTeacher + ) + + @property + def teacher(self) -> SynergiaTeacher: + return self.objects.assembly('teacher') + + @classmethod + def create(cls, uid=None, path=('SchoolNotices',), session=None, extraction_key='SchoolNotices', + expire=timedelta(days=31)): + return super().create(uid, path, session, extraction_key, expire) + + def __repr__(self): + return f'' + + +class SynergiaSchool(SynergiaGenericClass): + """ + Obiekt zawierający informacje o szkole + """ + + def __init__(self, uid, resource, session): + super().__init__(uid, resource, session) + self.name = resource['Name'] #: str: nazwa szkoły + self.location = f'{resource["Town"]} {resource["Street"]} {resource["BuildingNumber"]}' #: str: adres szkoły + + @classmethod + def create(cls, uid=None, path=('Schools',), session=None, extraction_key='School', expire=timedelta(seconds=1)): + return super().create(uid, path, session, extraction_key, expire) + + def __repr__(self): + return f'<{self.__class__.__name__} {self.name}>' + + +class SynergiaSubstitution(SynergiaGenericClass): + def __init__(self, uid, resource, session): + super().__init__(uid, resource, session) + + @classmethod + def create(cls, uid=None, path=('Calendars', 'Substitutions'), session=None, extraction_key='Substitution', + expire=timedelta(days=7)): + return super().create(uid, path, session, extraction_key, expire) + + +class SynergiaRealization(SynergiaGenericClass): + def __init__(self, uid, resource, session): + super().__init__(uid, resource, session) + self.topic = resource['Topic'] + self.date = datetime.strptime(resource['Date'], '%Y-%m-%d') + self.is_trip = resource['IsTrip'] + self.teaching_program = resource.get('TeachingProgramTopic') + self.lesson_no = resource['LessonNo'] + self.objects.set_object( + 'teacher', self._json_resource['AddedBy']['Id'], SynergiaTeacher + ).set_object( + 'lesson', resource['Lesson']['Id'], SynergiaLesson + ) + + @property + def teacher(self): + """ + :rtype: librus_tricks.classes.SynergiaTeacher + """ + return self.objects.assembly('teacher') + + @property + def lesson(self): + """ + :rtype: librus_tricks.classes.SynergiaLesson + """ + return self.objects.assembly('lesson') + + @classmethod + def create(cls, uid=None, path=('Realizations',), session=None, extraction_key='Realization', + expire=timedelta(seconds=1)): + return super().create(uid, path, session, extraction_key, expire) + + def __repr__(self): + return f'<{self.__class__.__name__} {self.topic}>' diff --git a/core.py b/core.py new file mode 100644 index 0000000..609fecc --- /dev/null +++ b/core.py @@ -0,0 +1,483 @@ +import logging +from datetime import timedelta + +import requests + +from librus_tricks import cache as cache_lib +from librus_tricks import exceptions, tools +from librus_tricks.classes import * +from librus_tricks.messages import MessageReader + + +class SynergiaClient: + """Sesja z API Synergii""" + + def __init__(self, user, api_url='https://api.librus.pl/2.0', user_agent='LibrusMobileApp', + cache=cache_lib.AlchemyCache()): + """ + Tworzy sesję z API Synergii. + + :param librus_tricks.auth.SynergiaUser user: Użytkownik sesji + :param str api_url: Bazowy url api, zmieniaj jeżeli chcesz używać proxy typu beeceptor + :param str user_agent: User-agent klienta http, domyślnie się podszywa pod aplikację + :param librus_tricks.cache.CacheBase cache: Obiekt, który zarządza cache + """ + self.user = user + self.session = requests.session() + + self.session.headers.update({'User-Agent': user_agent}) + self.__auth_headers = {'Authorization': f'Bearer {user.token}'} + self.__api_url = api_url + + if cache_lib.CacheBase in cache.__class__.__bases__: + self.cache = cache + self.li_session = self + else: + raise exceptions.InvalidCacheManager(f'{cache} can not be a cache object!') + + self.__message_reader = None + + @property + def message_reader(self): + if self.__message_reader is None: + self.__message_reader = MessageReader(self) + return self.__message_reader + + def __repr__(self): + return f'' + + def __update_auth_header(self): + self.__auth_headers = {'Authorization': f'Bearer {self.user.token}'} + logging.debug('Updating headers to %s', self.__auth_headers) + + @staticmethod + def assembly_path(*elements, prefix='', suffix='', sep='/'): + """ + Składa str w jednego str, przydatne przy tworzeniu url. + + :param str elements: Elementy do stworzenia str + :param str prefix: Początek str + :param str suffix: Koniec str + :param str sep: str wstawiany pomiędzy elementy + :return: Złożony str + :rtype: str + """ + for element in elements: + prefix += sep + str(element) + return prefix + suffix + + # HTTP part + + def dispatch_http_code(self, response: requests.Response, callback=None, callback_args=tuple(), + callback_kwargs=None): + """ + Sprawdza czy serwer zgłasza błąd poprzez podanie kodu http, w przypadku błędu, rzuca wyjątkiem. + + :param requests.Response response: + :raises librus_tricks.exceptions.SynergiaNotFound: 404 + :raises librus_tricks.exceptions.SynergiaForbidden: 403 + :raises librus_tricks.exceptions.SynergiaAccessDenied: 401 + :raises librus_tricks.exceptions.SynergiaInvalidRequest: 401 + :rtype: requests.Response + :return: sprawdzona odpowiedź http + """ + if callback_kwargs is None: + callback_kwargs = dict() + + logging.debug('Dispatching response status') + if response.json().get('Code') == 'TokenIsExpired': + logging.info('Server returned error code "TokenIsExpired", trying to obtain new token') + self.user.revalidate_user() + self.__update_auth_header() + logging.debug('Repeating failed response') + return callback(*callback_args, **callback_kwargs) + + logging.debug('Dispatching http status code') + if response.status_code >= 400: + try: + raise { + 503: exceptions.SynergiaMaintenanceError(response.url, response.json()), + 500: exceptions.SynergiaServerError(response.url, response.json()), + 404: exceptions.SynergiaAPIEndpointNotFound(response.url), + 403: exceptions.SynergiaForbidden(response.url, response.json()), + 401: exceptions.SynergiaAccessDenied(response.url, response.json()), + 400: exceptions.SynergiaAPIInvalidRequest(response.url, response.json()), + }[response.status_code] + except KeyError: + raise exceptions.OtherHTTPResponse('Not excepted HTTP error code!', response.status_code) + + return response.json() + + def get(self, *path, request_params=None): + """ + Wykonuje odpowiednio spreparowane zapytanie http GET. + + :param path: Ścieżka zawierająca węzeł API + :type path: str + :param request_params: dict zawierający kwargs dla zapytania http + :type request_params: dict + :return: json przekonwertowany na dict'a + :rtype: dict + """ + if request_params is None: + request_params = dict() + path_str = self.assembly_path(*path, prefix=self.__api_url) + response = self.session.get( + path_str, headers=self.__auth_headers, params=request_params + ) + + response = self.dispatch_http_code(response, callback=self.get, callback_args=path, + callback_kwargs=request_params) + + return response + + def post(self, *path, request_params=None): + """ + Pozwala na dokonanie zapytania http POST. + + :param path: Ścieżka zawierająca węzeł API + :type path: str + :param request_params: dict zawierający kwargs dla zapytania http + :type request_params: dict + :return: json przekonwertowany na dict'a + :rtype: dict + """ + if request_params is None: + request_params = dict() + path_str = self.assembly_path(*path, prefix=self.__api_url) + response = self.session.post( + path_str, headers=self.__auth_headers, params=request_params + ) + + response = self.dispatch_http_code(response, callback=self.post, callback_args=path, + callback_kwargs=request_params) + + return response + + # Cache + + def get_cached_response(self, *path, http_params=None, max_lifetime=timedelta(hours=1)): + """ + Wykonuje zapytanie http GET z poprzednim sprawdzeniem cache. + + :param path: Niezłożona ścieżka do węzła API + :param http_params: dict zawierający kwargs dla zapytania http + :type http_params: dict + :param timedelta max_lifetime: Maksymalny czas ważności cache dla tego zapytania http + :return: dict zawierający odpowiedź zapytania + :rtype: dict + """ + uri = self.assembly_path(*path, prefix=self.__api_url) + response_cached = self.cache.get_query(uri, self.user.uid) + + if response_cached is None: + logging.debug('Response is not present in cache!') + http_response = self.get(*path, request_params=http_params) + self.cache.add_query(uri, http_response, self.user.uid) + return http_response + + try: + age = datetime.now() - response_cached.last_load + except TypeError: + age = datetime.now() - response_cached.last_load.replace(tzinfo=None) + + if age > max_lifetime: + logging.debug('Response is too old! Trying to get latest response from api') + http_response = self.get(*path, request_params=http_params) + self.cache.del_query(uri, self.user.uid) + self.cache.add_query(uri, http_response, self.user.uid) + return http_response + return response_cached.response + + def get_cached_object(self, uid, cls, max_lifetime=timedelta(hours=1)): + """ + Pobiera dany obiekt z poprzednim sprawdzeniem cache. Nie używane domyślnie w bibliotece. + + :param str uid: Id żądanego obiektu + :param cls: Klasa żądanego obiektu + :param timedelta max_lifetime: Maksymalny czas ważności cache dla tego obiektu + :return: Żądany obiekt + """ + requested_object = self.cache.get_object(uid, cls) + + if requested_object is None: + logging.debug('Obejct is not present in cache!') + requested_object = cls.create(uid=uid, session=self) + self.cache.add_object(uid, cls, requested_object.export_resource()) + return requested_object + + try: + age = datetime.now() - requested_object.last_load + except TypeError: + age = datetime.now() - requested_object.last_load.replace(tzinfo=None) + + if age > max_lifetime: + logging.debug('Object is too old! Trying to get latest object from api') + requested_object = cls.create(uid=uid, session=self) + self.cache.del_object(uid) + self.cache.add_object(uid, cls, requested_object.export_resource()) + + return requested_object + + # API query part + + def return_objects(self, *path, cls, extraction_key=None, lifetime=timedelta(seconds=10), bypass_cache=False): + """ + Zwraca listę obiektów lub obiekt, wygenerowaną z danych danej ścieżki API. + + :param str path: Niezłożona ścieżka do węzła API + :param cls: Klasa żądanych obiektów + :param str extraction_key: Klucz do wyjęcia danych, pozostawienie tego parametru na None powoduje + automatyczną próbę zczytania danych + :param timedelta lifetime: Maksymalny czas ważności cache dla tego zapytania http + :param bool bypass_cache: Ustawienie tego parametru na True powoduje ignorowanie mechanizmu cache + :return: Lista żądanych obiektów + :rtype: list[SynergiaGenericClass] + """ + if bypass_cache: + raw = self.get(*path) + else: + raw = self.get_cached_response(*path, max_lifetime=lifetime) + + if extraction_key is None: + extraction_key = SynergiaGenericClass.auto_extract(raw) + + raw = raw[extraction_key] + + if isinstance(raw, list): + stack = [] + for stored_payload in raw: + stack.append(cls.assembly(stored_payload, self)) + return tuple(stack) + if isinstance(raw, dict): + return cls.assembly(raw, self) + + return None + + def grades(self, *grades): + """ + :param int grades: Id ocen + :rtype: tuple[librus_tricks.classes.SynergiaGrade] + :return: krotka z wszystkimi/wybranymi ocenami + """ + if grades.__len__() == 0: + return self.return_objects('Grades', cls=SynergiaGrade, extraction_key='Grades') + ids_computed = self.assembly_path(*grades, sep=',', suffix=',')[1:] + return self.return_objects('Grades', ids_computed, cls=SynergiaGrade, extraction_key='Grades') + + @property + def grades_categorized(self): + grades_categorized = {} + for subject in self.subjects(): + grades_categorized[subject.name] = [] + + for grade in self.grades(): + grades_categorized[grade.subject.name].append( + grade + ) + + for subjects in grades_categorized.copy().keys(): + if grades_categorized[subjects].__len__() == 0: + del (grades_categorized[subjects]) + + return grades_categorized + + def attendances(self, *attendances): + """ + :param int attendances: Id obecności + :rtype: tuple[librus_tricks.classes.SynergiaAttendance] + :return: krotka z wszystkimi/wybranymi obecnościami + """ + if attendances.__len__() == 0: + return self.return_objects('Attendances', cls=SynergiaAttendance, extraction_key='Attendances') + ids_computed = self.assembly_path(*attendances, sep=',', suffix=',')[1:] + return self.return_objects('Attendances', ids_computed, cls=SynergiaAttendance, extraction_key='Attendances') + + @property + def illegal_absences(self): + """ + :rtype: tuple[librus_tricks.classes.SynergiaAttendance] + :return: krotka z nieusprawiedliwionymi nieobecnościami + """ + + def is_absence(k): + if k.type.uid == '1': + return True + return False + + return tuple(filter(is_absence, self.attendances())) + + @property + def all_absences(self): + """ + :rtype: tuple[librus_tricks.classes.SynergiaAttendance] + :return: krotka z wszystkimi nieobecnościami + """ + return tuple(filter(lambda k: not k.type.is_presence_kind, self.attendances())) + + def exams(self, *exams): + """ + :param int exams: Id egzaminów + :rtype: tuple[librus_tricks.classes.SynergiaExam] + :return: krotka z wszystkimi egzaminami + """ + if exams.__len__() == 0: + return self.return_objects('HomeWorks', cls=SynergiaExam, extraction_key='HomeWorks') + ids_computed = self.assembly_path(*exams, sep=',', suffix=',')[1:] + return self.return_objects('HomeWorks', ids_computed, cls=SynergiaExam, extraction_key='HomeWorks') + + def colors(self, *colors): + """ + :param int colors: Id kolorów + :rtype: tuple[librus_tricks.classes.SynergiaColors] + """ + if colors.__len__() == 0: + return self.return_objects('Colors', cls=SynergiaColor, extraction_key='Colors') + ids_computed = self.assembly_path(*colors, sep=',', suffix=',')[1:] + return self.return_objects('Colors', ids_computed, cls=SynergiaColor, extraction_key='Colors') + + def timetable(self, for_date=datetime.now()): + """ + Plan lekcji na cały tydzień. + + :param datetime.datetime for_date: Data dnia, który ma być w planie lekcji + :rtype: librus_tricks.classes.SynergiaTimetable + :return: obiekt tygodniowego planu lekcji + """ + monday = tools.get_actual_monday(for_date).isoformat() + matrix = self.get('Timetables', request_params={'weekStart': monday}) + return SynergiaTimetable.assembly(matrix['Timetable'], self) + + def timetable_day(self, for_date: datetime): + return self.timetable(for_date).days[for_date.date()] + + @property + def today_timetable(self): + """ + Plan lekcji na dzisiejszy dzień. + + :rtype: librus_tricks.classes.SynergiaTimetableDay + :return: Plan lekcji na dziś + """ + try: + return self.timetable().days[datetime.now().date()] + except KeyError: + return None + + @property + def tomorrow_timetable(self): + """ + Plan lekcji na kolejny dzień. + + :rtype: librus_tricks.classes.SynergiaTimetableDay + :return: Plan lekcji na jutro + """ + try: + return self.timetable(datetime.now() + timedelta(days=1)).days[(datetime.now() + timedelta(days=1)).date()] + except KeyError: + return None + + def messages(self, *messages): + """ + Wymaga mobilnych dodatków. + + :param int messages: Id wiadomości + :rtype: tuple[librus_tricks.classes.SynergiaNativeMessage] + """ + if messages.__len__() == 0: + return self.return_objects('Messages', cls=SynergiaNativeMessage, extraction_key='Messages') + ids_computed = self.assembly_path(*messages, sep=',', suffix=',')[1:] + return self.return_objects('Messages', ids_computed, cls=SynergiaNativeMessage, extraction_key='Messages') + + def news_feed(self): + """ + :return: Wszystkie ogłoszenia szkolne + :rtype: tuple[librus_tricks.classes.SynergiaNews] + """ + return self.return_objects('SchoolNotices', cls=SynergiaNews, extraction_key='SchoolNotices') + + def subjects(self, *subject): + """ + :return: Wszystkie/wybrane przedmioty lekcyjne + :param int subject: Id przedmiotów + :rtype: tuple[librus_tricks.classes.SynergiaSubject] + """ + if subject.__len__() == 0: + return self.return_objects('Subjects', cls=SynergiaSubject, extraction_key='Subjects') + ids_computed = self.assembly_path(*subject, sep=',', suffix=',')[1:] + return self.return_objects('Subjects', ids_computed, cls=SynergiaSubject, extraction_key='Subjects') + + @property + def school(self): + """ + :return: Obiekt z informacjami o twojej szkole + :rtype: librus_tricks.classes.SynergiaSchool + """ + return self.return_objects('Schools', cls=SynergiaSchool, extraction_key='School') + + @property + def lucky_number(self): + """ + :return: Szczęśliwy numerek + :rtype: int + """ + return self.get('LuckyNumbers')['LuckyNumber']['LuckyNumber'] + + @staticmethod + def __is_future(day): + """ + + :type day: librus_tricks.classes.SynergiaTeacherFreeDays + :return: + """ + if day.ends >= datetime.now().date(): + return True + return False + + def teacher_free_days(self, *days_ids, only_future=True): + """ + Zwraca dane przedmioty. + + :param int days_ids: Id zwolnień + :rtype: tuple[librus_tricks.classes.SynergiaTeacherFreeDays] + """ + if days_ids.__len__() == 0: + days = self.return_objects('Calendars', 'TeacherFreeDays', cls=SynergiaTeacherFreeDays) + else: + ids_computed = self.assembly_path(*days_ids, sep=',', suffix=',')[1:] + days = self.return_objects('Calendars', 'TeacherFreeDays', ids_computed, cls=SynergiaTeacherFreeDays) + + days = tuple(sorted(days, key=lambda x: x.starts)) + if only_future: + return tuple(filter(self.__is_future, days)) + return days + + def school_free_days(self, *days_ids, only_future=True): + if days_ids.__len__() == 0: + days = self.return_objects('Calendars', 'SchoolFreeDays', cls=SynergiaSchoolFreeDays) + else: + ids_computed = self.assembly_path(*days_ids, sep=',', suffix=',')[1:] + days = self.return_objects('Calendars', 'SchoolFreeDays', ids_computed, cls=SynergiaSchoolFreeDays) + + days = tuple(sorted(days, key=lambda x: x.starts)) + if only_future: + return tuple(filter(self.__is_future, days)) + return days + + def realizations(self, *realizations_ids): + if realizations_ids.__len__() == 0: + return self.return_objects('Realizations', cls=SynergiaRealization, extraction_key='Realizations') + ids_computed = self.assembly_path(*realizations_ids, sep=',', suffix=',')[1:] + return self.return_objects('Realizations', ids_computed, cls=SynergiaRealization, extraction_key='Realizations') + + def substitutions(self): + pass + + def preload_cache(self): + self.cache.clear_objects() + + for thing in (*self.attendances(), *self.grades(), *self.subjects(), *self.school_free_days(only_future=False), + *self.teacher_free_days(only_future=False)): + self.cache.add_object(thing.uid, thing.__class__, thing.export_resource()) + + logging.info('Loaded %s objects into cache', self.cache.count_object()) diff --git a/exceptions.py b/exceptions.py new file mode 100644 index 0000000..6506ed5 --- /dev/null +++ b/exceptions.py @@ -0,0 +1,82 @@ +class LibrusTricksException(Exception): + pass + + +class LibrusTricksAuthException(LibrusTricksException): + pass + + +class LibrusTricksWrapperException(LibrusTricksAuthException): + pass + + +class LibrusLoginError(LibrusTricksAuthException): + pass + + +class SynergiaAPIEndpointNotFound(LibrusTricksWrapperException): + pass + + +class LibrusPortalInvalidPasswordError(LibrusTricksAuthException): + pass + + +class SynergiaAccessDenied(LibrusTricksWrapperException): + pass + + +class WrongHTTPMethod(Exception): + pass + + +class SynergiaAPIInvalidRequest(LibrusTricksWrapperException): + pass + + +class TokenExpired(LibrusTricksException): + pass + + +class SynergiaForbidden(LibrusTricksWrapperException): + pass + + +class InvalidCacheManager(LibrusTricksAuthException): + pass + + +class CaptchaRequired(LibrusTricksAuthException): + pass + + +class SynergiaServerError(LibrusTricksWrapperException): + pass + + +class SessionRequired(LibrusTricksWrapperException): + pass + + +class APIPathIsEmpty(LibrusTricksWrapperException): + pass + + +class OtherHTTPResponse(LibrusTricksWrapperException): + pass + + +class SecurityWarning(Warning): + pass + + +class PerformanceWarning(Warning): + pass + + +class GoodPracticeWarning(Warning): + pass + + +class SynergiaMaintenanceError(SynergiaServerError): + pass diff --git a/messages.py b/messages.py new file mode 100644 index 0000000..e0bfe8c --- /dev/null +++ b/messages.py @@ -0,0 +1,82 @@ +from datetime import datetime +import logging + +import requests +from bs4 import BeautifulSoup + + +class SynergiaScrappedMessage: + def __init__(self, url, parent_web_session, header, author, message_date, synergia_session): + """ + :type url: str + :type parent_web_session: requests.sessions.Session + :type message_date: datetime + :type synergia_session: librus_tricks.core.SynergiaClient + """ + self.web_session = parent_web_session + self.url = url + self.header = header + self.author_alias = author + self.msg_date = message_date + self.synergia_session = synergia_session + + def __read_from_server(self): + response = self.web_session.get('https://synergia.librus.pl' + self.url) + soup = BeautifulSoup(response.text, 'html.parser') + return soup.find('div', attrs={'class': 'container-message-content'}).text + + @property + def text(self): + return self.__read_from_server() + + @property + def author(self): + from librus_tricks.classes import SynergiaTeacher + teachers = self.synergia_session.return_objects('Users', cls=SynergiaTeacher, extraction_key='Users') + for teacher in teachers: + if str(teacher.name) in self.author_alias and str(teacher.last_name) in self.author_alias: + return teacher + return + + def __repr__(self): + return f'' + + +class MessageReader: + def __init__(self, session): + """ + + :param librus_tricks.core.SynergiaClient session: + """ + self._syn_session = session + self._web_session = requests.session() + logging.debug('Obtain AutoLoginToken from server') + token = session.post('AutoLoginToken')['Token'] + self._web_session.get(f'https://synergia.librus.pl/loguj/token/{token}/przenies/wiadomosci') + logging.debug('Webscrapper logged into Synergia') + + def read_messages(self): + response = self._web_session.get('https://synergia.librus.pl/wiadomosci') + soup = BeautifulSoup(response.text, 'html.parser') + table = soup.find('table', attrs={'class': 'decorated stretch'}) + tbody = table.find('tbody') + + if 'Brak wiadomości' in tbody.text: + return None + + rows = tbody.find_all('tr') + messages = [] + for message in rows: + cols = message.find_all('td') + messages.append(SynergiaScrappedMessage( + url=cols[3].a['href'], + header=cols[3].text.strip(), + author=cols[2].text.strip(), + parent_web_session=self._web_session, + message_date=datetime.strptime(cols[4].text, '%Y-%m-%d %H:%M:%S'), + synergia_session=self._syn_session + )) + return messages + + def __repr__(self): + return f'<{self.__class__.__name__} for {self._syn_session.user}>' diff --git a/tools.py b/tools.py new file mode 100644 index 0000000..9dc7aef --- /dev/null +++ b/tools.py @@ -0,0 +1,185 @@ +from datetime import datetime, timedelta +import re + + +def get_next_monday(now=datetime.now()): + for _ in range(8): + if now.weekday() == 0: + return now.date() + now = now + timedelta(days=1) + return + + +def get_actual_monday(now=datetime.now()): + for _ in range(8): + if now.weekday() == 0: + return now.date() + now = now - timedelta(days=1) + return + + +def extract_percentage(grade): + """ + + :param librus_tricks.classes.SynergiaGrade grade: + :return: + """ + for comment in grade.comments: + matches = re.findall(r'(\d+)%', comment.text) + if matches.__len__() > 0: + return float(matches[0]) + return + + +def weighted_average(*grades_and_weights): + values = 0 + count = 0 + for grade_weight in grades_and_weights: + count += grade_weight[1] + for _ in range(grade_weight[1]): + values += grade_weight[0] + if count == 0: + return 0 + return values / count + + +def extracted_percentages(grades): + grades = [grade for grade in grades if extract_percentage(grade) is not None] + subjects = set([grade.subject.name for grade in grades]) + categorized = {} + for subject in subjects: + categorized[subject] = [] + for grade in grades: + categorized[grade.subject.name].append((grade, extract_percentage(grade))) + return categorized + + +def no_cache(func): + def wrapper(*args, **kwargs): + return func(*args, **{**kwargs, 'expire': 0}) + + return wrapper + + +def percentage_average(grades, generic_top_value=5): + def compare_lists(list_a, list_b): + result = [] + for item in list_a: + if item not in list_b: + result.append(item) + return result + + percentages = extracted_percentages(grades) + if percentages.keys().__len__() == 0: + return {} + generics = compare_lists(grades, [grade_weight_tuple[0] for grade_weight_tuple in tuple(percentages.values())[0]]) + averages = {} + for subject_name in percentages: + averages[subject_name] = weighted_average( + *[(grade_percent[1], grade_percent[0].category.weight) for grade_percent in percentages[subject_name]], + *[((generic.real_value / generic_top_value) * 100, generic.category.weight) for generic in generics if + generic_top_value is not None or not False if generic.real_value is not None if + generic.subject.name == subject_name] + ) + return averages + + +def subjects_averages(subject_keyed_grades): + averages = {} + for subject in subject_keyed_grades: + averages[subject] = weighted_average( + *[(grade.real_value, grade.category.weight) for grade in subject_keyed_grades[subject] if + grade.real_value is not None] + ) + + return averages + + +def count_attendances(attendances): + """ + + :param iterable[librus_tricks.classes.SynergiaAttendance] attendances: + :return: + """ + categories = set() + for attendance in attendances: + categories.add(attendance.type) + + results = {} + for cat in categories: + results[cat] = 0 + + for attendance in attendances: + results[attendance.type] += 1 + + return results + + +def present_percentage(attendances): + """ + + :param list[librus_tricks.classes.SynergiaAttendance] attendances: + :return: + """ + present = 0 + absent = 0 + for attendance in attendances: + if attendance.type.is_presence_kind: + present += 1 + else: + absent += 1 + + return present / attendances.__len__() * 100 + + +def percentages_of_attendances(attendances): + """ + + :param list[librus_tricks.classes.SynergiaAttendance] attendances: + :return: + """ + results = count_attendances(attendances) + for category in results: + results[category] = results[category] / attendances.__len__() * 100 + + return results + + +def attendance_per_subject(attendances): + """ + :type attendances: list of librus_tricks.classes.SynergiaAttendance + """ + subjects = set() + att_types = set() + + for att in attendances: + subjects.add( + att.lesson.subject + ) + att_types.add( + att.type + ) + + attendances_by_subject = dict() + + for sub in subjects: + attendances_by_subject[sub] = dict() + for attyp in att_types: + attendances_by_subject[sub][attyp] = list() + + for att in attendances: + attendances_by_subject[att.lesson.subject][att.type].append( + att + ) + + redundant = [] + + for subject in attendances_by_subject: + for at_type in attendances_by_subject[subject]: + if attendances_by_subject[subject][at_type].__len__() == 0: + redundant.append((subject, at_type)) + + for n in redundant: + del(attendances_by_subject[n[0]][n[1]]) + + return attendances_by_subject