cl0ne
/
cryptopotato-bot
1
0
Fork 0
Code Issues 9 Pull Requests Releases Activity
Browse Source

Track given titles, count streaks 

Note: AssignmentRecords has never been used, so we can safely replace
its table with new table for GivenInevitableTitle
pull/29/head
Vladislav Glinsky 5 years ago
parent
0b7e237887
commit
5c855ce370
Signed by: cl0ne GPG Key ID: 9D058DD29491782E
8 changed files with 430 additions and 112 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 123
      alembic/versions/981e95ceff06_add_title_assignment_log.py
  2. 2
      devpotato_bot/commands/daily_titles/_strings.py
  3. 300
      devpotato_bot/commands/daily_titles/assign_titles.py
  4. 3
      devpotato_bot/commands/daily_titles/models/__init__.py
  5. 32
      devpotato_bot/commands/daily_titles/models/assignment_record.py
  6. 29
      devpotato_bot/commands/daily_titles/models/given_inevitable_title.py
  7. 26
      devpotato_bot/commands/daily_titles/models/given_shuffled_title.py
  8. 27
      devpotato_bot/commands/daily_titles/models/given_title.py

123
alembic/versions/981e95ceff06_add_title_assignment_log.py
View File

@ -0,0 +1,123 @@
"""Add title assignment log
Revision ID: 981e95ceff06
Revises: f786ea9057b0
Create Date: 2020-12-06 20:50:53.222690
"""
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision = "981e95ceff06"
down_revision = "f786ea9057b0"
branch_labels = None
depends_on = None
def upgrade():
op.drop_table("daily_titles_assignment_records")
op.create_table(
"daily_titles_given_inevitable_titles",
sa.Column("id", sa.Integer(), nullable=False),
sa.Column("participant_id", sa.Integer(), nullable=False),
sa.Column("title_id", sa.Integer(), nullable=False),
sa.Column(
"given_on",
sa.DateTime(),
nullable=False,
),
sa.Column("streak_length", sa.Integer(), nullable=False),
sa.ForeignKeyConstraint(
["participant_id"],
["daily_titles_participants.id"],
name=op.f(
"fk_daily_titles_given_inevitable_titles_participant_id_daily_titles_participants"
),
onupdate="CASCADE",
ondelete="CASCADE",
),
sa.ForeignKeyConstraint(
["title_id"],
["daily_titles_inevitable_titles.id"],
name=op.f(
"fk_daily_titles_given_inevitable_titles_title_id_daily_titles_inevitable_titles"
),
onupdate="CASCADE",
ondelete="CASCADE",
),
sa.PrimaryKeyConstraint(
"id", name=op.f("pk_daily_titles_given_inevitable_titles")
),
)
op.create_table(
"daily_titles_given_shuffled_titles",
sa.Column("id", sa.Integer(), nullable=False),
sa.Column("participant_id", sa.Integer(), nullable=False),
sa.Column("title_id", sa.Integer(), nullable=False),
sa.Column(
"given_on",
sa.DateTime(),
nullable=False,
),
sa.ForeignKeyConstraint(
["participant_id"],
["daily_titles_participants.id"],
name=op.f(
"fk_daily_titles_given_shuffled_titles_title_id_daily_titles_shuffled_titles"
),
onupdate="CASCADE",
ondelete="CASCADE",
),
sa.ForeignKeyConstraint(
["title_id"],
["daily_titles_shuffled_titles.id"],
name=op.f(
"fk_daily_titles_given_shuffled_titles_title_id_daily_titles_shuffled_titles"
),
onupdate="CASCADE",
ondelete="CASCADE",
),
sa.PrimaryKeyConstraint(
"id", name=op.f("pk_daily_titles_given_shuffled_titles")
),
)
def downgrade():
op.drop_table("daily_titles_given_inevitable_titles")
op.create_table(
"daily_titles_assignment_records",
sa.Column("id", sa.Integer(), nullable=False),
sa.Column("participant_id", sa.Integer(), nullable=False),
sa.Column("title_id", sa.Integer(), nullable=False),
sa.Column("count", sa.Integer(), nullable=False),
sa.Column("last_assigned_on", sa.DateTime(), nullable=True),
sa.ForeignKeyConstraint(
["participant_id"],
["daily_titles_participants.id"],
name=op.f(
"fk_daily_titles_assignment_records_participant_id_daily_titles_participants"
),
onupdate="CASCADE",
ondelete="CASCADE",
),
sa.ForeignKeyConstraint(
["title_id"],
["daily_titles_inevitable_titles.id"],
name=op.f(
"fk_daily_titles_assignment_records_title_id_daily_titles_inevitable_titles"
),
onupdate="CASCADE",
ondelete="CASCADE",
),
sa.PrimaryKeyConstraint("id", name=op.f("pk_daily_titles_assignment_records")),
sa.UniqueConstraint(
"participant_id",
"title_id",
name=op.f("uq_daily_titles_assignment_records_participant_id"),
),
)
op.drop_table("daily_titles_given_shuffled_titles")

2
devpotato_bot/commands/daily_titles/_strings.py
View File

@ -1,5 +1,7 @@
__ACTIVITY_NAME = '📣 Daily 🏅Titles 🎲'
STREAK_MARK = '❗️'
MESSAGE__DISABLED_FOR_PRIVATE_CHATS = (
f'*{__ACTIVITY_NAME} is not available for private chats*.\n'
'\n'

300
devpotato_bot/commands/daily_titles/assign_titles.py
View File

@ -1,115 +1,257 @@
import itertools
from typing import List, Tuple, Union, Iterator
from __future__ import annotations
from sqlalchemy.orm import Session
from telegram import Chat, User, error, ChatMember
from typing import TYPE_CHECKING, Generic, TypeVar
from sqlalchemy import tuple_
from telegram import error, ChatMember
from telegram.utils.helpers import escape_markdown, mention_markdown
from .models import GroupChat, InevitableTitle, ShuffledTitle
from ._strings import STREAK_MARK
from .models import GivenInevitableTitle
from .models import GivenShuffledTitle
from .models import InevitableTitle
from .models import ShuffledTitle
from .models.title import TitleFromGroupChat
from ...sample import sample_items_inplace
if TYPE_CHECKING:
from datetime import datetime
from typing import List, Tuple, Union, Optional, Dict, Iterable
from sqlalchemy.orm import Session
from telegram import Chat, User
from .models import GroupChat
ParticipantIDPairs = List[Tuple[int, int]]
SHUFFLED_TITLES_TO_GIVE = 5
class MissingUserWrapper:
def __init__(self, user_id):
self.user_id = user_id
self.full_name = ''
self.full_name = ""
def mention_markdown_v2(self):
return mention_markdown(self.user_id, self.full_name, version=2)
def _get_titles_assignment(chat: GroupChat):
if TYPE_CHECKING:
ParticipantFormatter = Union[User, MissingUserWrapper]
def assign_titles(
session: Session, chat_data: GroupChat, tg_chat: Chat, trigger_time: datetime
):
chat_data.name = tg_chat.title
new_titles = _choose_new_titles(chat_data)
if new_titles is None:
chat_data.last_titles_plain, chat_data.last_titles = None, None
else:
inevitable, shuffled = new_titles
_save_given_inevitable_titles(
session, trigger_time, chat_data.last_triggered, inevitable
)
_save_given_shuffled_titles(session, trigger_time, shuffled)
_refresh_user_data(tg_chat, chat_data, new_titles)
new_texts = _build_titles_text(inevitable, shuffled)
chat_data.last_titles_plain, chat_data.last_titles = new_texts
chat_data.last_triggered = trigger_time
session.commit()
TitleT = TypeVar("TitleT", bound=TitleFromGroupChat)
class NewTitles(Generic[TitleT]):
def __init__(self, titles: List[TitleT], participant_id_pairs: ParticipantIDPairs):
assert len(titles) == len(participant_id_pairs)
self.titles: List[TitleT] = titles
self.participant_id_pairs = participant_id_pairs
self.tg_users: List[ParticipantFormatter] = []
def is_empty(self) -> bool:
return len(self.participant_id_pairs) == 0
def get_title_ids(self) -> Iterable[int]:
return (i.id for i in self.titles)
def get_tg_user_ids(self) -> Iterable[int]:
return (i for _, i in self.participant_id_pairs)
def get_participant_ids(self) -> Iterable[int]:
return (i for i, _ in self.participant_id_pairs)
def format_lines(self, lines_plain: list, lines_mention: list):
for title, user in zip(self.titles, self.tg_users):
user_name = escape_markdown(user.full_name, version=2)
title_text = escape_markdown(title.text, version=2)
lines_plain.append(f" {user_name} \\- {title_text}")
lines_mention.append(f" {user.mention_markdown_v2()} \\- {title_text}")
class NewInevitableTitles(NewTitles[InevitableTitle]):
def __init__(
self, titles: List[InevitableTitle], participant_id_pairs: ParticipantIDPairs
):
super().__init__(titles, participant_id_pairs)
self.records: List[GivenInevitableTitle] = []
@staticmethod
def _format_streak_marks(streak_length: int):
if streak_length < 2:
return ""
return f"{streak_length}x{STREAK_MARK} "
def format_lines(self, lines_plain: list, lines_mention: list):
for title, user, record in zip(self.titles, self.tg_users, self.records):
user_name = escape_markdown(user.full_name, version=2)
title_text = escape_markdown(title.text, version=2)
streak_mark = self._format_streak_marks(record.streak_length)
lines_plain.append(f" {streak_mark}{user_name} \\- {title_text}")
lines_mention.append(
f" {streak_mark}{user.mention_markdown_v2()} \\- {title_text}"
)
NewShuffledTitles = NewTitles[ShuffledTitle]
def _choose_new_titles(
chat: GroupChat,
) -> Optional[Tuple[NewInevitableTitles, NewShuffledTitles]]:
participant_ids = chat.get_participant_ids()
participant_count = len(participant_ids)
if not participant_count:
return None
inevitable_titles: List[InevitableTitle] = chat.inevitable_titles[:participant_count]
inevitable_titles: List[InevitableTitle] = chat.inevitable_titles[
:participant_count
]
inevitable_titles_count = len(inevitable_titles)
if inevitable_titles_count:
sample_items_inplace(participant_ids, inevitable_titles_count)
shuffled_titles_needed = min(participant_count - inevitable_titles_count, SHUFFLED_TITLES_TO_GIVE)
shuffled_titles_needed = min(
participant_count - inevitable_titles_count, SHUFFLED_TITLES_TO_GIVE
)
shuffled_titles = chat.dequeue_shuffled_titles(shuffled_titles_needed)
shuffled_titles_count = len(shuffled_titles)
if shuffled_titles_count:
item_limit = participant_count - inevitable_titles_count
sample_items_inplace(participant_ids, shuffled_titles_count, item_limit=item_limit)
sampled_count = shuffled_titles_count + inevitable_titles_count
sampled_ids = list(itertools.islice(reversed(participant_ids), sampled_count))
return sampled_ids, inevitable_titles, shuffled_titles
def _get_assigned_users(tg_chat: Chat, chat: GroupChat, user_ids):
users = []
if not user_ids:
return users
sample_size = shuffled_titles_count + inevitable_titles_count
sample_items_inplace(participant_ids, sample_size)
first_shuffled_title = -inevitable_titles_count - 1
return (
NewInevitableTitles(
inevitable_titles, participant_ids[:first_shuffled_title:-1]
),
NewShuffledTitles(
shuffled_titles,
participant_ids[first_shuffled_title : -sample_size - 1 : -1],
),
)
def _save_given_inevitable_titles(
session: Session,
given_on: datetime,
old_given_on: Optional[datetime],
new_titles: NewInevitableTitles,
):
"""Calculates new streak lengths and adds new given inevitable titles to database"""
if old_given_on is None:
previous_streaks = None
else:
previous_streaks = _load_old_streaks(session, old_given_on, new_titles)
for participant_id, title in zip(
new_titles.get_participant_ids(), new_titles.titles
):
new_streak_length = 1
if previous_streaks:
new_streak_length += previous_streaks.get(participant_id, 0)
record = GivenInevitableTitle(
participant_id=participant_id,
title=title,
given_on=given_on,
streak_length=new_streak_length,
)
new_titles.records.append(record)
session.add(record)
def _load_old_streaks(
session: Session, old_given_on: datetime, records_filter: NewInevitableTitles
) -> Dict[int, int]:
query = session.query(
GivenInevitableTitle.participant_id, GivenInevitableTitle.streak_length
).filter(
tuple_(GivenInevitableTitle.participant_id, GivenInevitableTitle.title_id).in_(
zip(records_filter.get_participant_ids(), records_filter.get_title_ids())
),
GivenInevitableTitle.given_on == old_given_on,
)
return dict(query)
def _save_given_shuffled_titles(
session: Session,
given_on: datetime,
new_titles: NewShuffledTitles,
):
"""Adds new given shuffled titles to database"""
for participant_id, title in zip(
new_titles.get_participant_ids(), new_titles.titles
):
record = GivenShuffledTitle(
participant_id=participant_id, title=title, given_on=given_on
)
session.add(record)
def _refresh_user_data(tg_chat: Chat, chat: GroupChat, new_titles: Iterable[NewTitles]):
"""Gets current names and usernames of participants, updates database with this data and participation status"""
if all(map(NewTitles.is_empty, new_titles)):
return
participants_new_data, load_from_db = [], {}
for participant_id, user_id in user_ids:
new_user_data = dict(id=participant_id)
try:
member = tg_chat.get_member(user_id)
u = member.user
new_user_data.update(full_name=u.full_name, username=u.username)
if member.status in (ChatMember.KICKED, ChatMember.LEFT):
new_user_data.update(is_active=False)
participants_new_data.append(new_user_data)
except error.BadRequest as e:
if e.message == 'User not found':
new_user_data.update(is_active=False, is_missing=True)
for title_set in new_titles:
for participant_id, user_id in title_set.participant_id_pairs:
new_user_data = dict(id=participant_id)
try:
member = tg_chat.get_member(user_id)
user = member.user
new_user_data.update(full_name=user.full_name, username=user.username)
if member.status in (ChatMember.KICKED, ChatMember.LEFT):
new_user_data.update(is_active=False)
participants_new_data.append(new_user_data)
u = MissingUserWrapper(user_id)
load_from_db[participant_id] = u
users.append(u)
except error.BadRequest as ex:
if ex.message == "User not found":
new_user_data.update(is_active=False, is_missing=True)
participants_new_data.append(new_user_data)
user = MissingUserWrapper(user_id)
load_from_db[participant_id] = user
title_set.tg_users.append(user)
session: Session = Session.object_session(chat)
from .models import Participant
session.bulk_update_mappings(Participant, participants_new_data)
session.commit()
loaded_names = session.query(
Participant.id, Participant.full_name
).filter(
Participant.id.in_(load_from_db)
).all()
loaded_names = (
session.query(Participant.id, Participant.full_name)
.filter(Participant.id.in_(load_from_db))
.all()
)
for participant_id, participant_name in loaded_names:
load_from_db[participant_id].full_name = participant_name
return users
def _format_assignment_lines(participants: Iterator, titles: list, lines_plain: list, lines_mention: list):
for title, user in zip(titles, participants):
user_name = escape_markdown(user.full_name, version=2)
title_text = escape_markdown(title.text, version=2)
lines_plain.append(f' {user_name} \\- {title_text}')
lines_mention.append(f' {user.mention_markdown_v2()} \\- {title_text}')
def _build_titles_text(
participants: List[Union[User, MissingUserWrapper]],
inevitable_titles: List[InevitableTitle],
shuffled_titles: List[ShuffledTitle]
inevitable_titles: NewInevitableTitles,
shuffled_titles: NewShuffledTitles,
) -> Tuple[str, str]:
lines_mention, lines_plain = [], []
participants_iter = iter(participants)
_format_assignment_lines(participants_iter, inevitable_titles, lines_plain, lines_mention)
if inevitable_titles and shuffled_titles:
lines_plain.append('')
lines_mention.append('')
_format_assignment_lines(participants_iter, shuffled_titles, lines_plain, lines_mention)
return '\n'.join(lines_plain), '\n'.join(lines_mention)
def assign_titles(session, chat_data, tg_chat, trigger_time):
assignment = _get_titles_assignment(chat_data)
new_titles = None, None
if assignment is not None:
participant_ids, *titles = assignment
users = _get_assigned_users(tg_chat, chat_data, participant_ids)
new_titles = _build_titles_text(users, *titles)
chat_data.last_titles_plain, chat_data.last_titles = new_titles
chat_data.last_triggered = trigger_time
chat_data.name = tg_chat.title
session.commit()
lines_mention, lines_plain = [], [] # type: List[str], List[str]
inevitable_titles.format_lines(lines_plain, lines_mention)
if not (inevitable_titles.is_empty() or shuffled_titles.is_empty()):
lines_plain.append("")
lines_mention.append("")
shuffled_titles.format_lines(lines_plain, lines_mention)
return "\n".join(lines_plain), "\n".join(lines_mention)

3
devpotato_bot/commands/daily_titles/models/__init__.py
View File

@ -1,4 +1,5 @@
from .assignment_record import AssignmentRecord
from .given_inevitable_title import GivenInevitableTitle
from .given_shuffled_title import GivenShuffledTitle
from .group_chat import GroupChat
from .inevitable_title import InevitableTitle
from .participant import Participant

32
devpotato_bot/commands/daily_titles/models/assignment_record.py
View File

@ -1,32 +0,0 @@
from sqlalchemy import Column, Integer, ForeignKey, UniqueConstraint
from sqlalchemy.orm import relationship
from .base import Base
from .inevitable_title import InevitableTitle
from .participant import Participant
from .utc_datetime import UTCDateTime
class AssignmentRecord(Base):
__tablename__ = f'{Base.TABLENAME_PREFIX}assignment_records'
__table_args__ = (
UniqueConstraint('participant_id', 'title_id'),
)
id = Column(Integer, primary_key=True)
participant_id = Column(Integer, ForeignKey(Participant.id, onupdate="CASCADE", ondelete="CASCADE"), nullable=False)
participant = relationship('Participant')
title_id = Column(Integer, ForeignKey(InevitableTitle.id, onupdate="CASCADE", ondelete="CASCADE"), nullable=False)
title = relationship('InevitableTitle')
count = Column(Integer, nullable=False, default=0)
last_assigned_on = Column(UTCDateTime)
def __repr__(self):
return ('<AssignmentRecord('
f'participant_id={self.participant_id}, '
f'title_id={self.title_id}, '
f'count={self.count}, '
f'last_assigned_on="{self.last_assigned_on}"'
')>')

29
devpotato_bot/commands/daily_titles/models/given_inevitable_title.py
View File

@ -0,0 +1,29 @@
from sqlalchemy import Column, Integer, ForeignKey
from sqlalchemy.orm import relationship
from .base import Base
from .given_title import GivenTitle
from .inevitable_title import InevitableTitle
class GivenInevitableTitle(GivenTitle):
__tablename__ = f"{Base.TABLENAME_PREFIX}given_inevitable_titles"
title_id = Column(
Integer,
ForeignKey(InevitableTitle.id, onupdate="CASCADE", ondelete="CASCADE"),
nullable=False,
)
title = relationship("InevitableTitle")
streak_length = Column(Integer, nullable=False, default=1)
def __repr__(self):
return (
"<GivenInevitableTitle("
f"participant_id={self.participant_id}, "
f"title_id={self.title_id}, "
f'given_on="{self.given_on}", '
f'streak_length="{self.streak_length}"'
")>"
)

26
devpotato_bot/commands/daily_titles/models/given_shuffled_title.py
View File

@ -0,0 +1,26 @@
from sqlalchemy import Column, Integer, ForeignKey
from sqlalchemy.orm import relationship
from .base import Base
from .given_title import GivenTitle
from .shuffled_title import ShuffledTitle
class GivenShuffledTitle(GivenTitle):
__tablename__ = f"{Base.TABLENAME_PREFIX}given_shuffled_titles"
title_id = Column(
Integer,
ForeignKey(ShuffledTitle.id, onupdate="CASCADE", ondelete="CASCADE"),
nullable=False,
)
title = relationship("ShuffledTitle")
def __repr__(self):
return (
"<GivenShuffledTitle("
f"participant_id={self.participant_id}, "
f"title_id={self.title_id}, "
f'given_on="{self.given_on}"'
")>"
)

27
devpotato_bot/commands/daily_titles/models/given_title.py
View File

@ -0,0 +1,27 @@
from sqlalchemy import Column, Integer, ForeignKey
from sqlalchemy.ext.declarative import declared_attr
from sqlalchemy.orm import relationship
from .base import Base
from .participant import Participant
from .utc_datetime import UTCDateTime
class GivenTitle(Base):
__abstract__ = True
id = Column(Integer, primary_key=True)
@declared_attr
def participant_id(cls):
return Column(
Integer,
ForeignKey(Participant.id, onupdate="CASCADE", ondelete="CASCADE"),
nullable=False,
)
@declared_attr
def participant(cls):
return relationship("Participant")
given_on = Column(UTCDateTime, nullable=False)
Write Preview
Loading…
Cancel
Save