Coverage for odmpy/odm.py: 84.8%

1# Copyright (C) 2018 github.com/ping 

2# 

3# This file is part of odmpy. 

4# 

5# odmpy is free software: you can redistribute it and/or modify 

6# it under the terms of the GNU General Public License as published by 

7# the Free Software Foundation, either version 3 of the License, or 

8# (at your option) any later version. 

9# 

10# odmpy is distributed in the hope that it will be useful, 

11# but WITHOUT ANY WARRANTY; without even the implied warranty of 

12# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

13# GNU General Public License for more details. 

14# 

15# You should have received a copy of the GNU General Public License 

16# along with odmpy. If not, see <http://www.gnu.org/licenses/>. 

17# 

18 

19import argparse 

20import io 

21import json 

22import logging 

23import os 

24import sys 

25import time 

26from http.client import HTTPConnection 

27from pathlib import Path 

28from typing import Dict, List, Optional 

29 

30from termcolor import colored 

31 

32from .cli_utils import ( 

33 OdmpyCommands, 

34 OdmpyNoninteractiveOptions, 

35 positive_int, 

36 valid_book_folder_file_format, 

37 DEFAULT_FORMAT_FIELDS, 

38) 

39from .errors import LibbyNotConfiguredError, OdmpyRuntimeError 

40from .libby import LibbyClient, LibbyFormats 

41from .libby_errors import ClientBadRequestError, ClientError 

42from .overdrive import OverDriveClient 

43from .processing import ( 

44 process_odm, 

45 process_audiobook_loan, 

46 process_odm_return, 

47 process_ebook_loan, 

48) 

49from .processing.shared import ( 

50 generate_names, 

51 generate_cover, 

52 get_best_cover_url, 

53 init_session, 

54 extract_authors_from_openbook, 

55) 

56from .utils import slugify, plural_or_singular_noun as ps 

57 

58# 

59# Orchestrates the interaction between the CLI, APIs and the processing bits 

60# 

61 

62logger = logging.getLogger(__name__) 

63requests_logger = logging.getLogger("urllib3") 

64ch = logging.StreamHandler(io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8")) 

65ch.setLevel(logging.DEBUG) 

66logger.addHandler(ch) 

67logger.setLevel(logging.INFO) 

68requests_logger.addHandler(ch) 

69requests_logger.setLevel(logging.ERROR) 

70requests_logger.propagate = True 

71 

72__version__ = "0.8.1" # also update ../setup.py 

73TAGS_ENDPOINT = "https://api.github.com/repos/ping/odmpy/tags" 

74REPOSITORY_URL = "https://github.com/ping/odmpy" 

75OLD_SETTINGS_FOLDER_DEFAULT = Path("./odmpy_settings") 

76 

77 

78def check_version(timeout: int, max_retries: int) -> None: 

79 sess = init_session(max_retries) 

80 # noinspection PyBroadException 

81 try: 

82 res = sess.get(TAGS_ENDPOINT, timeout=timeout) 

83 res.raise_for_status() 

84 curr_version = res.json()[0].get("name", "") 

85 if curr_version and curr_version != __version__: 

86 logger.warning( 

87 f"⚠️ A new version {curr_version} is available at {REPOSITORY_URL}." 

88 ) 

89 except: # noqa: E722, pylint: disable=bare-except 

90 pass 

91 

92 

93def add_common_libby_arguments(parser_libby: argparse.ArgumentParser) -> None: 

94 parser_libby.add_argument( 

95 "--settings", 

96 dest="settings_folder", 

97 type=str, 

98 default="", 

99 metavar="SETTINGS_FOLDER", 

100 help="Settings folder to store odmpy required settings, e.g. Libby authentication.", 

101 ) 

102 parser_libby.add_argument( 

103 "--ebooks", 

104 dest="include_ebooks", 

105 default=False, 

106 action="store_true", 

107 help=( 

108 "Include ebook (EPUB/PDF) loans (experimental). An EPUB/PDF (DRM) loan will be downloaded as an .acsm file" 

109 "\nwhich can be opened in Adobe Digital Editions for offline reading." 

110 "\nRefer to https://help.overdrive.com/en-us/0577.html and " 

111 "\nhttps://help.overdrive.com/en-us/0005.html for more information." 

112 "\nAn open EPUB/PDF (no DRM) loan will be downloaded as an .epub/.pdf file which can be opened" 

113 "\nin any EPUB/PDF-compatible reader." 

114 if parser_libby.prog == f"odmpy {OdmpyCommands.Libby}" 

115 else "Include ebook (EPUB/PDF) loans." 

116 ), 

117 ) 

118 parser_libby.add_argument( 

119 "--magazines", 

120 dest="include_magazines", 

121 default=False, 

122 action="store_true", 

123 help=( 

124 "Include magazines loans (experimental)." 

125 if parser_libby.prog == f"odmpy {OdmpyCommands.Libby}" 

126 else "Include magazines loans." 

127 ), 

128 ) 

129 parser_libby.add_argument( 

130 "--noaudiobooks", 

131 dest="exclude_audiobooks", 

132 default=False, 

133 action="store_true", 

134 help="Exclude audiobooks.", 

135 ) 

136 

137 

138def add_common_download_arguments(parser_dl: argparse.ArgumentParser) -> None: 

139 """ 

140 Add common arguments needed for downloading 

141 

142 :param parser_dl: 

143 :return: 

144 """ 

145 parser_dl.add_argument( 

146 "-d", 

147 "--downloaddir", 

148 dest="download_dir", 

149 default=".", 

150 help="Download folder path.", 

151 ) 

152 parser_dl.add_argument( 

153 "-c", 

154 "--chapters", 

155 dest="add_chapters", 

156 action="store_true", 

157 help="Add chapter marks (experimental). For audiobooks.", 

158 ) 

159 parser_dl.add_argument( 

160 "-m", 

161 "--merge", 

162 dest="merge_output", 

163 action="store_true", 

164 help="Merge into 1 file (experimental, requires ffmpeg). For audiobooks.", 

165 ) 

166 parser_dl.add_argument( 

167 "--mergeformat", 

168 dest="merge_format", 

169 choices=["mp3", "m4b"], 

170 default="mp3", 

171 help="Merged file format (m4b is slow, experimental, requires ffmpeg). For audiobooks.", 

172 ) 

173 parser_dl.add_argument( 

174 "--mergecodec", 

175 dest="merge_codec", 

176 choices=["aac", "libfdk_aac"], 

177 default="aac", 

178 help="Audio codec of merged m4b file. (requires ffmpeg; using libfdk_aac requires ffmpeg compiled with libfdk_aac support). For audiobooks. Has no effect if mergeformat is not set to m4b.", 

179 ) 

180 parser_dl.add_argument( 

181 "-k", 

182 "--keepcover", 

183 dest="always_keep_cover", 

184 action="store_true", 

185 help="Always generate the cover image file (cover.jpg).", 

186 ) 

187 parser_dl.add_argument( 

188 "-f", 

189 "--keepmp3", 

190 dest="keep_mp3", 

191 action="store_true", 

192 help="Keep downloaded mp3 files (after merging). For audiobooks.", 

193 ) 

194 parser_dl.add_argument( 

195 "--nobookfolder", 

196 dest="no_book_folder", 

197 action="store_true", 

198 help="Don't create a book subfolder.", 

199 ) 

200 

201 available_fields_help = [ 

202 "%%(Title)s : Title", 

203 "%%(Author)s: Comma-separated Author names", 

204 "%%(Series)s: Series", 

205 ] 

206 if parser_dl.prog == "odmpy libby": 

207 available_fields_help.append("%%(ReadingOrder)s: Series Reading Order") 

208 available_fields_help.extend(["%%(Edition)s: Edition", "%%(ID)s: Title/Loan ID"]) 

209 available_fields_help_text = "Available fields:\n " + "\n ".join( 

210 available_fields_help 

211 ) 

212 

213 available_fields = list(DEFAULT_FORMAT_FIELDS) 

214 if parser_dl.prog != "odmpy libby": 

215 available_fields.remove("ReadingOrder") 

216 

217 parser_dl.add_argument( 

218 "--bookfolderformat", 

219 dest="book_folder_format", 

220 type=lambda v: valid_book_folder_file_format(v, tuple(available_fields)), 

221 default="%(Title)s - %(Author)s", 

222 help=f'Book folder format string. Default "%%(Title)s - %%(Author)s".\n{available_fields_help_text}', 

223 ) 

224 parser_dl.add_argument( 

225 "--bookfileformat", 

226 dest="book_file_format", 

227 type=lambda v: valid_book_folder_file_format(v, tuple(available_fields)), 

228 default="%(Title)s - %(Author)s", 

229 help=( 

230 'Book file format string (without extension). Default "%%(Title)s - %%(Author)s".\n' 

231 f"This applies to only merged audiobooks, ebooks, and magazines.\n{available_fields_help_text}" 

232 ), 

233 ) 

234 parser_dl.add_argument( 

235 "--removefrompaths", 

236 dest="remove_from_paths", 

237 metavar="ILLEGAL_CHARS", 

238 type=str, 

239 help=r'Remove characters in string specified from folder and file names, example "<>:"/\|?*"', 

240 ) 

241 parser_dl.add_argument( 

242 "--overwritetags", 

243 dest="overwrite_tags", 

244 action="store_true", 

245 help=( 

246 "Always overwrite ID3 tags.\n" 

247 "By default odmpy tries to non-destructively tag audiofiles.\n" 

248 "This option forces odmpy to overwrite tags where possible. For audiobooks." 

249 ), 

250 ) 

251 parser_dl.add_argument( 

252 "--tagsdelimiter", 

253 dest="tag_delimiter", 

254 metavar="DELIMITER", 

255 type=str, 

256 default=";", 

257 help=( 

258 "For ID3 tags with multiple values, this defines the delimiter.\n" 

259 'For example, with the default delimiter ";", authors are written\n' 

260 'to the artist tag as "Author A;Author B;Author C". For audiobooks.' 

261 ), 

262 ) 

263 parser_dl.add_argument( 

264 "--id3v2version", 

265 dest="id3v2_version", 

266 type=int, 

267 default=4, 

268 choices=[3, 4], 

269 help="ID3 v2 version. 3 = v2.3, 4 = v2.4", 

270 ) 

271 parser_dl.add_argument( 

272 "--opf", 

273 dest="generate_opf", 

274 action="store_true", 

275 help="Generate an OPF file for the downloaded audiobook/magazine/ebook.", 

276 ) 

277 parser_dl.add_argument( 

278 "-r", 

279 "--retry", 

280 dest="obsolete_retries", 

281 type=int, 

282 default=0, 

283 help="Obsolete. Do not use.", 

284 ) 

285 parser_dl.add_argument( 

286 "-j", 

287 "--writejson", 

288 dest="write_json", 

289 action="store_true", 

290 help="Generate a meta json file (for debugging).", 

291 ) 

292 parser_dl.add_argument( 

293 "--hideprogress", 

294 dest="hide_progress", 

295 action="store_true", 

296 help="Hide the download progress bar (e.g. during testing).", 

297 ) 

298 

299 

300def extract_bundled_contents( 

301 libby_client: LibbyClient, 

302 overdrive_client: OverDriveClient, 

303 selected_loan: Dict, 

304 cards: List[Dict], 

305 args: argparse.Namespace, 

306): 

307 format_id = libby_client.get_loan_format(selected_loan) 

308 format_info: Dict = next( 

309 iter([f for f in selected_loan.get("formats", []) if f["id"] == format_id]), 

310 {}, 

311 ) 

312 card: Dict = next( 

313 iter([c for c in cards if c["cardId"] == selected_loan["cardId"]]), {} 

314 ) 

315 if format_info.get("isBundleParent") and format_info.get("bundledContent", []): 

316 bundled_contents_ids = list( 

317 set([bc["titleId"] for bc in format_info["bundledContent"]]) 

318 ) 

319 for bundled_content_id in bundled_contents_ids: 

320 bundled_media = overdrive_client.library_media( 

321 card["advantageKey"], bundled_content_id 

322 ) 

323 if not libby_client.is_downloadable_ebook_loan(bundled_media): 

324 continue 

325 try: 

326 # patch in cardId from parent loan details 

327 bundled_media["cardId"] = selected_loan["cardId"] 

328 extract_loan_file(libby_client, bundled_media, args) 

329 except ClientError: 

330 # Oddly having a valid loan to the audiobook does not always mean 

331 # having access to the bundled content? Ref https://github.com/ping/odmpy/issues/54 

332 # Let this fail without affecting the main download since it's not critical 

333 logger.exception( 

334 'Unable to download bundled content for "%s"', 

335 bundled_media.get("title", ""), 

336 ) 

337 

338 

339def extract_loan_file( 

340 libby_client: LibbyClient, selected_loan: Dict, args: argparse.Namespace 

341) -> Optional[Path]: 

342 """ 

343 Extracts the ODM / ACSM / EPUB(open) file 

344 

345 :param libby_client: 

346 :param selected_loan: 

347 :param args: 

348 :return: The path to the ODM file 

349 """ 

350 try: 

351 format_id = LibbyClient.get_loan_format(selected_loan) 

352 except ValueError as err: 

353 err_msg = str(err) 

354 if "kindle" in str(err): 

355 logger.error( 

356 "You may have already sent the loan to your Kindle device: %s", 

357 colored(err_msg, "red"), 

358 ) 

359 else: 

360 logger.error(colored(err_msg, "red")) 

361 return None 

362 

363 file_ext = "odm" 

364 file_name = f'{selected_loan["title"]} {selected_loan["id"]}' 

365 loan_file_path = Path( 

366 args.download_dir, f"{slugify(file_name, allow_unicode=True)}.{file_ext}" 

367 ) 

368 if ( 

369 args.libby_direct 

370 and libby_client.has_format(selected_loan, LibbyFormats.EBookOverdrive) 

371 and not ( 

372 # don't do direct downloads for PDF loans because these turn out badly 

373 libby_client.has_format(selected_loan, LibbyFormats.EBookPDFAdobe) 

374 or libby_client.has_format(selected_loan, LibbyFormats.EBookPDFOpen) 

375 ) 

376 ): 

377 format_id = LibbyFormats.EBookOverdrive 

378 

379 openbook: Dict = {} 

380 rosters: List[Dict] = [] 

381 # pre-extract openbook first so that we can use it to create the book folder 

382 # with the creator names (needed to place the cover.jpg download) 

383 if format_id in (LibbyFormats.EBookOverdrive, LibbyFormats.MagazineOverDrive): 

384 _, openbook, rosters = libby_client.process_ebook(selected_loan) 

385 

386 cover_path = None 

387 if format_id in ( 

388 LibbyFormats.EBookEPubAdobe, 

389 LibbyFormats.EBookEPubOpen, 

390 LibbyFormats.EBookOverdrive, 

391 LibbyFormats.MagazineOverDrive, 

392 LibbyFormats.EBookPDFAdobe, 

393 LibbyFormats.EBookPDFOpen, 

394 ): 

395 file_ext = ( 

396 "acsm" 

397 if format_id in (LibbyFormats.EBookEPubAdobe, LibbyFormats.EBookPDFAdobe) 

398 else "pdf" 

399 if format_id == LibbyFormats.EBookPDFOpen 

400 else "epub" 

401 ) 

402 book_folder, book_file_name = generate_names( 

403 title=selected_loan["title"], 

404 series=selected_loan.get("series") or "", 

405 series_reading_order=selected_loan.get("detailedSeries", {}).get( 

406 "readingOrder", "" 

407 ), 

408 authors=extract_authors_from_openbook(openbook) 

409 or ( 

410 [selected_loan["firstCreatorName"]] 

411 if selected_loan.get("firstCreatorName") 

412 else [] 

413 ), # for open-epub 

414 edition=selected_loan.get("edition") or "", 

415 title_id=selected_loan["id"], 

416 args=args, 

417 logger=logger, 

418 ) 

419 loan_file_path = book_file_name.with_suffix(f".{file_ext}") 

420 if ( 

421 format_id 

422 in ( 

423 LibbyFormats.EBookOverdrive, 

424 LibbyFormats.MagazineOverDrive, 

425 ) 

426 and not loan_file_path.exists() 

427 ): 

428 # we need the cover for embedding 

429 cover_path, _ = generate_cover( 

430 book_folder=book_folder, 

431 cover_url=get_best_cover_url(selected_loan), 

432 session=init_session(args.retries), 

433 timeout=args.timeout, 

434 logger=logger, 

435 force_square=False, 

436 ) 

437 

438 # don't re-download odm if it already exists so that we don't 

439 # needlessly use up the fulfillment limits 

440 if not loan_file_path.exists(): 

441 if format_id in (LibbyFormats.EBookOverdrive, LibbyFormats.MagazineOverDrive): 

442 process_ebook_loan( 

443 loan=selected_loan, 

444 cover_path=cover_path, 

445 openbook=openbook, 

446 rosters=rosters, 

447 libby_client=libby_client, 

448 args=args, 

449 logger=logger, 

450 ) 

451 else: 

452 # formats: odm, acsm, open-epub, open-pdf 

453 try: 

454 odm_res_content = libby_client.fulfill_loan_file( 

455 selected_loan["id"], selected_loan["cardId"], format_id 

456 ) 

457 with loan_file_path.open("wb") as f: 

458 f.write(odm_res_content) 

459 logger.info( 

460 'Downloaded %s to "%s"', 

461 file_ext, 

462 colored(str(loan_file_path), "magenta"), 

463 ) 

464 except ClientError as ce: 

465 if ce.http_status == 400 and libby_client.is_downloadable_ebook_loan( 

466 selected_loan 

467 ): 

468 logger.error( 

469 "%s %s", 

470 colored( 

471 f"Unable to download {file_ext}.", "red", attrs=["bold"] 

472 ), 

473 colored("You may have sent the loan to a Kindle.", "red"), 

474 ) 

475 return None 

476 raise 

477 else: 

478 logger.info( 

479 "Already downloaded %s file %s", 

480 file_ext, 

481 colored(str(loan_file_path), "magenta"), 

482 ) 

483 

484 if cover_path and cover_path.exists() and not args.always_keep_cover: 

485 # clean up 

486 cover_path.unlink() 

487 

488 return loan_file_path 

489 

490 

491def run(custom_args: Optional[List[str]] = None, be_quiet: bool = False) -> None: 

492 """ 

493 

494 :param custom_args: Used by unittests 

495 :param be_quiet: Used by unittests 

496 :return: 

497 """ 

498 parser = argparse.ArgumentParser( 

499 prog="odmpy", 

500 description="Manage your OverDrive loans", 

501 epilog=( 

502 f"Version {__version__}. " 

503 f"[Python {sys.version_info.major}.{sys.version_info.minor}.{sys.version_info.micro}-{sys.platform}] " 

504 f"Source at {REPOSITORY_URL}" 

505 ), 

506 fromfile_prefix_chars="@", 

507 ) 

508 parser.add_argument( 

509 "--version", 

510 action="version", 

511 version=( 

512 f"%(prog)s {__version__} " 

513 f"[Python {sys.version_info.major}.{sys.version_info.minor}.{sys.version_info.micro}-{sys.platform}]" 

514 ), 

515 ) 

516 parser.add_argument( 

517 "-v", 

518 "--verbose", 

519 dest="verbose", 

520 action="store_true", 

521 help="Enable more verbose messages for debugging.", 

522 ) 

523 parser.add_argument( 

524 "-t", 

525 "--timeout", 

526 dest="timeout", 

527 type=int, 

528 default=10, 

529 help="Timeout (seconds) for network requests. Default 10.", 

530 ) 

531 parser.add_argument( 

532 "-r", 

533 "--retry", 

534 dest="retries", 

535 type=int, 

536 default=1, 

537 help="Number of retries if a network request fails. Default 1.", 

538 ) 

539 parser.add_argument( 

540 "--noversioncheck", 

541 dest="dont_check_version", 

542 default=False, 

543 action="store_true", 

544 help="Do not check if newer version is available.", 

545 ) 

546 

547 subparsers = parser.add_subparsers( 

548 title="Available commands", 

549 dest="command_name", 

550 help="To get more help, use the -h option with the command.", 

551 ) 

552 

553 # libby download parser 

554 parser_libby = subparsers.add_parser( 

555 OdmpyCommands.Libby, 

556 description="Interactive Libby Interface for downloading loans.", 

557 help="Download audiobook/ebook/magazine loans via Libby.", 

558 formatter_class=argparse.RawTextHelpFormatter, 

559 ) 

560 add_common_libby_arguments(parser_libby) 

561 add_common_download_arguments(parser_libby) 

562 parser_libby.add_argument( 

563 "--direct", 

564 dest="libby_direct", 

565 action="store_true", 

566 help=( 

567 "Process the download directly from Libby without " 

568 "\ndownloading an odm/acsm file. For audiobooks/eBooks." 

569 ), 

570 ) 

571 parser_libby.add_argument( 

572 "--keepodm", 

573 action="store_true", 

574 help="Keep the downloaded odm and license files. For audiobooks.", 

575 ) 

576 parser_libby.add_argument( 

577 "--latest", 

578 dest=OdmpyNoninteractiveOptions.DownloadLatestN, 

579 type=positive_int, 

580 default=0, 

581 metavar="N", 

582 help="Non-interactive mode that downloads the latest N number of loans.", 

583 ) 

584 parser_libby.add_argument( 

585 "--select", 

586 dest=OdmpyNoninteractiveOptions.DownloadSelectedN, 

587 type=positive_int, 

588 nargs="+", 

589 metavar="N", 

590 help=( 

591 "Non-interactive mode that downloads loans by the index entered.\n" 

592 'For example, "--select 1 5" will download the first and fifth loans in ' 

593 "order of the checked out date.\n" 

594 "If the 5th loan does not exist, it will be skipped." 

595 ), 

596 ) 

597 parser_libby.add_argument( 

598 "--selectid", 

599 dest=OdmpyNoninteractiveOptions.DownloadSelectedId, 

600 type=positive_int, 

601 nargs="+", 

602 metavar="ID", 

603 help=( 

604 "Non-interactive mode that downloads loans by the loan ID entered.\n" 

605 'For example, "--selectid 12345" will download the loan with the ID 12345.\n' 

606 "If the loan with the ID does not exist, it will be skipped." 

607 ), 

608 ) 

609 parser_libby.add_argument( 

610 "--exportloans", 

611 dest=OdmpyNoninteractiveOptions.ExportLoans, 

612 metavar="LOANS_JSON_FILEPATH", 

613 type=str, 

614 help="Non-interactive mode that exports loan information into a json file at the path specified.", 

615 ) 

616 parser_libby.add_argument( 

617 "--reset", 

618 dest="reset_settings", 

619 action="store_true", 

620 help="Remove previously saved odmpy Libby settings.", 

621 ) 

622 parser_libby.add_argument( 

623 "--check", 

624 dest=OdmpyNoninteractiveOptions.Check, 

625 action="store_true", 

626 help="Non-interactive mode that displays Libby signed-in status and token if authenticated.", 

627 ) 

628 parser_libby.add_argument( 

629 "--debug", 

630 dest="is_debug_mode", 

631 action="store_true", 

632 help="Debug switch for use during development. Please do not use.", 

633 ) 

634 

635 # libby return parser 

636 parser_libby_return = subparsers.add_parser( 

637 OdmpyCommands.LibbyReturn, 

638 description="Interactive Libby Interface for returning loans.", 

639 help="Return loans via Libby.", 

640 formatter_class=argparse.RawTextHelpFormatter, 

641 ) 

642 add_common_libby_arguments(parser_libby_return) 

643 

644 # libby renew parser 

645 parser_libby_renew = subparsers.add_parser( 

646 OdmpyCommands.LibbyRenew, 

647 description="Interactive Libby Interface for renewing loans.", 

648 help="Renew loans via Libby.", 

649 formatter_class=argparse.RawTextHelpFormatter, 

650 ) 

651 add_common_libby_arguments(parser_libby_renew) 

652 

653 # odm download parser 

654 parser_dl = subparsers.add_parser( 

655 OdmpyCommands.Download, 

656 description="Download from an audiobook loan file (odm).", 

657 help="Download from an audiobook loan file (odm).", 

658 formatter_class=argparse.RawTextHelpFormatter, 

659 ) 

660 parser_dl.add_argument("odm_file", type=str, help="ODM file path.") 

661 add_common_download_arguments(parser_dl) 

662 

663 # odm return parser 

664 parser_ret = subparsers.add_parser( 

665 OdmpyCommands.Return, 

666 description="Return an audiobook loan file (odm).", 

667 help="Return an audiobook loan file (odm).", 

668 ) 

669 parser_ret.add_argument("odm_file", type=str, help="ODM file path.") 

670 

671 # odm info parser 

672 parser_info = subparsers.add_parser( 

673 OdmpyCommands.Information, 

674 description="Get information about an audiobook loan file (odm).", 

675 help="Get information about an audiobook loan file (odm).", 

676 ) 

677 parser_info.add_argument( 

678 "-f", 

679 "--format", 

680 dest="format", 

681 choices=["text", "json"], 

682 default="text", 

683 help="Format for output.", 

684 ) 

685 parser_info.add_argument("odm_file", type=str, help="ODM file path.") 

686 

687 args = parser.parse_args(custom_args) 

688 

689 if be_quiet: 

690 # in test mode 

691 ch.setLevel(logging.ERROR) 

692 elif args.verbose: 

693 logger.setLevel(logging.DEBUG) 

694 requests_logger.setLevel(logging.DEBUG) 

695 HTTPConnection.debuglevel = 1 

696 

697 if hasattr(args, "download_dir") and args.download_dir: 

698 download_dir = Path(args.download_dir) 

699 if not download_dir.exists(): 

700 # prevents FileNotFoundError when using libby odm-based downloads 

701 # because the odm is first downloaded into the download dir 

702 # without a book folder 

703 download_dir.mkdir(parents=True, exist_ok=True) 

704 args.download_dir = str(download_dir.expanduser()) 

705 

706 if hasattr(args, "settings_folder"): 

707 default_config_folder = ( 

708 Path( 

709 os.environ.get("APPDATA") 

710 or os.environ.get("XDG_CONFIG_HOME") 

711 or Path(os.environ.get("HOME", "./")).joinpath(".config") 

712 ) 

713 .joinpath("odmpy") 

714 .expanduser() 

715 ) 

716 if args.settings_folder: 

717 args.settings_folder = str(Path(args.settings_folder).expanduser()) 

718 elif OLD_SETTINGS_FOLDER_DEFAULT.joinpath("libby.json").exists(): 

719 # handle backward-compat for versions <= 0.8.1 

720 args.settings_folder = str(OLD_SETTINGS_FOLDER_DEFAULT) 

721 else: 

722 args.settings_folder = str(default_config_folder) 

723 

724 if hasattr(args, "export_loans_path") and args.export_loans_path: 

725 args.export_loans_path = str(Path(args.export_loans_path).expanduser()) 

726 

727 # suppress warnings 

728 logging.getLogger("eyed3").setLevel( 

729 logging.WARNING if logger.level == logging.DEBUG else logging.ERROR 

730 ) 

731 

732 if not args.dont_check_version: 

733 check_version(args.timeout, args.retries) 

734 

735 if hasattr(args, "obsolete_retries") and args.obsolete_retries: 

736 # retire --retry on the subcommands, after v0.6.7 

737 logger.warning( 

738 f"{'*' * 60}\n⚠️ The %s option for the %s command is no longer valid, and\n" 

739 f"has been moved to the base odmpy command, example: 'odmpy --retry {args.obsolete_retries}'.\n" 

740 f"Please change your command to '%s' instead\n{'*' * 60}", 

741 colored("--retry/-r", "red"), 

742 colored(args.command_name, "red"), 

743 colored( 

744 f"odmpy --retry {args.obsolete_retries} {args.command_name}", 

745 attrs=["bold"], 

746 ), 

747 ) 

748 time.sleep(3) 

749 

750 try: 

751 # Libby-based commands 

752 if args.command_name in ( 

753 OdmpyCommands.Libby, 

754 OdmpyCommands.LibbyReturn, 

755 OdmpyCommands.LibbyRenew, 

756 ): 

757 logger.info( 

758 "%s Interactive Client for Libby", colored("odmpy", attrs=["bold"]) 

759 ) 

760 logger.info("-" * 70) 

761 

762 token = os.environ.get("LIBBY_TOKEN") 

763 if token: 

764 # use token auth if available 

765 libby_client = LibbyClient( 

766 identity_token=token, 

767 max_retries=args.retries, 

768 timeout=args.timeout, 

769 logger=logger, 

770 ) 

771 else: 

772 libby_client = LibbyClient( 

773 settings_folder=args.settings_folder, 

774 max_retries=args.retries, 

775 timeout=args.timeout, 

776 logger=logger, 

777 ) 

778 

779 overdrive_client = OverDriveClient( 

780 user_agent=libby_client.user_agent, 

781 timeout=args.timeout, 

782 retry=args.retries, 

783 ) 

784 

785 if args.command_name == OdmpyCommands.Libby and args.reset_settings: 

786 libby_client.clear_settings() 

787 logger.info("Cleared settings.") 

788 return 

789 

790 if args.command_name == OdmpyCommands.Libby and args.check_signed_in: 

791 if not libby_client.get_token(): 

792 raise LibbyNotConfiguredError("Libby has not been setup.") 

793 if not libby_client.is_logged_in(): 

794 raise LibbyNotConfiguredError("Libby is not signed-in.") 

795 logger.info( 

796 "Libby is signed-in with token:\n%s", libby_client.get_token() 

797 ) 

798 return 

799 

800 # detect if non-interactive command options are selected before setup 

801 if not libby_client.get_token(): 

802 if [ 

803 opt_name 

804 for opt_name in OdmpyNoninteractiveOptions 

805 if hasattr(args, opt_name) and getattr(args, opt_name) 

806 ]: 

807 raise OdmpyRuntimeError( 

808 'Libby has not been setup. Please run "odmpy libby" first.' 

809 ) 

810 

811 if not libby_client.get_token(): 

812 instructions = ( 

813 "A Libby setup code is needed to allow odmpy to interact with Libby.\n" 

814 "To get a Libby code, see https://help.libbyapp.com/en-us/6070.htm\n" 

815 ) 

816 logger.info(instructions) 

817 while True: 

818 sync_code = input( 

819 "Enter the 8-digit Libby code and press enter: " 

820 ).strip() 

821 if not sync_code: 

822 return 

823 if not LibbyClient.is_valid_sync_code(sync_code): 

824 logger.warning("Invalid code: %s", colored(sync_code, "red")) 

825 continue 

826 break 

827 

828 try: 

829 libby_client.get_chip() 

830 libby_client.clone_by_code(sync_code) 

831 if not libby_client.is_logged_in(): 

832 libby_client.clear_settings() 

833 raise OdmpyRuntimeError( 

834 "Could not log in with code.\n" 

835 "Make sure that you have entered the right code and within the time limit.\n" 

836 "You also need to have at least 1 registered library card." 

837 ) 

838 logger.info("Login successful.\n") 

839 except ClientError as ce: 

840 libby_client.clear_settings() 

841 raise OdmpyRuntimeError( 

842 "Could not log in with code.\n" 

843 "Make sure that you have entered the right code and within the time limit." 

844 ) from ce 

845 

846 synced_state = libby_client.sync() 

847 cards = synced_state.get("cards", []) 

848 # sort by checkout date so that recent most is at the bottom 

849 libby_loans = sorted( 

850 [ 

851 book 

852 for book in synced_state.get("loans", []) 

853 if ( 

854 (not args.exclude_audiobooks) 

855 and libby_client.is_downloadable_audiobook_loan(book) 

856 ) 

857 or ( 

858 args.include_ebooks 

859 and libby_client.is_downloadable_ebook_loan(book) 

860 ) 

861 or ( 

862 args.include_magazines 

863 and libby_client.is_downloadable_magazine_loan(book) 

864 ) 

865 ], 

866 key=lambda ln: ln["checkoutDate"], # type: ignore[no-any-return] 

867 ) 

868 

869 if args.command_name == OdmpyCommands.Libby and args.export_loans_path: 

870 logger.info( 

871 "Non-interactive mode. Exporting loans json to %s...", 

872 colored(args.export_loans_path, "magenta"), 

873 ) 

874 with open(args.export_loans_path, "w", encoding="utf-8") as f: 

875 json.dump(libby_loans, f) 

876 logger.info( 

877 'Saved loans as "%s"', 

878 colored(args.export_loans_path, "magenta", attrs=["bold"]), 

879 ) 

880 return 

881 

882 if args.command_name == OdmpyCommands.LibbyRenew: 

883 libby_loans = [ 

884 loan for loan in libby_loans if libby_client.is_renewable(loan) 

885 ] 

886 if not libby_loans: 

887 logger.info("No renewable loans found.") 

888 return 

889 

890 if not libby_loans: 

891 logger.info("No downloadable loans found.") 

892 return 

893 

894 if args.command_name == OdmpyCommands.Libby and ( 

895 args.selected_loans_indices 

896 or args.download_latest_n 

897 or args.selected_loans_ids 

898 ): 

899 # Non-interactive selection 

900 selected_loans_indices = [] 

901 total_loans_count = len(libby_loans) 

902 if args.selected_loans_indices: 

903 selected_loans_indices.extend( 

904 [ 

905 j 

906 for j in args.selected_loans_indices 

907 if j <= total_loans_count 

908 ] 

909 ) 

910 logger.info( 

911 "Non-interactive mode. Downloading selected %s %s...", 

912 ps(len(selected_loans_indices), "loan"),