The KISS Twitch bot
No puede seleccionar más de 25 temas Los temas deben comenzar con una letra o número, pueden incluir guiones ('-') y pueden tener hasta 35 caracteres de largo.

moderator.py 4.6 KiB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150
  1. # Twason - The KISS Twitch bot
  2. # Copyright (C) 2021 Jérôme Deuchnord
  3. #
  4. # This program is free software: you can redistribute it and/or modify
  5. # it under the terms of the GNU Affero General Public License as
  6. # published by the Free Software Foundation, either version 3 of the
  7. # License, or (at your option) any later version.
  8. #
  9. # This program is distributed in the hope that it will be useful,
  10. # but WITHOUT ANY WARRANTY; without even the implied warranty of
  11. # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  12. # GNU Affero General Public License for more details.
  13. #
  14. # You should have received a copy of the GNU Affero General Public License
  15. # along with this program. If not, see <https://www.gnu.org/licenses/>.
  16. from abc import ABC, abstractmethod
  17. from enum import Enum
  18. from typing import Union
  19. from datetime import datetime, timedelta
  20. EPOCH = datetime(1970, 1, 1)
  21. class ModerationDecision(Enum):
  22. ABSTAIN = -1
  23. DELETE_MSG = 0
  24. TIMEOUT_USER = 1
  25. class Moderator(ABC):
  26. message: str
  27. decision: ModerationDecision
  28. def __init__(self, message: str, decision: ModerationDecision, timeout_duration: Union[None, int]):
  29. self.message = message
  30. self.timeout_duration = timeout_duration
  31. self.decision = decision
  32. @abstractmethod
  33. def get_name(self) -> str:
  34. pass
  35. @abstractmethod
  36. def vote(self, msg: str, author: str) -> ModerationDecision:
  37. pass
  38. class CapsLockModerator(Moderator):
  39. def __init__(
  40. self,
  41. message: str,
  42. decision: ModerationDecision,
  43. timeout_duration: Union[None, int],
  44. min_size: int,
  45. threshold: int
  46. ):
  47. super().__init__(message, decision, timeout_duration)
  48. self.min_size = min_size
  49. self.threshold = threshold / 100
  50. def get_name(self) -> str:
  51. return 'Caps Lock'
  52. def vote(self, msg: str, author: str) -> ModerationDecision:
  53. msg = ''.join(filter(str.isalpha, msg))
  54. if len(msg) < self.min_size:
  55. return ModerationDecision.ABSTAIN
  56. n = 0
  57. for char in msg:
  58. if char.strip() == '':
  59. continue
  60. if char == char.upper():
  61. n += 1
  62. if n / len(msg) >= self.threshold:
  63. return self.decision
  64. return ModerationDecision.ABSTAIN
  65. class FloodModerator(Moderator):
  66. def __init__(
  67. self,
  68. message: str,
  69. decision: ModerationDecision,
  70. timeout_duration: Union[None, int],
  71. max_word_length: Union[None, int],
  72. raid_cooldown: Union[None, int],
  73. ignore_hashtags: bool,
  74. max_msg_occurrences: Union[None, int],
  75. min_time_between_occurrence: Union[None, int]
  76. ):
  77. super().__init__(message, decision, timeout_duration)
  78. self.max_word_length = max_word_length
  79. self.raid_cooldown = raid_cooldown
  80. self.last_raid = EPOCH
  81. self.ignore_hashtags = ignore_hashtags
  82. self.max_msg_occurrences = max_msg_occurrences
  83. self.min_time_between_occurrence = min_time_between_occurrence
  84. self.last_msgs = []
  85. def get_name(self) -> str:
  86. return 'Flood'
  87. def vote(self, msg: str, author: str) -> ModerationDecision:
  88. if self.raid_cooldown is not None and self.last_raid + timedelta(minutes=self.raid_cooldown) > datetime.now():
  89. return ModerationDecision.ABSTAIN
  90. if self.max_word_length is not None:
  91. for word in msg.split(' '):
  92. if word.startswith('#'):
  93. continue
  94. if len(word) > self.max_word_length:
  95. return ModerationDecision.TIMEOUT_USER
  96. if self.max_msg_occurrences is None or self.min_time_between_occurrence is None:
  97. return ModerationDecision.ABSTAIN
  98. clean_msg = None
  99. for last_msg in self.last_msgs:
  100. if last_msg['first-occurrence'] + timedelta(seconds=self.min_time_between_occurrence) <= datetime.now():
  101. clean_msg = last_msg
  102. break
  103. if author != last_msg['author'] or msg != last_msg['message']:
  104. break
  105. last_msg['occurrences'] += 1
  106. if last_msg['occurrences'] >= self.max_msg_occurrences:
  107. return ModerationDecision.TIMEOUT_USER
  108. if clean_msg is not None:
  109. self.last_msgs.remove(clean_msg)
  110. self.last_msgs.append({
  111. 'first-occurrence': datetime.now(),
  112. 'author': author,
  113. 'message': msg,
  114. 'occurrences': 1
  115. })
  116. return ModerationDecision.ABSTAIN
  117. def declare_raid(self):
  118. self.last_raid = datetime.now()