import requests from requests.exceptions import ConnectionError import os.path import xml.etree.ElementTree as etree from xml.parsers.expat import ExpatError import collections import logging CourseByOwnerResponse = collections.namedtuple('CourseByOwnerResponse', ('message', 'courses')) Course = collections.namedtuple('Course', ('id', 'name', 'available')) GetCategoryResponse = collections.namedtuple('GetCategoryResponse', ('message', 'categories')) Category = collections.namedtuple('Category', ('id', 'name')) logger = logging.getLogger(__name__) PARSE_ERRORS = (ExpatError, ) try: from xml.etree.ElementTree import ParseError PARSE_ERRORS += (ParseError,) except ImportError: pass def bool2str(b): return 'YES' if b else 'NO' class BlackboardConnector(object): def __init__(self, url, login=None, password=None, shared_key=None): self.url = url self.login = login self.password = password self.shared_key = shared_key self.session = requests.session() url = os.path.join(self.url, 'login/') try: self.session.get(url) self.session.post(url, data=dict(user_id=self.login, password=self.password, action='login', login='Connexion')) except ConnectionError: pass def _get_course_helper(self, r): r.encoding = 'utf-8' if r.status_code == 200 and r.content.startswith('<'): try: x = etree.XML(r.content.replace('ISO-8859-1', 'UTF-8')) except PARSE_ERRORS: logger.exception('XML Parsing error request to %s', r.url) return False, 'Erreur BlackBoard' info_message_elt = x.find('infoMessage') course_elts = x.findall('*/course') courses = [] for course_elt in course_elts: id_elt = course_elt.find('ID') name_elt = course_elt.find('name') available_elt = course_elt.find('available') if None in (id_elt, name_elt, available_elt) or available_elt.text not in ('YES', 'NO'): logger.error('invalid answer to GetCourseByOwner: %r', r.text) return False, 'Erreur BlackBoard' course = Course(id_elt.text.strip(), name_elt.text.strip(), available_elt.text == 'YES') courses.append(course) return True, CourseByOwnerResponse(info_message_elt.text.strip(), courses) elif r.status_code in (200,400): return False, r.text else: logger.error('BlackBoard code HTTP %s', r.status_code) return False, 'Erreur BlackBoard' def get_course_by_owner(self, user_login): try: r = self.send_request('GetCoursesByOwner', user_login=user_login.encode('ascii')) except ConnectionError: return True, [] return self._get_course_helper(r) def get_course_by_ue(self, entity_code): try: r = self.send_request('GetCoursesByUE', UE_ID=entity_code.encode('ascii')) except ConnectionError: return True, [] return self._get_course_helper(r) def send_file(self, uploadfile, courses_id, visible): assert hasattr(uploadfile, 'read') assert hasattr(courses_id, '__iter__') assert len(courses_id) > 0 assert all(map(bool, courses_id)) visible = bool2str(visible) r = self.send_request('SendFile', course_ID=courses_id, files=dict(uploadfile=uploadfile), method='post', visible=visible) if r.status_code == 200: return True, r.text return False, r.text def create_course(self, course_id, course_name, category, open_to_visitors=True, auto_inscription=True, password=None): assert isinstance(course_id, unicode) and course_id assert isinstance(course_name, unicode) and course_name assert isinstance(category, unicode) and unicode open_to_visitors = bool2str(open_to_visitors) auto_inscription = bool2str(auto_inscription) params = { 'course_ID': course_id, 'course_name': course_name, 'open_to_visitors': open_to_visitors, 'auto_inscription': auto_inscription, 'category': category, } if password: params['password'] = password r = self.send_request('CreateCourse', **params) if r.status_code == 200: return True, r.text return False, r.text def send_request(self, request_name, **params): params['secret_key'] = self.shared_key kwargs = { 'params': params } files = params.pop('files', None) method = params.pop('method', 'get') if files: kwargs['files'] = files url = os.path.join(self.url, 'ui-ui-Polynum-BBLEARN/app', request_name) logger.debug('sending request %s', url) r = getattr(self.session, method)(url, **kwargs) logger.debug('got response %r', r.content.encode('quopri')) r.encoding = 'utf-8' return r def get_categories(self): try: r = self.send_request('GetCategories') except ConnectionError: return True, GetCategoryResponse('', []) if r.status_code == 200 and r.content.startswith('<'): try: x = etree.XML(r.content.replace('ISO-8859-1', 'UTF-8')) except PARSE_ERRORS: logger.exception('XML Parsing error request to %s', r.url) return False, 'Erreur BlackBoard' info_message_elt = x.find('infoMessage') category_elts = x.findall('categories/category') categories = [] for category_elt in category_elts: id_elt = category_elt.find('ID') name_elt = category_elt.find('name') category = Category(id_elt.text.strip(), name_elt.text.strip()) categories.append(category) return True, GetCategoryResponse(info_message_elt.text.strip(), categories) elif r.status_code in (200,400): return False, r.text else: logger.error('BlackBoard code HTTP %s', r.status_code) logger.error(r.text.replace('\n', '\\n')) return False, 'Erreur BlackBoard {0}'.format(r.status_code) if __name__ == '__main__': import sys import logging logging.basicConfig(level=logging.DEBUG) try: connector = BlackboardConnector(sys.argv[1], *sys.argv[2:5]) if len(sys.argv) == 7 and sys.argv[5] == 'GetCoursesByOwner': ok, response = connector.get_course_by_owner(sys.argv[6]) elif len(sys.argv) == 7 and sys.argv[5] == 'GetCoursesByUE': ok, response = connector.get_course_by_ue(sys.argv[6]) elif len(sys.argv) == 8 and sys.argv[5] == 'SendFile': ok, text = connector.send_file(file(sys.argv[6]), sys.argv[7].split(',')) elif len(sys.argv) == 6 and sys.argv[5] == 'GetCategories': ok, response = connector.get_categories() else: raise ValueError if ok: print 'OK' else: print 'NOK' print response except Exception, e: raise print 'Syntax: python', sys.argv[0], ' [GetCoursesByOwner |GetCoursesByUE |SendFile ]'