Function type_of_rooms_in_channel has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

    def type_of_rooms_in_channel(self, channel_id: str) -> str:
        object_type = self.env.cache.get_type_of_rooms_in_channel(channel_id)
        if object_type is not None:
            return object_type

• Read up
  Read up

Function ban_user_channel has 6 arguments (exceeds 4 allowed). Consider refactoring.

Open

    def ban_user_channel(self, user_id: str, ban_timestamp: str, ban_duration: str, channel_id: str, reason: str=None, banner_id: str=None):

Function is_banned_from_channel has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

    def is_banned_from_channel(self, channel_id: str, user_id: str) -> (bool, Union[str, None]):
        now = datetime.utcnow()
        duration, time, username = self.env.cache.get_channel_ban_timestamp(channel_id, user_id)
        if time is not None and len(time.strip()) != 0:
            if time == '':

• Read up
  Read up

Function is_banned_from_room has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

    def is_banned_from_room(self, room_id: str, user_id: str) -> (bool, Union[str, None]):
        now = datetime.utcnow()
        duration, time, username = self.env.cache.get_room_ban_timestamp(room_id, user_id)
        if time is not None and len(time.strip()) != 0:
            if time == '':

• Read up
  Read up

Function ban_user_channel has 6 arguments (exceeds 4 allowed). Consider refactoring.

Open

    def ban_user_channel(self, user_id: str, ban_timestamp: str, ban_duration: str, channel_id: str, reason: str=None, banner_id: str=None) -> None:

Function room_created has 6 arguments (exceeds 4 allowed). Consider refactoring.

Open

    def room_created(self, user_id, user_name, room_id, room_name, session_ids, namespace) -> None:

Function do_get has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

    def do_get(self):
        the_json = self.validate_json()
        logger.debug('GET request: %s' % str(the_json))

        room_id = the_json.get('room_id', '')

• Read up
  Read up

Function __call__ has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

    def __call__(self, *args, **kwargs):
        activity = args[0]
        # env = args[1]
        # acl_type = args[2]
        # acl_values = args[3]

• Read up
  Read up

Function set_user_offline has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

    def set_user_offline(self, user_id: str) -> None:
        @with_session
        def _set_user_offline(session=None):
            status = session.query(UserStatus).filter(UserStatus.uuid == user_id).first()
            if status is None:

• Read up
  Read up

Function do_post has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

    def do_post(self):
        is_valid, msg, json = self.validate_json()
        if not is_valid:
            logger.error('invalid json: %s' % msg)
            raise RuntimeError('invalid json')

• Read up
  Read up

Function mute_user has 6 arguments (exceeds 4 allowed). Consider refactoring.

Open

    def mute_user(

Function __call__ has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

    def __call__(self, *args, **kwargs):
        # activity = args[0]
        env = args[1]
        acl_type = args[2]
        acl_values = args[3]

• Read up
  Read up

Avoid deeply nested control flow statements.

Open

                        if not utils.can_send_whisper_in_channel(activity, channel_id):
                            return False, ECodes.NOT_ALLOWED_TO_WHISPER_CHANNEL, 'not allowed to whisper in channel'

Avoid deeply nested control flow statements.

Open

                if error_msg:
                    print("room name: {}".format(room_name))
                    print(error_msg)
                    print("\n" * 10)
                    sys.exit(1)

Function kick has 6 arguments (exceeds 4 allowed). Consider refactoring.

Open

    def kick(self, orig_data: dict, activity: Activity, room_id: str, user_id: str, user_sids: list, namespace: str) -> None:

Function ban_room has 6 arguments (exceeds 4 allowed). Consider refactoring.

Open

    def ban_room(self, data: dict, act: Activity, room_id: str, user_id: str, user_sids: list, namespace: str) -> None:

Function consume has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

def consume():
    if len(environ.env.config) == 0 or environ.env.config.get(ConfigKeys.TESTING, False):
        return

    if environ.env.config.get(ConfigKeys.TYPE, domain=ConfigKeys.QUEUE) == "mock":

• Read up
  Read up

Function ban_globally has 6 arguments (exceeds 4 allowed). Consider refactoring.

Open

    def ban_globally(self, data: dict, act: Activity, rooms: dict, user_id: str, user_sids: list, namespace: str) -> None:

Consider simplifying this complex logical expression.

Open

return{empty:!1,unusedTokens:[],unusedInput:[],overflow:-2,charsLeftOver:0,nullInput:!1,invalidMonth:null,invalidFormat:!1,userInvalidated:!1,iso:!1,parsedDateParts:[],meridiem:null}}function l(a){return null==a._pf&&(a._pf=k()),a._pf}function m(a){if(null==a._isValid){var b=l(a),c=nd.call(b.parsedDateParts,function(a){return null!=a}),d=!isNaN(a._d.getTime())&&b.overflow<0&&!b.empty&&!b.invalidMonth&&!b.invalidWeekday&&!b.nullInput&&!b.invalidFormat&&!b.userInvalidated&&(!b.meridiem||b.meridiem&&c);if(a._strict&&(d=d&&0===b.charsLeftOver&&0===b.unusedTokens.length&&void 0===b.bigHour),null!=Object.isFrozen&&Object.isFrozen(a))return d;a._isValid=d}return a._isValid}function n(a){var b=j(NaN);return null!=a?i(l(b),a):l(b).userInvalidated=!0,b}function o(a){return void 0===a}function p(a,b){var c,d,e;if(o(b._isAMomentObject)||(a._isAMomentObject=b._isAMomentObject),o(b._i)||(a._i=b._i),o(b._f)||(a._f=b._f),o(b._l)||(a._l=b._l),o(b._strict)||(a._strict=b._strict),o(b._tzm)||(a._tzm=b._tzm),o(b._isUTC)||(a._isUTC=b._isUTC),o(b._offset)||(a._offset=b._offset),o(b._pf)||(a._pf=l(b)),o(b._locale)||(a._locale=b._locale),od.length>0)for(c in od)d=od[c],e=b[d],o(e)||(a[d]=e);return a}

Consider simplifying this complex logical expression.

Open

if(
// make the regex if we don't have it already
e=j([2e3,1]).day(d),c&&!this._fullWeekdaysParse[d]&&(this._fullWeekdaysParse[d]=new RegExp("^"+this.weekdays(e,"").replace(".",".?")+"$","i"),this._shortWeekdaysParse[d]=new RegExp("^"+this.weekdaysShort(e,"").replace(".",".?")+"$","i"),this._minWeekdaysParse[d]=new RegExp("^"+this.weekdaysMin(e,"").replace(".",".?")+"$","i")),this._weekdaysParse[d]||(f="^"+this.weekdays(e,"")+"|^"+this.weekdaysShort(e,"")+"|^"+this.weekdaysMin(e,""),this._weekdaysParse[d]=new RegExp(f.replace(".",""),"i")),c&&"dddd"===b&&this._fullWeekdaysParse[d].test(a))return d;if(c&&"ddd"===b&&this._shortWeekdaysParse[d].test(a))return d;if(c&&"dd"===b&&this._minWeekdaysParse[d].test(a))return d;if(!c&&this._weekdaysParse[d].test(a))return d}}