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>
4This file is part of otaku-info-bot.
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.
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.
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"""
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
39class OtakuInfoBot(Bot):
40 """
41 The OtakuInfo Bot class that defines the anime reminder
42 functionality.
43 """
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 """
51 @classmethod
52 def name(cls) -> str:
53 """
54 :return: The name of the bot
55 """
56 return "otaku-info-bot"
58 @classmethod
59 def version(cls) -> str:
60 """
61 :return: The current version of the bot
62 """
63 return version
65 @classmethod
66 def parsers(cls) -> List[CommandParser]:
67 """
68 :return: A list of parser the bot supports for commands
69 """
70 return [OtakuInfoCommandParser()]
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
90 if exists:
91 msg = "Configuration already activated"
92 self.send_txt(address, msg, "Already Active")
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()
106 self.send_txt(address, "Configuration Activated", "Activated")
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()
124 if existing is not None:
125 db_session.delete(existing)
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)
132 db_session.commit()
134 self.send_txt(address, "Deactivated Configuration", "Deactivated")
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")
144 newest_anime_episodes = load_newest_episodes()
145 manga_progress = {} # type: Dict[int, int]
147 for config_cls in [AnimeNotificationConfig, MangaNotificationConfig]:
149 media_type = config_cls.media_type()
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 )
158 anilist_ids = []
159 for entry in anilist:
161 anilist_id = entry["media"]["id"]
162 anilist_ids.append(anilist_id)
164 releasing = entry["media"]["status"] == "RELEASING"
165 completed = entry["media"]["status"] == "FINISHED"
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
173 user_progress = entry["progress"]
174 user_score = entry["score"]
176 db_entry = db_session.query(AnilistEntry).filter_by(
177 anilist_id=anilist_id, media_type=media_type
178 ).first()
180 # Calculate newest
181 if media_type == "anime":
182 latest = newest_anime_episodes.get(anilist_id)
183 if db_entry is None:
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
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
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
222 notification = db_session.query(Notification).filter_by(
223 entry=db_entry, address=config.address
224 ).first()
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
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)
247 db_session.commit()
248 self.logger.info("Finished updating anilist entries")
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")
279 due = {} # type: Dict[int, Dict[str, Any]]
281 for notification in db_session.query(Notification).all():
283 if notification.entry.releasing and not send_releasing:
284 continue
285 elif notification.entry.completed and not send_completed:
286 continue
288 address_id = notification.address_id
290 if address_limit is not None and address_limit.id != address_id:
291 continue
293 if address_id not in due:
294 due[address_id] = {
295 "address": notification.address,
296 "anime": [],
297 "manga": []
298 }
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)
307 for _, data in due.items():
308 address = data["address"]
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
317 for media_type in ["manga", "anime"]:
319 if media_type_limit is not None \
320 and media_type != media_type_limit:
321 continue
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)
327 if use_ranked:
328 notifications.sort(
329 key=lambda x: x.user_score, reverse=True
330 )
332 media_name = media_type[0].upper() + media_type[1:]
333 unit_type = "Episode" if media_type == "anime" else "Chapter"
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")
361 for notification in notifications:
362 notification.last_update = notification.entry.latest
364 db_session.commit()
365 self.logger.info("Finished Sending Notifications")
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()
382 if cached is None:
384 self.logger.debug("Creating new manga chapter guess cache for {}"
385 .format(anilist_id))
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
394 cached = MangaChapterGuess(id=anilist_id, last_check=0)
395 db_session.add(cached)
397 cached.update()
398 cached.last_check += delay
400 was_updated = cached.update()
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))
409 db_session.commit()
411 return cached.guess
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 )
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)
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")
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 )
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 )
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 )
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)
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")
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 )
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 )
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")
605 now = datetime.utcnow()
607 if year is None:
608 year = now.year
609 if month is None:
610 month = now.strftime("%B")
612 releases = load_ln_releases().get(year, {}).get(month.lower(), [])
613 body = "Light Novel Releases {} {}\n\n".format(month, year)
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)
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}")
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)