glances.plugins.tailer.PluginModel._build_file_stat() - Code Metrics - Inspection of "Create a file tailer plugin" - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Pull Request — develop (#3071)

unknown

created 2025-01-05 02:38 UTC

PluginModel._build_file_stat() A

↳ Parent: glances.plugins.tailer

Complexity

Conditions

Size

Total Lines	34
Code Lines	21

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	3
eloc	21
nop	3
dl	0
loc	34
rs	9.376
c	0
b	0
f	0

# -*- coding: utf-8 -*-
#
# This file is part of Glances.
#
# SPDX-FileCopyrightText: 2024 <benjimons>
#
# SPDX-License-Identifier: LGPL-3.0-only
#
"""
Tailer plugin for Glances.

This plugin tails a file (given by the user), displaying:
- last modification time
- total line count
- last N lines
"""

import os
import time
import datetime

from glances.logger import logger
from glances.plugins.plugin.model import GlancesPluginModel
from glances.globals import bytes2human

# -----------------------------------------------------------------------------
# Globals
# -----------------------------------------------------------------------------

fields_description = {
    "filename": {
        "description": "Name of the file",
    },
    "file_size": {
        "description": "File size in bytes",
        "unit": "byte",
    },
    "last_modified": {
        "description": "Last modification time of the file",
    },
    "line_count": {
        "description": "Line count for the entire file",
        "unit": "lines",
    },
    "last_lines": {
        "description": "The last N lines of the file",
        # No specific unit, it's textual
    },
}

# If you need to store some metrics in the history, you can define them here:
items_history_list = [
    # Example: you could keep track of file size over time
    # {"name": "file_size", "description": "Size of the tailed file", "y_unit": "byte"},
]

# -----------------------------------------------------------------------------
# Plugin class
# -----------------------------------------------------------------------------

class PluginModel(GlancesPluginModel):
    """Tailer plugin main class.

    Attributes:
        self.stats (list): A list of dictionaries, each representing a file’s stats.
    """

    def __init__(self, args=None, config=None):
        """Initialize the plugin."""
        super().__init__(
            args=args,
            config=config,
            items_history_list=items_history_list,
            stats_init_value=[],
            fields_description=fields_description,
        )

        # We want to display the stat in the TUI
        self.display_curse = True

        # Optionally read from the config file [tail] section
        # e.g.:
        # [tail]
        # filename=/var/log/syslog
        # lines=10
        self.default_filename = config.get_value(self.plugin_name, 'filename', default='/var/log/syslog')
        self.default_lines = config.get_int_value(self.plugin_name, 'lines', default=10)

        # Force a first update
        self.update()
        self.refresh_timer.set(0)

    def get_key(self):
        """Return the key used in each stats dictionary."""
        # We'll use 'filename' as the key
        return 'filename'

    @GlancesPluginModel._check_decorator
    @GlancesPluginModel._log_result_decorator
    def update(self):
        """Update the plugin stats.

        Called automatically at each refresh. Must set self.stats.
        """
        if self.input_method == 'local':
            stats = self.update_local()
        else:
            stats = self.get_init_value()

        self.stats = stats
        return self.stats

    def update_local(self):
        """Collect and return stats for our plugin (tailing a file)."""
        stats = self.get_init_value()

        # In a real scenario, you might have the user pass these in
        # or read from the config. For demonstration, we’ll use the defaults.
        filename = self.default_filename
        num_lines = self.default_lines

        # Build a dictionary representing the file stats
        file_stat = self._build_file_stat(filename, num_lines)
        stats.append(file_stat)

        return stats

    def _build_file_stat(self, filename, num_lines):
        """Return a dictionary of stats for the given filename."""
        result = {
            "key": self.get_key(),
            "filename": filename,
            "file_size": 0,
            "last_modified": "",
            "line_count": 0,
            "last_lines": [],
        }

        if not os.path.isfile(filename):
            logger.debug(f"File not found: {filename}")
            return result

        try:
            # Last modification time
            mod_time = os.path.getmtime(filename)
            result["last_modified"] = datetime.datetime.fromtimestamp(mod_time).strftime('%Y-%m-%d %H:%M:%S')

            # File size
            result["file_size"] = os.path.getsize(filename)

            # Count lines, read last N lines
            line_count, last_lines = self._tail_file(filename, num_lines)
            result["line_count"] = line_count
            # Store the last lines as a single string or as a list.
            # For display convenience, we might store them as a list of strings.
            result["last_lines"] = last_lines

        except Exception as e:
            logger.debug(f"Error reading file {filename}: {e}")

        return result

    def _tail_file(self, filename, num_lines):
        """Return (total_line_count, list_of_last_N_lines)."""
        lines = []
        with open(filename, 'rb') as f:
            # If the file is huge, you might want a more efficient way to read
            # the last N lines rather than reading the entire file.
            # For simplicity, read all lines:
            content = f.read().splitlines()
            total_lines = len(content)
            # Extract the last num_lines lines
            last_lines = content[-num_lines:] if total_lines >= num_lines else content
            # Decode to str (assuming UTF-8) for each line
            last_lines_decoded = [line.decode('utf-8', errors='replace') for line in last_lines]

        return total_lines, last_lines_decoded

    def update_views(self):
        """Update stats views (optional).

        If you need to set decorations (alerts or color formatting),
        you can do it here.
        """
        super().update_views()

        # Example: if file_size is above a threshold, we could color it in TUI
        for stat_dict in self.get_raw():
            fsize = stat_dict.get("file_size", 0)
            # Example: decorate if file > 1GB
            if fsize > 1024 ** 3:
                self.views[stat_dict[self.get_key()]]["file_size"]["decoration"] = self.get_alert(
                    fsize, header='bigfile'
                )

    def msg_curse(self, args=None, max_width=None):
        """Return the dict (list of lines) to display in the TUI."""
        ret = []

        # If no stats or disabled, return empty
        if not self.stats or self.is_disabled():
            return ret

        if max_width:
            name_max_width = max_width - 20
        else:
            # No max_width defined
            logger.debug(f"No max_width defined for the {self.plugin_name} plugin, it will not be displayed.")
            return ret

        # Header
        ret.append(self.curse_add_line("FILE TAILER PLUGIN", "TITLE"))

        # Display the stats
        for stat in self.stats:
            filename = stat.get("filename", "N/A")
            file_size = stat.get("file_size", 0)
            line_count = stat.get("line_count", 0)
            last_modified = stat.get("last_modified", "")
            last_lines = stat.get("last_lines", [])

            # New line for each file
            ret.append(self.curse_new_line())

            # 1) Filename
            msg_filename = f"File: {filename}"
            ret.append(self.curse_add_line(msg_filename[:name_max_width], "NORMAL"))

            # 2) File size + last modified time
            msg_meta = (f"Size: {bytes2human(file_size)}, "
                        f"Last Modified: {last_modified}, "
                        f"Total Lines: {line_count}")
            ret.append(self.curse_new_line())
            ret.append(self.curse_add_line(msg_meta, "NORMAL"))

            # 3) Last N lines
            ret.append(self.curse_new_line())
            ret.append(self.curse_add_line("Last lines:", "NORMAL"))
            for line in last_lines:
                ret.append(self.curse_new_line())
                ret.append(self.curse_add_line(f"  {line}", "NORMAL"))

            ret.append(self.curse_new_line())

        return ret


1			# -- coding: utf-8 --
2			#
3			# This file is part of Glances.
4			#
5			# SPDX-FileCopyrightText: 2024 <benjimons>
6			#
7			# SPDX-License-Identifier: LGPL-3.0-only
8			#
9			"""
10			Tailer plugin for Glances.
11
12			This plugin tails a file (given by the user), displaying:
13			- last modification time
14			- total line count
15			- last N lines
16			"""
17
18			import os
19			import time
20			import datetime
21
22			from glances.logger import logger
23			from glances.plugins.plugin.model import GlancesPluginModel
24			from glances.globals import bytes2human
25
26			# -----------------------------------------------------------------------------
27			# Globals
28			# -----------------------------------------------------------------------------
29
30			fields_description = {
31			"filename": {
32			"description": "Name of the file",
33			},
34			"file_size": {
35			"description": "File size in bytes",
36			"unit": "byte",
37			},
38			"last_modified": {
39			"description": "Last modification time of the file",
40			},
41			"line_count": {
42			"description": "Line count for the entire file",
43			"unit": "lines",
44			},
45			"last_lines": {
46			"description": "The last N lines of the file",
47			# No specific unit, it's textual
48			},
49			}
50
51			# If you need to store some metrics in the history, you can define them here:
52			items_history_list = [
53			# Example: you could keep track of file size over time
54			# {"name": "file_size", "description": "Size of the tailed file", "y_unit": "byte"},
55			]
56
57			# -----------------------------------------------------------------------------
58			# Plugin class
59			# -----------------------------------------------------------------------------
60
61			class PluginModel(GlancesPluginModel):
62			"""Tailer plugin main class.
63
64			Attributes:
65			self.stats (list): A list of dictionaries, each representing a file’s stats.
66			"""
67
68			def __init__(self, args=None, config=None):
69			"""Initialize the plugin."""
70			super().__init__(
71			args=args,
72			config=config,
73			items_history_list=items_history_list,
74			stats_init_value=[],
75			fields_description=fields_description,
76			)
77
78			# We want to display the stat in the TUI
79			self.display_curse = True
80
81			# Optionally read from the config file [tail] section
82			# e.g.:
83			# [tail]
84			# filename=/var/log/syslog
85			# lines=10
86			self.default_filename = config.get_value(self.plugin_name, 'filename', default='/var/log/syslog')
87			self.default_lines = config.get_int_value(self.plugin_name, 'lines', default=10)
88
89			# Force a first update
90			self.update()
91			self.refresh_timer.set(0)
92
93			def get_key(self):
94			"""Return the key used in each stats dictionary."""
95			# We'll use 'filename' as the key
96			return 'filename'
97
98			@GlancesPluginModel._check_decorator
99			@GlancesPluginModel._log_result_decorator
100			def update(self):
101			"""Update the plugin stats.
102
103			Called automatically at each refresh. Must set self.stats.
104			"""
105			if self.input_method == 'local':
106			stats = self.update_local()
107			else:
108			stats = self.get_init_value()
109
110			self.stats = stats
111			return self.stats
112
113			def update_local(self):
114			"""Collect and return stats for our plugin (tailing a file)."""
115			stats = self.get_init_value()
116
117			# In a real scenario, you might have the user pass these in
118			# or read from the config. For demonstration, we’ll use the defaults.
119			filename = self.default_filename
120			num_lines = self.default_lines
121
122			# Build a dictionary representing the file stats
123			file_stat = self._build_file_stat(filename, num_lines)
124			stats.append(file_stat)
125
126			return stats
127
128			def _build_file_stat(self, filename, num_lines):
129			"""Return a dictionary of stats for the given filename."""
130			result = {
131			"key": self.get_key(),
132			"filename": filename,
133			"file_size": 0,
134			"last_modified": "",
135			"line_count": 0,
136			"last_lines": [],
137			}
138
139			if not os.path.isfile(filename):
140			logger.debug(f"File not found: {filename}")
141			return result
142
143			try:
144			# Last modification time
145			mod_time = os.path.getmtime(filename)
146			result["last_modified"] = datetime.datetime.fromtimestamp(mod_time).strftime('%Y-%m-%d %H:%M:%S')
147
148			# File size
149			result["file_size"] = os.path.getsize(filename)
150
151			# Count lines, read last N lines
152			line_count, last_lines = self._tail_file(filename, num_lines)
153			result["line_count"] = line_count
154			# Store the last lines as a single string or as a list.
155			# For display convenience, we might store them as a list of strings.
156			result["last_lines"] = last_lines
157
158			except Exception as e:
159			logger.debug(f"Error reading file {filename}: {e}")
160
161			return result
162
163			def _tail_file(self, filename, num_lines):
164			"""Return (total_line_count, list_of_last_N_lines)."""
165			lines = []
166			with open(filename, 'rb') as f:
167			# If the file is huge, you might want a more efficient way to read
168			# the last N lines rather than reading the entire file.
169			# For simplicity, read all lines:
170			content = f.read().splitlines()
171			total_lines = len(content)
172			# Extract the last num_lines lines
173			last_lines = content[-num_lines:] if total_lines >= num_lines else content
174			# Decode to str (assuming UTF-8) for each line
175			last_lines_decoded = [line.decode('utf-8', errors='replace') for line in last_lines]
176
177			return total_lines, last_lines_decoded
178
179			def update_views(self):
180			"""Update stats views (optional).
181
182			If you need to set decorations (alerts or color formatting),
183			you can do it here.
184			"""
185			super().update_views()
186
187			# Example: if file_size is above a threshold, we could color it in TUI
188			for stat_dict in self.get_raw():
189			fsize = stat_dict.get("file_size", 0)
190			# Example: decorate if file > 1GB
191			if fsize > 1024 ** 3:
192			self.views[stat_dict[self.get_key()]]["file_size"]["decoration"] = self.get_alert(
193			fsize, header='bigfile'
194			)
195
196			def msg_curse(self, args=None, max_width=None):
197			"""Return the dict (list of lines) to display in the TUI."""
198			ret = []
199
200			# If no stats or disabled, return empty
201			if not self.stats or self.is_disabled():
202			return ret
203
204			if max_width:
205			name_max_width = max_width - 20
206			else:
207			# No max_width defined
208			logger.debug(f"No max_width defined for the {self.plugin_name} plugin, it will not be displayed.")
209			return ret
210
211			# Header
212			ret.append(self.curse_add_line("FILE TAILER PLUGIN", "TITLE"))
213
214			# Display the stats
215			for stat in self.stats:
216			filename = stat.get("filename", "N/A")
217			file_size = stat.get("file_size", 0)
218			line_count = stat.get("line_count", 0)
219			last_modified = stat.get("last_modified", "")
220			last_lines = stat.get("last_lines", [])
221
222			# New line for each file
223			ret.append(self.curse_new_line())
224
225			# 1) Filename
226			msg_filename = f"File: {filename}"
227			ret.append(self.curse_add_line(msg_filename[:name_max_width], "NORMAL"))
228
229			# 2) File size + last modified time
230			msg_meta = (f"Size: {bytes2human(file_size)}, "
231			f"Last Modified: {last_modified}, "
232			f"Total Lines: {line_count}")
233			ret.append(self.curse_new_line())
234			ret.append(self.curse_add_line(msg_meta, "NORMAL"))
235
236			# 3) Last N lines
237			ret.append(self.curse_new_line())
238			ret.append(self.curse_add_line("Last lines:", "NORMAL"))
239			for line in last_lines:
240			ret.append(self.curse_new_line())
241			ret.append(self.curse_add_line(f" {line}", "NORMAL"))
242
243			ret.append(self.curse_new_line())
244
245			return ret
246

nicolargo / glances

Pull Request — develop (#3071)

PluginModel._build_file_stat() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like