[added] The new --verbose command line argument turns on more logging.

.gitignore

@@ -1 +1,2 @@
 /.idea/
+/kobo-book-downloader/__pycache__/

kobo-book-downloader/Commands.py

@@ -208,8 +208,6 @@ def __GetBookList( listAll: bool ) -> list:
 
 	@staticmethod
 	def ListBooks( listAll: bool ) -> None:
-		colorama.init()
-
 		rows = Commands.__GetBookList( listAll )
 		for columns in rows:
 			revisionId = colorama.Style.DIM + columns[ 0 ] + colorama.Style.RESET_ALL
@@ -283,17 +281,13 @@ def __DownloadPickedBooks( outputPath: str, rows: list ) -> None:
 
 	@staticmethod
 	def PickBooks( outputPath: str, listAll: bool ) -> None:
-		colorama.init()
-
 		rows = Commands.__GetBookList( listAll )
 		Commands.__ListBooksToPickFrom( rows )
 		rowsToDownload = Commands.__GetPickedBookRows( rows )
 		Commands.__DownloadPickedBooks( outputPath, rowsToDownload )
 
 	@staticmethod
 	def ListWishListedBooks() -> None:
-		colorama.init()
-
 		rows = []
 
 		wishList = Globals.Kobo.GetMyWishList()

kobo-book-downloader/Globals.py

@@ -1,3 +1,11 @@
+from logging import Logger
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+	from Kobo import Kobo
+	from Settings import Settings
+
 class Globals:
-	Kobo = None
-	Settings = None
+	Logger = None # type: Logger | None
+	Kobo = None # type: Kobo | None
+	Settings = None # type: Settings | None

kobo-book-downloader/Kobo.py

@@ -20,7 +20,7 @@ def ReauthenticationHook( r, *args, **kwargs ):
 	if r.status_code != requests.codes.unauthorized: # 401
 		return
 
-	print( "Refreshing expired authentication token" )
+	Globals.Logger.debug( "Refreshing expired authentication token" )
 
 	# Consume content and release the original connection to allow our new request to reuse the same one.
 	r.content
@@ -81,6 +81,8 @@ def __GetReauthenticationHook() -> dict:
 	# The initial device authentication request for a non-logged in user doesn't require a user key, and the returned
 	# user key can't be used for anything.
 	def AuthenticateDevice( self, userKey: str = "" ) -> None:
+		Globals.Logger.debug( "Kobo.AuthenticateDevice" )
+
 		if len( Globals.Settings.DeviceId ) == 0:
 			Globals.Settings.DeviceId = str( uuid.uuid4() )
 			Globals.Settings.AccessToken = ""
@@ -115,6 +117,8 @@ def AuthenticateDevice( self, userKey: str = "" ) -> None:
 		Globals.Settings.Save()
 
 	def RefreshAuthentication( self ) -> None:
+		Globals.Logger.debug( "Kobo.RefreshAuthentication" )
+
 		headers = Kobo.GetHeaderWithAccessToken()
 
 		postData = {
@@ -140,6 +144,8 @@ def RefreshAuthentication( self ) -> None:
 		Globals.Settings.Save()
 
 	def LoadInitializationSettings( self ) -> None:
+		Globals.Logger.debug( "Kobo.LoadInitializationSettings" )
+
 		headers = Kobo.GetHeaderWithAccessToken()
 		hooks = Kobo.__GetReauthenticationHook()
 		response = self.Session.get( "https://storeapi.kobo.com/v1/initialization", headers = headers, hooks = hooks )
@@ -148,6 +154,8 @@ def LoadInitializationSettings( self ) -> None:
 		self.InitializationSettings = jsonResponse[ "Resources" ]
 
 	def __GetExtraLoginParameters( self ) -> Tuple[ str, str, str ]:
+		Globals.Logger.debug( "Kobo.__GetExtraLoginParameters" )
+
 		signInUrl = self.InitializationSettings[ "sign_in_page" ]
 
 		params = {
@@ -178,6 +186,8 @@ def __GetExtraLoginParameters( self ) -> Tuple[ str, str, str ]:
 		return koboSignInUrl, workflowId, requestVerificationToken
 
 	def Login( self, email: str, password: str, captcha: str ) -> None:
+		Globals.Logger.debug( "Kobo.Login" )
+
 		signInUrl, workflowId, requestVerificationToken = self.__GetExtraLoginParameters()
 
 		postData = {
@@ -208,6 +218,8 @@ def Login( self, email: str, password: str, captcha: str ) -> None:
 		self.AuthenticateDevice( userKey )
 
 	def GetBookInfo( self, productId: str ) -> dict:
+		Globals.Logger.debug( "Kobo.GetBookInfo" )
+
 		url = self.InitializationSettings[ "book" ].replace( "{ProductId}", productId )
 		headers = Kobo.GetHeaderWithAccessToken()
 		hooks = Kobo.__GetReauthenticationHook()
@@ -218,6 +230,8 @@ def GetBookInfo( self, productId: str ) -> dict:
 		return jsonResponse
 
 	def __GetMyBookListPage( self, syncToken: str ) -> Tuple[ list, str ]:
+		Globals.Logger.debug( "Kobo.__GetMyBookListPage" )
+
 		url = self.InitializationSettings[ "library_sync" ]
 		headers = Kobo.GetHeaderWithAccessToken()
 		hooks = Kobo.__GetReauthenticationHook()
@@ -252,6 +266,8 @@ def GetMyBookList( self ) -> list:
 		return fullBookList
 
 	def GetMyWishList( self ) -> list:
+		Globals.Logger.debug( "Kobo.GetMyWishList" )
+
 		items = []
 		currentPageIndex = 0
 
@@ -278,6 +294,8 @@ def GetMyWishList( self ) -> list:
 		return items
 
 	def __GetContentAccessBook( self, productId: str, displayProfile: str ) -> dict:
+		Globals.Logger.debug( "Kobo.__GetContentAccessBook" )
+
 		url = self.InitializationSettings[ "content_access_book" ].replace( "{ProductId}", productId )
 		params = { "DisplayProfile": displayProfile }
 		headers = Kobo.GetHeaderWithAccessToken()
@@ -322,6 +340,8 @@ def __GetDownloadInfo( productId: str, contentAccessBookResponse: dict ) -> Tupl
 		raise KoboException( message )
 
 	def __DownloadToFile( self, url, outputPath: str ) -> None:
+		Globals.Logger.debug( "Kobo.__DownloadToFile" )
+
 		response = self.Session.get( url, stream = True )
 		response.raise_for_status()
 		with open( outputPath, "wb" ) as f:
@@ -331,6 +351,8 @@ def __DownloadToFile( self, url, outputPath: str ) -> None:
 	# Downloading archived books is not possible, the "content_access_book" API endpoint returns with empty ContentKeys
 	# and ContentUrls for them.
 	def Download( self, productId: str, displayProfile: str, outputPath: str ) -> None:
+		Globals.Logger.debug( "Kobo.Download" )
+
 		jsonResponse = self.__GetContentAccessBook( productId, displayProfile )
 		contentKeys = Kobo.__GetContentKeys( jsonResponse )
 		downloadUrl, hasDrm = Kobo.__GetDownloadInfo( productId, jsonResponse )

kobo-book-downloader/LogFormatter.py

@@ -0,0 +1,16 @@
+import colorama
+
+import logging
+
+class LogFormatter( logging.Formatter ):
+	def __init__( self ):
+		self.DebugFormatter = logging.Formatter( colorama.Style.BRIGHT + colorama.Fore.GREEN + "%(levelname)s: " + colorama.Style.RESET_ALL + "%(message)s" )
+		self.ErrorFormatter = logging.Formatter( colorama.Style.BRIGHT + colorama.Fore.RED + "%(levelname)s: " + colorama.Style.RESET_ALL + "%(message)s" )
+
+	def format( self, record ):
+		if record.levelname == "DEBUG":
+			return self.DebugFormatter.format( record )
+		elif record.levelname == "ERROR":
+			return self.ErrorFormatter.format( record )
+		else:
+			return super().format( record )

kobo-book-downloader/__main__.py

@@ -1,11 +1,21 @@
 from Commands import Commands
 from Globals import Globals
 from Kobo import Kobo, KoboException
+from LogFormatter import LogFormatter
 from Settings import Settings
 
+import colorama
+import requests
+
 import argparse
+import logging
 
 def InitializeGlobals() -> None:
+	streamHandler = logging.StreamHandler()
+	streamHandler.setFormatter( LogFormatter() )
+	Globals.Logger = logging.getLogger()
+	Globals.Logger.addHandler( streamHandler )
+
 	Globals.Kobo = Kobo()
 	Globals.Settings = Settings()
 
@@ -44,8 +54,12 @@ def InitializeKoboApi() -> None:
 		Globals.Kobo.Login( email, password, captcha )
 
 def Main() -> None:
+	InitializeGlobals()
+	colorama.init()
+
 	argumentParser = argparse.ArgumentParser( add_help = False )
 	argumentParser.add_argument( "--help", "-h", default = False, action = "store_true" )
+	argumentParser.add_argument( "--verbose", default = False, action = "store_true", dest = "VerboseLogging" )
 	subparsers = argumentParser.add_subparsers( dest = "Command", title = "commands", metavar = "command" )
 	getParser = subparsers.add_parser( "get", help = "Download book" )
 	getParser.add_argument( "OutputPath", metavar = "output-path", help = "If the output path is a directory then the file will be named automatically." )
@@ -60,30 +74,29 @@ def Main() -> None:
 	wishListParser = subparsers.add_parser( "wishlist", help = "List your wish listed books" )
 	arguments = argumentParser.parse_args()
 
+	if arguments.VerboseLogging:
+		Globals.Logger.setLevel( logging.DEBUG )
+
 	if arguments.Command is None:
 		Commands.ShowUsage()
-		return
-
-	InitializeGlobals()
-
-	if arguments.Command == "info":
+	elif arguments.Command == "info":
 		Commands.Info()
-		return
-
-	InitializeKoboApi()
-
-	if arguments.Command == "get":
-		Commands.GetBookOrBooks( arguments.RevisionId, arguments.OutputPath, arguments.all )
-	elif arguments.Command == "list":
-		Commands.ListBooks( arguments.all )
-	elif arguments.Command == "pick":
-		Commands.PickBooks( arguments.OutputPath, arguments.all )
-	elif arguments.Command == "wishlist":
-		Commands.ListWishListedBooks()
-
+	else:
+		InitializeKoboApi()
+
+		if arguments.Command == "get":
+			Commands.GetBookOrBooks( arguments.RevisionId, arguments.OutputPath, arguments.all )
+		elif arguments.Command == "list":
+			Commands.ListBooks( arguments.all )
+		elif arguments.Command == "pick":
+			Commands.PickBooks( arguments.OutputPath, arguments.all )
+		elif arguments.Command == "wishlist":
+			Commands.ListWishListedBooks()
 
 if __name__ == '__main__':
 	try:
 		Main()
 	except KoboException as e:
-		print( "ERROR: %s" % e )
+		Globals.Logger.error( e )
+	except requests.exceptions.Timeout as e:
+		Globals.Logger.error( "The request has timed out." )