meerschaum.utils.formatting

137def colored(text: str, *colors, as_rich_text: bool=False, **kw) -> Union[str, 'rich.text.Text']:
138    """Apply colors and rich styles to a string.
139    If a `style` keyword is provided, a `rich.text.Text` object will be parsed into a string.
140    Otherwise attempt to use the legacy `more_termcolor.colored` method.
141
142    Parameters
143    ----------
144    text: str
145        The string to apply formatting to.
146        
147    *colors:
148        A list of colors to pass to `more_termcolor.colored()`.
149        Has no effect if `style` is provided.
150
151    style: str, default None
152        If provided, pass to `rich` for processing.
153
154    as_rich_text: bool, default False
155        If `True`, return a `rich.Text` object.
156        `style` must be provided.
157        
158    **kw:
159        Keyword arguments to pass to `rich.text.Text` or `more_termcolor`.
160        
161
162    Returns
163    -------
164    An ANSI-formatted string or a `rich.text.Text` object if `as_rich_text` is `True`.
165
166    """
167    from meerschaum.utils.packages import import_rich, attempt_import
168    global _colorama_init
169    _init()
170    with _locks['_colorama_init']:
171        if not _colorama_init:
172            _colorama_init = _init()
173
174    if 'style' in kw:
175        rich = import_rich()
176        rich_text = attempt_import('rich.text')
177        text_obj = rich_text.Text(text, **kw)
178        if as_rich_text:
179            return text_obj
180        return rich_text_to_str(text_obj)
181
182    more_termcolor = attempt_import('more_termcolor', lazy=False)
183    try:
184        colored_text = more_termcolor.colored(text, *colors, **kw)
185    except Exception as e:
186        colored_text = None
187
188    if colored_text is not None:
189        return colored_text
190
191    try:
192        _colors = translate_rich_to_termcolor(*colors)
193        colored_text = more_termcolor.colored(text, *_colors, **kw)
194    except Exception as e:
195        colored_text = None
196
197    if colored_text is None:
198        ### NOTE: warn here?
199        return text
200
201    return colored_text

487def extract_stats_from_message(
488    message: str,
489    stat_keys: Optional[List[str]] = None,
490) -> Dict[str, int]:
491    """
492    Given a sync message, return the insert, update, upsert stats from within.
493
494    Parameters
495    ----------
496    message: str
497        The message to parse for statistics.
498
499    stat_keys: Optional[List[str]], default None
500        If provided, search for these words (case insensitive) in the message.
501        Defaults to `['inserted', 'updated', 'upserted']`.
502
503    Returns
504    -------
505    A dictionary mapping the stat keys to the total number of rows affected.
506    """
507    stat_keys = stat_keys or ['inserted', 'updated', 'upserted', 'checked']
508    lines_stats = [extract_stats_from_line(line, stat_keys) for line in message.split('\n')]
509    message_stats = {
510        stat_key: sum(stats.get(stat_key, 0) for stats in lines_stats)
511        for stat_key in stat_keys
512    }
513    return message_stats

469def fill_ansi(string: str, style: str = '') -> str:
470    """
471    Fill in non-formatted segments of ANSI text.
472
473    Parameters
474    ----------
475    string: str
476        A string which contains ANSI escape codes.
477
478    style: str
479        Style arguments to pass to `rich.text.Text`.
480
481    Returns
482    -------
483    A string with ANSI styling applied to the segments which don't yet have a style applied.
484    """
485    from meerschaum.utils.packages import import_rich, attempt_import
486    from meerschaum.utils.misc import iterate_chunks
487    _ = import_rich()
488    rich_ansi, rich_text = attempt_import('rich.ansi', 'rich.text')
489    Text = rich_text.Text
490    try:
491        msg = Text.from_ansi(string)
492    except AttributeError:
493        import traceback
494        traceback.print_exc()
495        msg = ''
496
497    plain_indices = []
498    for left_span, right_span in iterate_chunks(msg.spans, 2, fillvalue=len(msg)):
499        left = left_span.end
500        right = right_span.start if not isinstance(right_span, int) else right_span
501        if left != right:
502            plain_indices.append((left, right))
503    if msg.spans:
504        if msg.spans[0].start != 0:
505            plain_indices = [(0, msg.spans[0].start)] + plain_indices
506        if plain_indices and msg.spans[-1].end != len(msg) and plain_indices[-1][1] != len(msg):
507            plain_indices.append((msg.spans[-1].end, len(msg)))
508
509    if plain_indices:
510        for left, right in plain_indices:
511            msg.stylize(style, left, right)
512    else:
513        msg = Text(str(msg), style)
514
515    return rich_text_to_str(msg)

277def format_success_tuple(
278    tup: mrsm.SuccessTuple,
279    upper_padding: int = 0,
280    lower_padding: int = 0,
281    left_padding: int = 1,
282    calm: bool = False,
283    _progress: Optional['rich.progress.Progress'] = None,
284) -> str:
285    """
286    Format `meerschaum.utils.typing.SuccessTuple`.
287
288    Parameters
289    ----------
290    upper_padding: int, default 0
291        How many newlines to prepend to the message.
292
293    lower_padding: int, default 0
294        How many newlines to append to the message.
295
296    left_padding: int, default 1
297        How mant spaces to preprend to the message.
298
299    calm: bool, default False
300        If `True`, use the default emoji and color scheme.
301    """
302    _init()
303    try:
304        status = 'success' if tup[0] else 'failure'
305    except TypeError:
306        status = 'failure'
307        tup = None, None
308
309    if calm:
310        status += '_calm'
311
312    ANSI, CHARSET = __getattr__('ANSI'), __getattr__('CHARSET')
313    from meerschaum.config import get_config
314    status_config = get_config('formatting', status, patch=True)
315
316    msg = (' ' * left_padding) + status_config[CHARSET]['icon'] + ' ' + str(highlight_pipes(tup[1]))
317    lines = msg.split('\n')
318    lines = [lines[0]] + [
319        (('    ' + line if not line.startswith(' ') else line))
320        for line in lines[1:]
321    ]
322    if ANSI:
323        lines[0] = fill_ansi(lines[0], **status_config['ansi']['rich'])
324
325    msg = '\n'.join(lines)
326    msg = ('\n' * upper_padding) + msg + ('\n' * lower_padding)
327    return msg

204def get_console():
205    """Get the rich console."""
206    global console
207    from meerschaum.utils.packages import import_rich, attempt_import
208    rich = import_rich()
209    rich_console = attempt_import('rich.console')
210    try:
211        console = rich_console.Console(force_terminal=True, color_system='truecolor')
212    except Exception:
213        import traceback
214        traceback.print_exc()
215        console = None
216    return console

345def highlight_pipes(message: str) -> str:
346    """
347    Add syntax highlighting to an info message containing stringified `meerschaum.Pipe` objects.
348    """
349    if 'Pipe(' not in message:
350        return message
351
352    from meerschaum.utils.misc import parse_arguments_str
353    segments = message.split('Pipe(')
354    msg = ''
355    for i, segment in enumerate(segments):
356        if i == 0:
357            msg += segment
358            continue
359
360        paren_index = segment.find(')')
361        if paren_index == -1:
362            msg += 'Pipe(' + segment
363            continue
364        
365        pipe_args_str = segment[:paren_index]
366        try:
367            args, kwargs = parse_arguments_str(pipe_args_str)
368            pipe_dict = {
369                'connector_keys': args[0],
370                'metric_key': args[1],
371            }
372            if len(args) > 2:
373                pipe_dict['location_key'] = args[2]
374            if 'instance' in kwargs:
375                pipe_dict['instance_keys'] = kwargs['instance']
376            
377            _to_add = pipe_repr(pipe_dict) + segment[paren_index + 1:]
378        except Exception:
379            _to_add = 'Pipe(' + segment
380        msg += _to_add
381    return msg

17def make_header(message: str, ruler: str = '─', left_pad: int = 2) -> str:
18    """Format a message string with a ruler.
19    Length of the ruler is the length of the longest word.
20    
21    Example:
22        'My\nheader' -> '  My\n  header\n  ──────'
23    """
24
25    from meerschaum.utils.formatting import ANSI, UNICODE, colored
26    if not UNICODE:
27        ruler = '-'
28    words = message.split('\n')
29    max_length = 0
30    for w in words:
31        length = len(w)
32        if length > max_length:
33            max_length = length
34
35    left_buffer = left_pad * ' '
36
37    return (
38        left_buffer
39        + message.replace('\n', '\n' + left_buffer)
40        + "\n"
41        + left_buffer
42        + (ruler * max_length)
43    )

280def pipe_repr(
281    pipe: Union[mrsm.Pipe, Dict[str, Any]],
282    as_rich_text: bool = False,
283    ansi: Optional[bool] = None,
284) -> Union[str, 'rich.text.Text']:
285    """
286    Return a formatted string for representing a `meerschaum.Pipe`.
287    """
288    from meerschaum.utils.formatting import ANSI, colored, rich_text_to_str
289    from meerschaum.utils.packages import import_rich, attempt_import
290    import meerschaum as mrsm
291
292    _ = import_rich()
293    Text = attempt_import('rich.text').Text
294
295    if isinstance(pipe, mrsm.Pipe):
296        connector_keys = pipe.connector_keys
297        metric_key = pipe.metric_key
298        location_key = pipe.location_key
299        instance_keys = pipe.instance_keys
300    else:
301        connector_keys = pipe.get('connector_keys')
302        metric_key = pipe.get('metric_key')
303        location_key = pipe.get('location_key')
304        instance_keys = pipe.get('instance_keys', get_config('meerschaum', 'instance'))
305
306    styles = get_config('formatting', 'pipes', '__repr__', 'ansi', 'styles')
307    if not ANSI or (ansi is False):
308        styles = {k: '' for k in styles}
309    _pipe_style_prefix, _pipe_style_suffix = (
310        (("[" + styles['Pipe'] + "]"), ("[/" + styles['Pipe'] + "]")) if styles['Pipe']
311        else ('', '')
312    )
313    text_obj = (
314        Text.from_markup(_pipe_style_prefix + "Pipe(" + _pipe_style_suffix)
315        + colored(("'" + connector_keys + "'"), style=styles['connector'], as_rich_text=True)
316        + Text.from_markup(_pipe_style_prefix + ", " + _pipe_style_suffix)
317        + colored(("'" + metric_key + "'"), style=styles['metric'], as_rich_text=True)
318        + (
319            (
320                colored(', ', style=styles['punctuation'], as_rich_text=True)
321                + colored(
322                    ("'" + location_key + "'"),
323                    style=styles['location'], as_rich_text=True
324                )
325            ) if location_key is not None
326            else colored('', style='', as_rich_text=True)
327        ) + (
328            ( ### Add the `instance=` argument.
329                colored(', instance=', style=styles['punctuation'], as_rich_text=True)
330                + colored(
331                    ("'" + instance_keys + "'"),
332                    style=styles['instance'], as_rich_text=True
333                )
334            ) if instance_keys != get_config('meerschaum', 'instance')
335            else colored('', style='', as_rich_text=True)
336        )
337        + Text.from_markup(_pipe_style_prefix + ")" + _pipe_style_suffix)
338    )
339    if as_rich_text:
340        return text_obj
341    return rich_text_to_str(text_obj).replace('\n', '')

 10def pprint(
 11    *args,
 12    detect_password: bool = True,
 13    nopretty: bool = False,
 14    **kw
 15) -> None:
 16    """Pretty print an object according to the configured ANSI and UNICODE settings.
 17    If detect_password is True (default), search and replace passwords with '*' characters.
 18    Does not mutate objects.
 19    """
 20    import copy
 21    import json
 22    from meerschaum.utils.packages import attempt_import, import_rich
 23    from meerschaum.utils.formatting import ANSI, get_console, print_tuple
 24    from meerschaum.utils.warnings import error
 25    from meerschaum.utils.misc import replace_password, dict_from_od, filter_keywords
 26    from collections import OrderedDict
 27
 28    if (
 29        len(args) == 1
 30        and
 31        isinstance(args[0], tuple)
 32        and
 33        len(args[0]) == 2
 34        and
 35        isinstance(args[0][0], bool)
 36        and
 37        isinstance(args[0][1], str)
 38    ):
 39        return print_tuple(args[0], **filter_keywords(print_tuple, **kw))
 40
 41    modify = True
 42    rich_pprint = None
 43    if ANSI and not nopretty:
 44        rich = import_rich()
 45        if rich is not None:
 46            rich_pretty = attempt_import('rich.pretty')
 47        if rich_pretty is not None:
 48            def _rich_pprint(*args, **kw):
 49                _console = get_console()
 50                _kw = filter_keywords(_console.print, **kw)
 51                _console.print(*args, **_kw)
 52            rich_pprint = _rich_pprint
 53    elif not nopretty:
 54        pprintpp = attempt_import('pprintpp', warn=False)
 55        try:
 56            _pprint = pprintpp.pprint
 57        except Exception :
 58            import pprint as _pprint_module
 59            _pprint = _pprint_module.pprint
 60
 61    func = (
 62        _pprint if rich_pprint is None else rich_pprint
 63    ) if not nopretty else print
 64
 65    try:
 66        args_copy = copy.deepcopy(args)
 67    except Exception:
 68        args_copy = args
 69        modify = False
 70
 71    _args = []
 72    for a in args:
 73        c = a
 74        ### convert OrderedDict into dict
 75        if isinstance(a, OrderedDict) or issubclass(type(a), OrderedDict):
 76            c = dict_from_od(copy.deepcopy(c))
 77        _args.append(c)
 78    args = _args
 79
 80    _args = list(args)
 81    if detect_password and modify:
 82        _args = []
 83        for a in args:
 84            c = a
 85            if isinstance(c, dict):
 86                c = replace_password(copy.deepcopy(c))
 87            if nopretty:
 88                try:
 89                    c = json.dumps(c)
 90                    is_json = True
 91                except Exception:
 92                    is_json = False
 93                if not is_json:
 94                    try:
 95                        c = str(c)
 96                    except Exception:
 97                        pass
 98            _args.append(c)
 99
100    ### filter out unsupported keywords
101    func_kw = filter_keywords(func, **kw) if not nopretty else {}
102    error_msg = None
103    try:
104        func(*_args, **func_kw)
105    except Exception as e:
106        error_msg = e
107    if error_msg is not None:
108        error(error_msg)

 17def pprint_pipes(pipes: PipesDict) -> None:
 18    """Print a stylized tree of a Pipes dictionary.
 19    Supports ANSI and UNICODE global settings."""
 20    from meerschaum.utils.warnings import error
 21    from meerschaum.utils.packages import attempt_import, import_rich
 22    from meerschaum.utils.misc import sorted_dict, replace_pipes_in_dict
 23    from meerschaum.utils.formatting import UNICODE, ANSI, CHARSET, pprint, colored, get_console
 24    import copy
 25    rich = import_rich('rich', warn=False)
 26    Text = None
 27    if rich is not None:
 28        rich_text = attempt_import('rich.text', lazy=False)
 29        Text = rich_text.Text
 30
 31    icons = get_config('formatting', 'pipes', CHARSET, 'icons')
 32    styles = get_config('formatting', 'pipes', 'ansi', 'styles')
 33    if not ANSI:
 34        styles = {k: '' for k in styles}
 35    print()
 36
 37    def ascii_print_pipes():
 38        """Print the dictionary with no unicode allowed. Also works in case rich fails to import
 39        (though rich should auto-install when `attempt_import()` is called)."""
 40        asciitree = attempt_import('asciitree')
 41        ascii_dict, replace_dict = {}, {'connector': {}, 'metric': {}, 'location': {}}
 42        for conn_keys, metrics in pipes.items():
 43            _colored_conn_key = colored(icons['connector'] + conn_keys, style=styles['connector'])
 44            if Text is not None:
 45                replace_dict['connector'][_colored_conn_key] = (
 46                    Text(conn_keys, style=styles['connector'])
 47                )
 48            ascii_dict[_colored_conn_key] = {}
 49            for metric, locations in metrics.items():
 50                _colored_metric_key = colored(icons['metric'] + metric, style=styles['metric'])
 51                if Text is not None:
 52                    replace_dict['metric'][_colored_metric_key] = (
 53                        Text(metric, style=styles['metric'])
 54                    )
 55                ascii_dict[_colored_conn_key][_colored_metric_key] = {}
 56                for location, pipe in locations.items():
 57                    _location_style = styles[('none' if location is None else 'location')]
 58                    pipe_addendum = '\n         ' + pipe.__repr__() + '\n'
 59                    _colored_location = colored(
 60                        icons['location'] + str(location), style=_location_style
 61                    )
 62                    _colored_location_key = _colored_location + pipe_addendum
 63                    if Text is not None:
 64                        replace_dict['location'][_colored_location] = (
 65                            Text(str(location), style=_location_style)
 66                        )
 67                    ascii_dict[_colored_conn_key][_colored_metric_key][_colored_location_key] = {}
 68
 69        tree = asciitree.LeftAligned()
 70        output = ''
 71        cols = []
 72
 73        ### This is pretty terrible, unreadable code.
 74        ### Please know that I'm normally better than this.
 75        key_str = (
 76            (Text("     ") if Text is not None else "     ") +
 77            (
 78                Text("Key", style='underline') if Text is not None else
 79                colored("Key", style='underline')
 80            ) + (Text('\n\n  ') if Text is not None else '\n\n  ') +
 81            (
 82                Text("Connector", style=styles['connector']) if Text is not None else
 83                colored("Connector", style=styles['connector'])
 84            ) + (Text('\n   +-- ') if Text is not None else '\n   +-- ') +
 85            (
 86                Text("Metric", style=styles['metric']) if Text is not None else
 87                colored("Metric", style=styles['metric'])
 88            ) + (Text('\n       +-- ') if Text is not None else '\n       +-- ') +
 89            (
 90                Text("Location", style=styles['location']) if Text is not None else
 91                colored("Location", style=styles['location'])
 92            ) + (Text('\n\n') if Text is not None else '\n\n')
 93        )
 94
 95        output += str(key_str)
 96        cols.append(key_str)
 97
 98        def replace_tree_text(tree_str : str) -> Text:
 99            """Replace the colored words with stylized Text instead.
100            Is not executed if ANSI and UNICODE are disabled."""
101            tree_text = Text(tree_str) if Text is not None else None
102            for k, v in replace_dict.items():
103                for _colored, _text in v.items():
104                    parts = []
105                    lines = tree_text.split(_colored)
106                    for part in lines:
107                        parts += [part, _text]
108                    if lines[-1] != Text(''):
109                        parts = parts[:-1]
110                    _tree_text = Text('')
111                    for part in parts:
112                        _tree_text += part
113                    tree_text = _tree_text
114            return tree_text
115
116        tree_output = ""
117        for k, v in ascii_dict.items():
118            branch = {k : v}
119            tree_output += tree(branch) + '\n\n'
120            if not UNICODE and not ANSI:
121                _col = (Text(tree(branch)) if Text is not None else tree(branch))
122            else:
123                _col = replace_tree_text(tree(branch))
124            cols.append(_col)
125        if len(output) > 0:
126            tree_output = tree_output[:-2]
127        output += tree_output
128
129        if rich is None:
130            return print(output)
131
132        rich_columns = attempt_import('rich.columns')
133        Columns = rich_columns.Columns
134        columns = Columns(cols)
135        get_console().print(columns)
136
137    if not UNICODE:
138        return ascii_print_pipes()
139
140    rich_panel, rich_tree, rich_text, rich_columns, rich_table = attempt_import(
141        'rich.panel',
142        'rich.tree',
143        'rich.text',
144        'rich.columns',
145        'rich.table',
146    )
147    from rich import box
148    Panel = rich_panel.Panel
149    Tree = rich_tree.Tree
150    Text = rich_text.Text
151    Columns = rich_columns.Columns
152    Table = rich_table.Table
153
154    key_panel = Panel(
155        (
156            Text("\n") +
157            Text(icons['connector'] + "Connector", style=styles['connector']) + Text("\n\n") +
158            Text(icons['metric'] + "Metric", style=styles['metric']) + Text("\n\n") +
159            Text(icons['location'] + "Location", style=styles['location']) + Text("\n")
160        ),
161        title = Text(icons['key'] + "Keys", style=styles['guide']),
162        border_style = styles['guide'],
163        expand = True
164    )
165
166    cols = []
167    conn_trees = {}
168    metric_trees = {}
169    pipes = sorted_dict(pipes)
170    for conn_keys, metrics in pipes.items():
171        conn_trees[conn_keys] = Tree(
172            Text(
173                icons['connector'] + conn_keys,
174                style = styles['connector'],
175            ),
176            guide_style = styles['connector']
177        )
178        metric_trees[conn_keys] = {}
179        for metric, locations in metrics.items():
180            metric_trees[conn_keys][metric] = Tree(
181                Text(
182                    icons['metric'] + metric,
183                    style = styles['metric']
184                ),
185                guide_style = styles['metric']
186            )
187            conn_trees[conn_keys].add(metric_trees[conn_keys][metric])
188            for location, pipe in locations.items():
189                _location = (
190                    Text(str(location), style=styles['none']) if location is None
191                    else Text(location, style=styles['location'])
192                )
193                _location = (
194                    Text(icons['location'])
195                    + _location + Text('\n')
196                    + pipe_repr(pipe, as_rich_text=True) + Text('\n')
197                )
198                metric_trees[conn_keys][metric].add(_location)
199
200    cols += [key_panel]
201    for k, t in conn_trees.items():
202        cols.append(t)
203
204    columns = Columns(cols)
205    get_console().print(columns)