Hide keyboard shortcuts

Hot-keys on this page

r m x p   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

1"""LICENSE 

2Copyright 2019 Hermann Krumrey <hermann@krumreyh.com> 

3 

4This file is part of otaku-info-bot. 

5 

6otaku-info-bot is free software: you can redistribute it and/or modify 

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

8the Free Software Foundation, either version 3 of the License, or 

9(at your option) any later version. 

10 

11otaku-info-bot is distributed in the hope that it will be useful, 

12but WITHOUT ANY WARRANTY; without even the implied warranty of 

13MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

14GNU General Public License for more details. 

15 

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

17along with otaku-info-bot. If not, see <http://www.gnu.org/licenses/>. 

18LICENSE""" 

19 

20from typing import Dict, List, Any, Optional, Type 

21from datetime import datetime 

22from kudubot.Bot import Bot 

23from kudubot.db.Address import Address 

24from kudubot.parsing.CommandParser import CommandParser 

25from sqlalchemy.orm import Session 

26from otaku_info_bot import version 

27from otaku_info_bot.OtakuInfoCommandParser import OtakuInfoCommandParser 

28from otaku_info_bot.fetching.anime import load_newest_episodes 

29from otaku_info_bot.fetching.ln import load_ln_releases 

30from otaku_info_bot.db.AnilistEntry import AnilistEntry 

31from otaku_info_bot.db.Notification import Notification 

32from otaku_info_bot.db.config import NotificationConfig, \ 

33 MangaNotificationConfig, AnimeNotificationConfig 

34from otaku_info_bot.fetching.anilist import load_anilist 

35from otaku_info_bot.db.MangaChapterGuess import MangaChapterGuess 

36from otaku_info_bot.db.RankedModeSetting import RankedModeSetting 

37 

38 

39class OtakuInfoBot(Bot): 

40 """ 

41 The OtakuInfo Bot class that defines the anime reminder 

42 functionality. 

43 """ 

44 

45 cache_delay_count = 0 

46 """ 

47 Used to stagger out cache stores of manga chapter guesses 

48 so that they don't all occur at the same time 

49 """ 

50 

51 @classmethod 

52 def name(cls) -> str: 

53 """ 

54 :return: The name of the bot 

55 """ 

56 return "otaku-info-bot" 

57 

58 @classmethod 

59 def version(cls) -> str: 

60 """ 

61 :return: The current version of the bot 

62 """ 

63 return version 

64 

65 @classmethod 

66 def parsers(cls) -> List[CommandParser]: 

67 """ 

68 :return: A list of parser the bot supports for commands 

69 """ 

70 return [OtakuInfoCommandParser()] 

71 

72 def activate_config( 

73 self, 

74 address: Address, 

75 args: Dict[str, Any], 

76 db_session: Session, 

77 config_cls: Any # Type[NotificationConfig] 

78 ): 

79 """ 

80 Activates a configuration for a user 

81 :param address: The user's address 

82 :param args: The command arguments 

83 :param db_session: The database session 

84 :param config_cls: The configuration class 

85 :return: None 

86 """ 

87 exists = db_session.query(config_cls)\ 

88 .filter_by(address=address).first() is not None 

89 

90 if exists: 

91 msg = "Configuration already activated" 

92 self.send_txt(address, msg, "Already Active") 

93 

94 else: 

95 username = args["anilist-username"] 

96 default_list_name = config_cls.default_list_name() 

97 # noinspection PyArgumentList 

98 config = config_cls( 

99 anilist_username=username, 

100 address=address, 

101 list_name=args.get("custom-list", default_list_name) 

102 ) 

103 db_session.add(config) 

104 db_session.commit() 

105 

106 self.send_txt(address, "Configuration Activated", "Activated") 

107 

108 def deactivate_config( 

109 self, 

110 address: Address, 

111 db_session: Session, 

112 config_cls: Type[NotificationConfig] 

113 ): 

114 """ 

115 Deactivates an anilist configuration for a user 

116 :param address: The address for which to deactivate the config 

117 :param db_session: The database session to use 

118 :param config_cls: The configuration class to use 

119 :return: None 

120 """ 

121 existing = db_session.query(config_cls) \ 

122 .filter_by(address=address).first() 

123 

124 if existing is not None: 

125 db_session.delete(existing) 

126 

127 for notification in db_session.query(Notification)\ 

128 .filter_by(address=address).all(): 

129 if notification.entry.media_type == config_cls.media_type(): 

130 db_session.delete(notification) 

131 

132 db_session.commit() 

133 

134 self.send_txt(address, "Deactivated Configuration", "Deactivated") 

135 

136 def _update_anilist_entries(self, db_session: Session): 

137 """ 

138 Updates anilist entries in the database 

139 :param db_session: The database session to use 

140 :return: None 

141 """ 

142 self.logger.info("Updating anilist entries") 

143 

144 newest_anime_episodes = load_newest_episodes() 

145 manga_progress = {} # type: Dict[int, int] 

146 

147 for config_cls in [AnimeNotificationConfig, MangaNotificationConfig]: 

148 

149 media_type = config_cls.media_type() 

150 

151 for config in db_session.query(config_cls).all(): 

152 anilist = load_anilist( 

153 config.anilist_username, 

154 media_type, 

155 config.list_name 

156 ) 

157 

158 anilist_ids = [] 

159 for entry in anilist: 

160 

161 anilist_id = entry["media"]["id"] 

162 anilist_ids.append(anilist_id) 

163 

164 releasing = entry["media"]["status"] == "RELEASING" 

165 completed = entry["media"]["status"] == "FINISHED" 

166 

167 romaji_name = entry["media"]["title"]["romaji"] 

168 english_name = entry["media"]["title"]["english"] 

169 name = romaji_name 

170 if media_type == "manga" and english_name is not None: 

171 name = english_name 

172 

173 user_progress = entry["progress"] 

174 user_score = entry["score"] 

175 

176 db_entry = db_session.query(AnilistEntry).filter_by( 

177 anilist_id=anilist_id, media_type=media_type 

178 ).first() 

179 

180 # Calculate newest 

181 if media_type == "anime": 

182 latest = newest_anime_episodes.get(anilist_id) 

183 if db_entry is None: 

184 

185 if latest is None: 

186 next_ep = entry["media"]["nextAiringEpisode"] 

187 if next_ep is not None: 

188 latest = next_ep["episode"] - 1 

189 if latest is None: 

190 latest = entry["media"]["episodes"] 

191 elif latest is None: 

192 latest = db_entry.latest 

193 

194 else: # media_type == "manga": 

195 latest = entry["media"]["chapters"] 

196 if latest is None: 

197 latest = manga_progress.get(anilist_id) 

198 if latest is None: 

199 latest = self.get_cached_manga_chapter_guess( 

200 anilist_id, db_session 

201 ) 

202 manga_progress[anilist_id] = latest 

203 

204 db_entry = db_session.query(AnilistEntry).filter_by( 

205 anilist_id=anilist_id, media_type=media_type 

206 ).first() 

207 if db_entry is None: 

208 db_entry = AnilistEntry( 

209 anilist_id=anilist_id, 

210 name=name, 

211 latest=latest, 

212 media_type=media_type, 

213 releasing=releasing, 

214 completed=completed 

215 ) 

216 db_session.add(db_entry) 

217 elif latest != 0: 

218 db_entry.latest = latest 

219 db_entry.releasing = releasing 

220 db_entry.completed = completed 

221 

222 notification = db_session.query(Notification).filter_by( 

223 entry=db_entry, address=config.address 

224 ).first() 

225 

226 if notification is None: 

227 notification = Notification( 

228 address=config.address, 

229 entry=db_entry, 

230 user_progress=user_progress, 

231 last_update=user_progress, 

232 user_score=user_score 

233 ) 

234 db_session.add(notification) 

235 else: 

236 notification.user_progress = user_progress 

237 notification.user_score = user_score 

238 

239 # Purge stale entries 

240 for existing in db_session.query(Notification) \ 

241 .filter_by(address=config.address).all(): 

242 if existing.entry.media_type != media_type: 

243 continue 

244 elif existing.entry.anilist_id not in anilist_ids: 

245 db_session.delete(existing) 

246 

247 db_session.commit() 

248 self.logger.info("Finished updating anilist entries") 

249 

250 def _send_notifications( 

251 self, 

252 db_session: Session, 

253 address_limit: Optional[Address] = None, 

254 use_user_progress: bool = False, 

255 media_type_limit: Optional[str] = None, 

256 send_completed: bool = True, 

257 send_releasing: bool = True, 

258 mincount: int = 1 

259 ): 

260 """ 

261 Sends out any due notifications 

262 :param db_session: The database session to use 

263 :param address_limit: Can be set to limit this to a single user 

264 :param use_user_progress: Instead of looking at the last time a 

265 notification was sent out, look at the user's 

266 progress 

267 :param media_type_limit: Limits the media type of the notifications to 

268 be sent 

269 :param send_completed: Whether to send notifications fore completed 

270 series 

271 :param send_releasing: Whether to send notifications for currently 

272 releasing series 

273 :param mincount: The minimum amount of chapters/episodes the user needs 

274 to be behind for the notification message to trigger 

275 :return: None 

276 """ 

277 self.logger.info("Sending Notifications") 

278 

279 due = {} # type: Dict[int, Dict[str, Any]] 

280 

281 for notification in db_session.query(Notification).all(): 

282 

283 if notification.entry.releasing and not send_releasing: 

284 continue 

285 elif notification.entry.completed and not send_completed: 

286 continue 

287 

288 address_id = notification.address_id 

289 

290 if address_limit is not None and address_limit.id != address_id: 

291 continue 

292 

293 if address_id not in due: 

294 due[address_id] = { 

295 "address": notification.address, 

296 "anime": [], 

297 "manga": [] 

298 } 

299 

300 media_type = notification.entry.media_type 

301 if notification.diff > 0: 

302 self.logger.debug("Notification updated: " + str(notification)) 

303 due[address_id][media_type].append(notification) 

304 elif use_user_progress and notification.user_diff >= mincount: 

305 due[address_id][media_type].append(notification) 

306 

307 for _, data in due.items(): 

308 address = data["address"] 

309 

310 ranked_setting = db_session.query(RankedModeSetting) \ 

311 .filter_by(address=address).first() 

312 if ranked_setting is None: 

313 use_ranked = False 

314 else: 

315 use_ranked = ranked_setting.value 

316 

317 for media_type in ["manga", "anime"]: 

318 

319 if media_type_limit is not None \ 

320 and media_type != media_type_limit: 

321 continue 

322 

323 notifications = data[media_type] 

324 notifications.sort(key=lambda x: x.entry.name) 

325 notifications.sort(key=lambda x: x.user_diff, reverse=True) 

326 

327 if use_ranked: 

328 notifications.sort( 

329 key=lambda x: x.user_score, reverse=True 

330 ) 

331 

332 media_name = media_type[0].upper() + media_type[1:] 

333 unit_type = "Episode" if media_type == "anime" else "Chapter" 

334 

335 if len(notifications) <= 5: 

336 for notification in notifications: 

337 message = "{} {} {} was released\n\n" 

338 message += "Current Progress: {}/{} (+{})\n\n{}" 

339 message = message.format( 

340 notification.entry.name, 

341 unit_type, 

342 notification.entry.latest, 

343 notification.user_progress, 

344 notification.entry.latest, 

345 notification.user_diff, 

346 notification.entry.anilist_url 

347 ) 

348 self.send_txt(address, message, "Notification") 

349 else: 

350 message = "New {} {}s:\n\n".format(media_name, unit_type) 

351 for notification in notifications: 

352 message += "\\[+{}] {} {} {}/{}\n".format( 

353 notification.user_diff, 

354 notification.entry.name, 

355 unit_type, 

356 notification.user_progress, 

357 notification.entry.latest 

358 ) 

359 self.send_txt(address, message, "Notifications") 

360 

361 for notification in notifications: 

362 notification.last_update = notification.entry.latest 

363 

364 db_session.commit() 

365 self.logger.info("Finished Sending Notifications") 

366 

367 def get_cached_manga_chapter_guess( 

368 self, 

369 anilist_id: int, 

370 db_session: Session 

371 ) -> int: 

372 """ 

373 Uses manga chapter guesses from the database to reduce requests to 

374 anilist servers and generally reduce loading times 

375 :param anilist_id: The anilist ID for which to get the chapter guess 

376 :param db_session: The database session to use 

377 :return: The manga chapter guess 

378 """ 

379 cached = db_session.query(MangaChapterGuess)\ 

380 .filter_by(id=anilist_id).first() 

381 

382 if cached is None: 

383 

384 self.logger.debug("Creating new manga chapter guess cache for {}" 

385 .format(anilist_id)) 

386 

387 # This delay makes sure that chapter guesses aren't all updated 

388 # at the same time 

389 delay = self.cache_delay_count * 60 

390 if self.cache_delay_count % 60 == 0: 

391 self.cache_delay_count = 0 

392 self.cache_delay_count += 1 

393 

394 cached = MangaChapterGuess(id=anilist_id, last_check=0) 

395 db_session.add(cached) 

396 

397 cached.update() 

398 cached.last_check += delay 

399 

400 was_updated = cached.update() 

401 

402 if was_updated: 

403 self.logger.debug("Cached chapter guess value for {} updated" 

404 .format(anilist_id)) 

405 else: 

406 self.logger.debug("Using cached chapter guess for {}" 

407 .format(anilist_id)) 

408 

409 db_session.commit() 

410 

411 return cached.guess 

412 

413 def on_activate_anime_notifications( 

414 self, 

415 address: Address, 

416 args: Dict[str, Any], 

417 db_session: Session 

418 ): 

419 """ 

420 Activates anime notifications for a user 

421 :param address: The user's address 

422 :param args: The arguments, containing the anilist username 

423 :param db_session: The database session to use 

424 :return: None 

425 """ 

426 self.activate_config( 

427 address, args, db_session, AnimeNotificationConfig 

428 ) 

429 

430 def on_deactivate_anime_notifications( 

431 self, 

432 address: Address, 

433 _, 

434 db_session: Session 

435 ): 

436 """ 

437 Deactivates anime notifications for a user 

438 :param address: The user's address 

439 :param _: The arguments 

440 :param db_session: The database session to use 

441 :return: None 

442 """ 

443 self.deactivate_config(address, db_session, AnimeNotificationConfig) 

444 

445 def on_list_new_anime_episodes( 

446 self, 

447 address: Address, 

448 _, 

449 db_session: Session 

450 ): 

451 """ 

452 Handles listing new episodes for a user 

453 :param address: The address that requested this 

454 :param _: The arguments 

455 :param db_session: The database session to use 

456 :return: None 

457 """ 

458 self._send_notifications(db_session, address, True, "anime") 

459 

460 def on_list_new_releasing_episodes( 

461 self, 

462 address: Address, 

463 args: Dict[str, Any], 

464 db_session: Session 

465 ): 

466 """ 

467 Handles listing new episodes for a user 

468 Only sends notifications for currently releasing anime series 

469 :param address: The address that requested this 

470 :param args: The arguments provided by the user 

471 :param db_session: The database session to use 

472 :return: None 

473 """ 

474 self._send_notifications( 

475 db_session, 

476 address, 

477 True, 

478 "anime", 

479 send_completed=False, 

480 mincount=args.get("mincount", 1) 

481 ) 

482 

483 def on_list_new_completed_episodes( 

484 self, 

485 address: Address, 

486 _, 

487 db_session: Session 

488 ): 

489 """ 

490 Handles listing new episodes for a user 

491 Only sends notifications for completed anime series 

492 :param address: The address that requested this 

493 :param _: The arguments 

494 :param db_session: The database session to use 

495 :return: None 

496 """ 

497 self._send_notifications( 

498 db_session, address, True, "anime", send_releasing=False 

499 ) 

500 

501 def on_activate_manga_notifications( 

502 self, 

503 address: Address, 

504 args: Dict[str, Any], 

505 db_session: Session 

506 ): 

507 """ 

508 Handles activating manga notifications for a user using anilist 

509 :param address: The user that sent this request 

510 :param args: The arguments to use 

511 :param db_session: The database session to use 

512 :return: None 

513 """ 

514 self.activate_config( 

515 address, args, db_session, MangaNotificationConfig 

516 ) 

517 

518 def on_deactivate_manga_notifications( 

519 self, 

520 address: Address, 

521 _, 

522 db_session: Session 

523 ): 

524 """ 

525 Handles deactivating manga notifications for a user using anilist 

526 :param address: The user that sent this request 

527 :param _: The arguments 

528 :param db_session: The database session to use 

529 :return: None 

530 """ 

531 self.deactivate_config(address, db_session, MangaNotificationConfig) 

532 

533 def on_list_new_manga_chapters( 

534 self, 

535 address: Address, 

536 _, 

537 db_session: Session 

538 ): 

539 """ 

540 Handles listing new manga chapters for a user 

541 :param address: The address that requested this 

542 :param _: The arguments 

543 :param db_session: The database session to use 

544 :return: None 

545 """ 

546 self._send_notifications(db_session, address, True, "manga") 

547 

548 def on_list_new_releasing_chapters( 

549 self, 

550 address: Address, 

551 args: Dict[str, Any], 

552 db_session: Session 

553 ): 

554 """ 

555 Handles listing new manga chapters for a user. 

556 Only sends notifications for currently releasing manga 

557 :param address: The address that requested this 

558 :param args: The arguments provided by the user 

559 :param db_session: The database session to use 

560 :return: None 

561 """ 

562 self._send_notifications( 

563 db_session, 

564 address, 

565 True, 

566 "manga", 

567 send_completed=False, 

568 mincount=args.get("mincount", 1) 

569 ) 

570 

571 def on_list_new_completed_chapters( 

572 self, 

573 address: Address, 

574 _, 

575 db_session: Session 

576 ): 

577 """ 

578 Handles listing new manga chapters for a user 

579 Only sends notifications for currently completed manga series 

580 :param address: The address that requested this 

581 :param _: The arguments 

582 :param db_session: The database session to use 

583 :return: None 

584 """ 

585 self._send_notifications( 

586 db_session, address, True, "manga", send_releasing=False 

587 ) 

588 

589 def on_list_ln_releases( 

590 self, 

591 address: Address, 

592 args: Dict[str, Any], 

593 _ 

594 ): 

595 """ 

596 Handles listing current light novel releases 

597 :param address: The user that sent this request 

598 :param args: The arguments to use 

599 :param _: The database session 

600 :return: None 

601 """ 

602 year = args.get("year") 

603 month = args.get("month") 

604 

605 now = datetime.utcnow() 

606 

607 if year is None: 

608 year = now.year 

609 if month is None: 

610 month = now.strftime("%B") 

611 

612 releases = load_ln_releases().get(year, {}).get(month.lower(), []) 

613 body = "Light Novel Releases {} {}\n\n".format(month, year) 

614 

615 for entry in releases: 

616 body += "{}: {} {} ({})\n".format( 

617 entry["day"], 

618 entry["title"], 

619 entry["volume"], 

620 entry["release_type"] 

621 ) 

622 self.send_txt(address, body) 

623 

624 def on_toggle_ranked_mode( 

625 self, 

626 address: Address, 

627 _: Dict[str, Any], 

628 db_session: Session 

629 ): 

630 """ 

631 Lets the user toggle "ranked" mode 

632 :param address: The address of the user attempting to toggle the mode 

633 :param _: The parmaters provided by the parser 

634 :param db_session: The database session to use 

635 :return: None 

636 """ 

637 setting = db_session.query(RankedModeSetting)\ 

638 .filter_by(address=address).first() 

639 if setting is None: 

640 setting = RankedModeSetting(address=address, value=False) 

641 db_session.add(setting) 

642 setting.value = not setting.value 

643 db_session.commit() 

644 self.send_txt(address, f"Successfully toggled to {setting.value}") 

645 

646 def bg_iteration(self, _: int, db_session: Session): 

647 """ 

648 Periodically checks for new notifications and sends them out 

649 :param _: The iteration count 

650 :param db_session: The database session to use 

651 :return: None 

652 """ 

653 self._update_anilist_entries(db_session) 

654 self._send_notifications(db_session)