# You should have received a copy of the GNU General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
-from collections import namedtuple
import json
from pathlib import PurePosixPath
-from urllib.parse import urlencode as mk_qs
import urllib.request
-import auth
import util
def walk_pages(opener, url):
import urllib.parse
import urllib.request
-import common
+from common import log
import util
ADDRESS = 'localhost'
TIMEOUT = timedelta(minutes = 30)
-LOGGER = common.LOGGER.getChild('auth')
-
class OauthCodeRequestHandler(http.server.BaseHTTPRequestHandler):
LANDING_PATH = '/landing'
def _handle(self, request):
token_doc = self.storage_mgr.get('authInfo')
if not token_doc:
- LOGGER.info("No stored access token. Requesting a new token.")
+ log('info', "No stored access token. Requesting a new token.")
token_doc = get_access_token(self.storage_mgr, self.api_iface)
self.storage_mgr['authInfo'] = token_doc
self._set_auth_header(request, token_doc)
https_request = _handle
def http_error_401(self, request, fp, code, msg, headers):
- LOGGER.info("Access token expired or revoked. Requesting a new token.")
+ log('info', "Access token expired or revoked. Requesting a new token.")
token_doc = get_access_token(self.storage_mgr, self.api_iface)
self.storage_mgr['authInfo'] = token_doc
self._set_auth_header(request, token_doc)
if resp.status == 200:
token_doc = body
else:
- LOGGER.info("Stored refresh token rejected. Obtaining new authorization code.")
+ log('info', "Stored refresh token rejected. Obtaining new authorization code.")
assert resp.status == 400
if token_doc is None:
)
Thread(target = server.serve_forever, daemon = True).start()
- LOGGER.info("Attempting to launch a web browser to authorize the application…")
+ log('info', "Attempting to launch a web browser to authorize the application…")
if not webbrowser.open(user_url):
- LOGGER.info("Failed to launch a browser. Please visit this URL to authorize the application:")
- LOGGER.info(" {}".format(user_url))
+ log('info', "Failed to launch a browser. Please visit this URL to authorize the application:")
+ log('info', user_url, indent = 1)
try:
resp = channel.get(timeout = TIMEOUT.total_seconds())
self.cookiejar.save()
def http_error_401(self, request, fp, code, msg, headers):
- LOGGER.info("Session cookies missing or expired. Logging in…")
+ log('info', "Session cookies missing or expired. Logging in…")
self.log_in()
return self.parent.open(request, timeout = request.timeout)
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
-from collections import deque, namedtuple
+from collections import namedtuple
import json
-import logging
from pathlib import PurePosixPath
-import urllib.request
+
BlackboardRoot = namedtuple('BlackboardRoot', ['host', 'path'])
BB_ROOT = BlackboardRoot(
path = PurePosixPath('/'),
)
-LOGGER = logging.getLogger('bb-sync-api')
-logging.basicConfig()
-LOGGER.setLevel(logging.INFO)
-
-class Adapter(logging.LoggerAdapter):
- def process(self, msg, kwargs):
- return (' ' * kwargs['indent'] + msg, kwargs)
-
-'''
-LOGGER = Adapter(LOGGER, {})
-LOGGER.debug('some message', indent = 1)
-class IndentingFormatter(logging.Formatter):
- def format(self, record):
- prefix = ' ' * record.indent if hasattr(record, 'indent') else ''
- return prefix + None
+def log(level, message, indent = 0):
+ import sys
+ print(level.upper().rjust(7) + ': ' + indent * ' ' + message, file = sys.stderr)
-LOGGER.setFormatter(logging.Formatter('', style = '{'))
-'''
class StorageManager():
def __init__(self, path):
from operator import attrgetter
from pathlib import Path
import shutil
-import sys
import tempfile
import urllib.parse
-import urllib.request
from common import *
import api
found_course_id = None
if found_course_id:
if found_course_id != course_id:
- LOGGER.warning("Using a course root previously used for another course!")
- LOGGER.warning("File versioning may misbehave.")
+ log('warning', "Using a course root previously used for another course!")
+ log('warning', "File versioning may misbehave.")
else:
with meta_path.open('x') as f:
json.dump({'course_id': course_id}, f)
content_path = api_iface.get_content_path(course_id, content_id)
- LOGGER.info("Blackboard content path: {}".format('/'.join(seg['id'] for seg in content_path)))
+ log('info', "Blackboard content path: {}".format('/'.join(seg['id'] for seg in content_path)))
local_content_root = functools.reduce(fs.join_content_path, content_path, local_course_root)
- LOGGER.info("Local content path: {}".format(local_content_root))
+ log('info', "Local content path: {}".format(local_content_root))
for child_doc in api_iface.get_children(course_id, content_id):
- LOGGER.info("Processing content item {id}: \"{title}\"".format(**child_doc))
+ log('info', "Processing content item {id}: \"{title}\"".format(**child_doc))
local_path = fs.join_content_path(local_content_root, child_doc)
versions = fs.get_child_versions(local_path)
for attachment_doc in api_iface.get_attachments(course_id, child_doc['id']):
att_id = attachment_doc['id']
- LOGGER.info(" Checking attachment {id}: \"{fileName}\"".format(**attachment_doc))
+ log('info', "Checking attachment {id}: \"{fileName}\"".format(**attachment_doc), indent = 1)
class Result:
NoVersions = namedtuple('NoVersions', [])
if Result.to_update(result):
if isinstance(result, Result.SingleLatest):
new_version = result.version.next()
- LOGGER.info(" Found new revision ({})".format(new_version.version))
+ log('info', "Found new revision ({})".format(new_version.version), indent = 2)
else:
new_version = fs.VersionInfo(att_id, 0)
- LOGGER.info(" Storing initial revision")
+ log('info', "Storing initial revision", indent = 2)
dest = fs.get_new_path(local_path, attachment_doc['fileName'])
- LOGGER.info(" Destination: {}".format(dest))
+ log('info', "Destination: {}".format(dest), indent = 2)
tmp_path.replace(dest)
fs.write_metadata(dest, new_version)
elif isinstance(result, Result.SingleLatest):
# versions match
tmp_path.unlink()
elif isinstance(result, Result.MultipleLatest):
- LOGGER.error(" Identified multiple latest versions for {id}: {fileName}"
- .format(**attachment_doc)
+ log(
+ 'error',
+ "Identified multiple latest versions for {id}: {fileName}".format(**attachment_doc),
+ indent = 2,
)
- LOGGER.error(" Please delete ")
+ log(
+ 'error',
+ "To allow the program to check this attachment, delete all or all but one of these files:",
+ indent = 2,
+ )
+ for path_str in sorted(map(str, result.paths)):
+ log('error', path_str, indent = 3)
+ else:
+ raise AssertionError()
projects / bb-sync-api.git / commitdiff