rDrama
/
rDrama
9
10
Fork 8
Code Issues Pull Requests Packages Releases Wiki Activity

two-way-blocks for ping groups

pull/136/head
Aevann 2023-03-01 22:28:34 +02:00
parent cb76ab0da9
commit 712239157f
3 changed files with 23 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
files/classes/user.py
View File

@ -478,6 +478,13 @@ class User(Base):
or_(and_(UserBlock.user_id == self.id, UserBlock.target_id == other.id), and_(
UserBlock.user_id == other.id, UserBlock.target_id == self.id))).first()
@property
@lazy
def all_twoway_blocks(self):
return [x[0] for x in g.db.query(UserBlock.target_id).filter_by(user_id=self.id).all() + \
g.db.query(UserBlock.user_id).filter_by(target_id=self.id).all()]
def validate_2fa(self, token):
x = pyotp.TOTP(self.mfa_secret)

9
files/helpers/alerts.py
View File

@ -129,7 +129,7 @@ def NOTIFY_USERS(text, v):
notify_users = set()
for word, id in NOTIFIED_USERS.items():
if word in text and id not in notify_users:
if word in text:
notify_users.add(id)
if FEATURES['PING_GROUPS']:
@ -151,14 +151,15 @@ def NOTIFY_USERS(text, v):
names = set(m.group(2) for m in mention_regex.finditer(text))
for user in get_users(names, graceful=True):
if v.id != user.id and not v.any_block_exists(user):
notify_users.add(user.id)
user_ids = get_users(names, ids_only=True, graceful=True)
notify_users.update(user_ids)
if SITE_NAME == "WPD" and 'daisy' in text:
admin_ids = [x[0] for x in g.db.query(User.id).filter(User.admin_level >= PERMS['NOTIFICATIONS_SPECIFIC_WPD_COMMENTS']).all()]
notify_users.update(admin_ids)
notify_users = set([id for id in notify_users if id not in v.all_twoway_blocks])
return notify_users - bots - {v.id, 0}

13
files/helpers/get.py
View File

@ -60,13 +60,19 @@ def get_user(username:Optional[str], v:Optional[User]=None, graceful=False, incl
user = add_block_props(user, v)
return user
def get_users(usernames:Iterable[str], graceful=False) -> List[User]:
def get_users(usernames:Iterable[str], ids_only=False, graceful=False) -> List[User]:
if not usernames: return []
usernames = [sanitize_username(n) for n in usernames]
if not any(usernames):
if graceful and len(usernames) == 0: return []
abort(404)
users = g.db.query(User).filter(
if ids_only:
users = g.db.query(User.id)
else:
users = g.db.query(User)
users = users.filter(
or_(
User.username.ilike(any_(usernames)),
User.original_username.ilike(any_(usernames))
@ -76,6 +82,9 @@ def get_users(usernames:Iterable[str], graceful=False) -> List[User]:
if len(users) != len(usernames) and not graceful:
abort(404)
if ids_only:
users = [x[0] for x in users]
return users
def get_account(id:Union[str, int], v:Optional[User]=None, graceful=False, include_blocks=False) -> Optional[User]: