def _print_logs()

in source/idea/idea-sdk/src/ideasdk/shell/log_tail.py [0:0]

• 60 lines of code
• 18 McCabe index (conditional complexity)

    def _print_logs(self):

        tail = sh.Command('tail')

        tail_params = []

        if len(self.files) == 1:
            if self.is_follow:
                tail_params.append(f'-{self.tail}f')
            else:
                tail_params.append(f'-{self.tail}')
        else:
            if self.is_follow:
                tail_params.append('-f')

        tail_params += self.files

        tokens = self.search_tokens
        has_tokens = tokens and len(tokens) > 0

        def print_line(line):
            if not has_tokens:
                self.context.echo(line, end='')
                return

            if self.is_and:
                if re.match(regex_expr, line):
                    text = Text(line.rstrip())
                    text.highlight_words(self.search_tokens, style='green')
                    self.context.print(text)
                return

            found = False
            for t in self.search_tokens:
                if t in line:
                    found = True
                    break

            if not found:
                return

            text = Text(line.rstrip())
            text.highlight_regex(regex_expr, style='green')
            self.context.print(text)

        if has_tokens:

            if self.is_and:
                regex_expr = ''
                for token in tokens:
                    regex_expr += f'(?=.*?\\b{token}\\b)'
                regex_expr += '.*'

                tail = tail.bake(*tail_params, _iter=True, _bg=True, _out=print_line)

                if not self._print_cmd(cmd=f'{tail} > custom regex filter: {regex_expr}'):
                    return

                process = tail()

            else:
                regex_expr = f"({'|'.join(tokens)})"

                tail = tail.bake(*tail_params, _iter=True, _bg=True, _out=print_line)

                if not self._print_cmd(cmd=f'{tail} | grep -E \'{regex_expr}\''):
                    return

                process = tail()

        else:

            tail = tail.bake(*tail_params, _iter=True, _bg=True, _out=print_line)
            if not self._print_cmd(cmd=str(tail)):
                return

            process = tail()

        try:
            process.wait()
        except KeyboardInterrupt:
            print()
            process.kill()