file_instance.py

Go to the documentation of this file.

"""
# +==== BEGIN rotary_logger =================+
# LOGO:
# ..........####...####..........
# ......###.....#.#########......
# ....##........#.###########....
# ...#..........#.############...
# ...#..........#.#####.######...
# ..#.....##....#.###..#...####..
# .#.....#.##...#.##..##########.
# #.....##########....##...######
# #.....#...##..#.##..####.######
# .#...##....##.#.##..###..#####.
# ..#.##......#.#.####...######..
# ..#...........#.#############..
# ..#...........#.#############..
# ...##.........#.############...
# ......#.......#.#########......
# .......#......#.########.......
# .........#####...#####.........
# /STOP
# PROJECT: rotary_logger
# FILE: file_instance.py
# CREATION DATE: 30-10-2025
# LAST Modified: 10:44:15 27-03-2026
# DESCRIPTION:
# A module that provides a universal python light on iops way of logging to files your program execution.
# /STOP
# COPYRIGHT: (c) Asperguide
# PURPOSE: This is the file containing the class that is in charge of handling writing and rotating files regardless of the number of external processes calling it
# // AR
# +==== END rotary_logger =================+
"""
 
import sys
import os
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional, Union, Dict, List
from threading import RLock
from warnings import warn
 
try:
    from . import constants as CONST
    from .rogger import Rogger, RI
except ImportError:
    import constants as CONST
    from rogger import Rogger, RI
 
 
class FileInstance:
    """Manage buffered writes, file descriptors, and log rotation.
 
    Public methods are thread-safe and documented below. Writes are
    appended to an in-memory buffer and flushed to disk when the
    configured flush size is reached. Rotation is performed when the
    underlying file grows beyond `max_size`.
    """
 
    def __init__(
        self,
        file_path: Optional[Union[str, Path, CONST.FileInfo]],
        override: Optional[bool] = None,
        merged: Optional[bool] = None,
        encoding: Optional[str] = None,
        prefix: Optional[CONST.Prefix] = None,
        *,
        max_size_mb: Optional[int] = None,
        flush_size_kb: Optional[int] = None,
        folder_prefix: Optional[CONST.StdMode] = None,
        log_to_file: bool = True,
        merge_stdin: Optional[bool] = None,
    ) -> None:
        """Create a FileInstance wrapper.
 
        Arguments:
            file_path (Optional[Union[str, Path, CONST.FileInfo]]): Initial file path, Path or FileInfo, or None to defer opening.
 
        Keyword Arguments:
            override (Optional[bool]): When True open files for write ('w') instead of append ('a'). Default: None
            merged (Optional[bool]): Whether multiple streams should share the same file. Default: None
            encoding (Optional[str]): Text encoding to use for file I/O. Default: None
            prefix (Optional[CONST.Prefix]): Optional Prefix configuration to use when teeing. Default: None
            max_size_mb (Optional[int]): Maximum logfile size in MB before rotation. Default: None
            flush_size_kb (Optional[int]): Buffer flush threshold in KB. Default: None
            folder_prefix (Optional[CONST.StdMode]): StdMode used to segregate per-stream subfolders. Default: None
            log_to_file (bool): Whether file logging is enabled. Default: True
            merge_stdin (Optional[bool]): Whether stdin is merged into the shared log file. Default: None
        """
 
        # per-instance mutable defaults (avoid sharing across instances)
        self.rogger: Rogger = RI
        self._file_lock_file_lock_file_lock: RLock = RLock()
        self._mode: str = "a"
        self._log_to_file: bool = log_to_file
        self.filefile: Optional[CONST.FileInfo] = None
        self.override: bool = False
        self.merged: bool = True
        self.merge_stdin: bool = False
        self.encoding: str = CONST.DEFAULT_ENCODING
        self.prefix: Optional[CONST.Prefix] = None
        self.max_size: int = CONST.DEFAULT_LOG_MAX_FILE_SIZE
        self.flush_size: int = CONST.BUFFER_FLUSH_SIZE
        self.folder_prefix: Optional[CONST.StdMode] = None
 
        self._buffer: List[str] = []
        if override is not None:
            self.set_override(override)
        if merged is not None:
            self.set_merged(merged)
        if merge_stdin is not None:
            self.set_merge_stdin(merge_stdin)
        if encoding is not None:
            self.set_encoding(encoding)
        if prefix is not None:
            self.set_prefix(prefix)
        if max_size_mb is not None:
            self._set_max_size(max_size_mb)
        if flush_size_kb is not None:
            self._set_flush_size(flush_size_kb)
        if folder_prefix is not None:
            self.set_folder_prefix(folder_prefix)
        if file_path is not None:
            self.set_filepath(file_path)
        try:
            self.rogger.log_success(
                f"Initialized FileInstance with file_path={file_path}, override={override}, merged={merged}, encoding={encoding}, prefix={prefix}, max_size_mb={max_size_mb}, flush_size_kb={flush_size_kb}, folder_prefix={folder_prefix}, log_to_file={log_to_file}, merge_stdin={merge_stdin}",
                stream=CONST.RAW_STDOUT
            )
        except (AttributeError, OSError, ValueError):
            pass
 
    def __del__(self) -> None:
        """Best-effort cleanup on object deletion.
 
        Attempt to close the current file descriptor if it exists and is
        open. This method must never raise during interpreter shutdown
        (where `__del__` may be called), so IO-related errors are
        swallowed. After attempting to close the descriptor the internal
        `file` reference is cleared.
        """
        if self.filefile and self.filefile.descriptor:
            if not self.filefile.descriptor.closed:
                try:
                    self._close_file()
                except OSError:
                    # Swallow errors during destructor cleanup
                    pass
        self.filefile = None
 
    def set_log_to_file(self, log_to_file: bool = True, *, lock: bool = True) -> None:
        """Enable or disable file logging.
 
        When `log_to_file` is False no data will be written to disk even
        if a file descriptor is open.
 
        Arguments:
            log_to_file (bool): Whether writes to the log file are enabled. Default: True
 
        Keyword Arguments:
            lock (bool): When True the instance lock is acquired while updating the flag. Default: True
        """
        if lock:
            with self._file_lock_file_lock_file_lock:
                self._log_to_file: bool = log_to_file
                try:
                    self.rogger.log_info(
                        f"set_log_to_file -> {log_to_file}",
                        stream=CONST.RAW_STDOUT
                    )
                except (AttributeError, OSError, ValueError):
                    pass
                return
        self._log_to_file: bool = log_to_file
        try:
            self.rogger.log_info(
                f"set_log_to_file -> {log_to_file}",
                stream=CONST.RAW_STDOUT
            )
        except (AttributeError, OSError, ValueError):
            pass
 
    def set_max_size(self, max_size_mb: int, *, lock: bool = True) -> None:
        """Public wrapper to set the maximum logfile size.
 
        Delegates to `_set_max_size` which normalises the value to bytes
        and applies range checks.
 
        Arguments:
            max_size_mb (int): Maximum size in megabytes. The value is normalised by `_set_max_size` and stored in `self.max_size` as bytes.
 
        Keyword Arguments:
            lock (bool): When True the instance lock is acquired while updating the configuration. Default: True
        """
        if lock:
            with self._file_lock_file_lock_file_lock:
                self._set_max_size(max_size_mb)
                try:
                    self.rogger.log_debug(
                        f"set_max_size -> {max_size_mb}MB",
                        stream=CONST.RAW_STDOUT
                    )
                except (AttributeError, OSError, ValueError):
                    pass
                return
        self._set_max_size(max_size_mb)
        try:
            self.rogger.log_debug(
                f"set_max_size -> {max_size_mb}MB",
                stream=CONST.RAW_STDOUT
            )
        except (AttributeError, OSError, ValueError):
            pass
 
    def set_folder_prefix(self, folder_prefix: Optional[CONST.StdMode], *, lock: bool = True) -> None:
        """Public setter for `folder_prefix`.
 
        When `lock` is True the instance lock is held while the value is
        updated. The internal method `_set_folder_prefix` performs the
        validation and assignment.
        """
        if lock:
            with self._file_lock_file_lock_file_lock:
                self._set_folder_prefix(folder_prefix)
                return
        self._set_folder_prefix(folder_prefix)
 
    def set_flush_size(self, flush_size: int, *, lock: bool = True) -> None:
        """Public wrapper to configure the buffer flush threshold.
 
        `flush_size` is provided as a KB-like value and normalised by
        `_set_flush_size`. If `lock` is True the operation is
        performed while holding the instance lock.
        """
        if lock:
            with self._file_lock_file_lock_file_lock:
                self._set_flush_size(flush_size)
                try:
                    self.rogger.log_debug(
                        f"set_flush_size -> {flush_size}KB",
                        stream=CONST.RAW_STDOUT
                    )
                except (AttributeError, OSError, ValueError):
                    pass
                return
        self._set_flush_size(flush_size)
        try:
            self.rogger.log_debug(
                f"set_flush_size -> {flush_size}KB",
                stream=CONST.RAW_STDOUT
            )
        except (AttributeError, OSError, ValueError):
            pass
 
    def set_merged(self, merged: bool, *, lock: bool = True) -> None:
        """Enable or disable stream merging.
 
        When `merged` is True multiple streams may share a single log
        file; when False separate per-stream files are used. The
        `lock` parameter controls whether the instance lock is
        acquired.
        """
        if lock:
            with self._file_lock_file_lock_file_lock:
                self.merged = bool(merged)
                try:
                    self.rogger.log_info(
                        f"set_merged -> {bool(merged)}",
                        stream=CONST.RAW_STDOUT
                    )
                except (AttributeError, OSError, ValueError):
                    pass
                return
        self.merged = bool(merged)
        try:
            self.rogger.log_info(
                f"set_merged -> {bool(merged)}",
                stream=CONST.RAW_STDOUT
            )
        except (AttributeError, OSError, ValueError):
            pass
 
    def set_merge_stdin(self, merge_stdin: bool, *, lock: bool = True) -> None:
        """Enable or disable stdin merging into the shared log file.
 
        When `merge_stdin` is True stdin data is written to the same file
        as stdout/stderr; when False stdin is kept separate or not logged.
        The `lock` parameter controls whether the instance lock is acquired.
 
        Arguments:
            merge_stdin (bool): Whether stdin should be merged into the shared log file.
 
        Keyword Arguments:
            lock (bool): When True the instance lock is acquired while updating the flag. Default: True
        """
        if lock:
            with self._file_lock_file_lock_file_lock:
                self.merge_stdin = bool(merge_stdin)
                try:
                    self.rogger.log_info(
                        f"set_merge_stdin -> {bool(merge_stdin)}",
                        stream=CONST.RAW_STDOUT
                    )
                except (AttributeError, OSError, ValueError):
                    pass
                return
        self.merge_stdin = bool(merge_stdin)
        try:
            self.rogger.log_info(
                f"set_merge_stdin -> {bool(merge_stdin)}",
                stream=CONST.RAW_STDOUT
            )
        except (AttributeError, OSError, ValueError):
            pass
 
    def set_encoding(self, encoding: str, *, lock: bool = True) -> None:
        """Set the text encoding used for file I/O.
 
        Arguments:
            encoding (str): A codec name such as 'utf-8'.
 
        Keyword Arguments:
            lock (bool): When True the change is performed while holding the instance lock. Default: True
        """
        if lock:
            with self._file_lock_file_lock_file_lock:
                self.encoding = encoding
                try:
                    self.rogger.log_info(
                        f"set_encoding -> {encoding}",
                        stream=CONST.RAW_STDOUT
                    )
                except (AttributeError, OSError, ValueError):
                    pass
                return
        self.encoding = encoding
        try:
            self.rogger.log_info(
                f"set_encoding -> {encoding}",
                stream=CONST.RAW_STDOUT
            )
        except (AttributeError, OSError, ValueError):
            pass
 
    def set_prefix(self, prefix: Optional[CONST.Prefix], *, lock: bool = True) -> None:
        """Public setter for `Prefix` configuration.
 
        Copies the provided `prefix` into an internal `CONST.Prefix`
        object. Use `get_prefix()` to obtain a safe copy of the
        current configuration.
        """
        if lock:
            with self._file_lock_file_lock_file_lock:
                self._set_prefix(prefix)
                try:
                    self.rogger.log_debug(
                        f"set_prefix -> {prefix}",
                        stream=CONST.RAW_STDOUT
                    )
                except (AttributeError, OSError, ValueError):
                    pass
                return
        self._set_prefix(prefix)
        try:
            self.rogger.log_debug(
                f"set_prefix -> {prefix}",
                stream=CONST.RAW_STDOUT
            )
        except (AttributeError, OSError, ValueError):
            pass
        return
 
    def set_override(self, override: bool = False, *, lock: bool = True) -> None:
        """Public setter for the file open mode.
 
        When `override` is True files will be opened with mode 'w'
        (overwrite); otherwise 'a' (append) is used. The lock
        behaviour is controlled by `lock`.
        """
        _value: Dict[bool, str] = {
            True: "w",
            False: "a"
        }
        if lock:
            with self._file_lock_file_lock_file_lock:
                self._set_mode(_value[bool(override)], lock=False)
                try:
                    self.rogger.log_info(
                        f"set_override -> {bool(override)}",
                        stream=CONST.RAW_STDOUT
                    )
                except (AttributeError, OSError, ValueError):
                    pass
                return
        self._set_mode(_value[bool(override)], lock=False)
        try:
            self.rogger.log_info(
                f"set_override -> {bool(override)}",
                stream=CONST.RAW_STDOUT
            )
        except (AttributeError, OSError, ValueError):
            pass
 
    def set_filepath(self, file_path: Optional[Union[str, Path, CONST.FileInfo]], *, lock: bool = True) -> None:
        """Set or clear the active file/file path for this instance.
 
        Arguments:
            file_path (Optional[Union[str, Path, CONST.FileInfo]]): A path-like object, a `CONST.FileInfo` instance describing an already-open file, or None to clear the current file.
 
        Keyword Arguments:
            lock (bool): When True the instance lock is held while the change is applied. Default: True
        """
        if not file_path:
            if lock:
                with self._file_lock_file_lock_file_lock:
                    self._close_file(lock=False)
                    self.filefile = None
                    try:
                        self.rogger.log_info(
                            "set_filepath -> cleared file_path",
                            stream=CONST.RAW_STDOUT
                        )
                    except (AttributeError, OSError, ValueError):
                        pass
                    return
            self._close_file(lock=False)
            self.filefile = None
            try:
                self.rogger.log_info(
                    "set_filepath -> cleared file_path",
                    stream=CONST.RAW_STDOUT
                )
            except (AttributeError, OSError, ValueError):
                pass
            return
        if lock:
            with self._file_lock_file_lock_file_lock:
                self._set_filepath_child(file_path)
                try:
                    self.rogger.log_info(
                        f"set_filepath -> set to {file_path}",
                        stream=CONST.RAW_STDOUT
                    )
                except (AttributeError, OSError, ValueError):
                    pass
                return
        self._set_filepath_child(file_path)
        try:
            self.rogger.log_info(
                f"set_filepath -> set to {file_path}",
                stream=CONST.RAW_STDOUT
            )
        except (AttributeError, OSError, ValueError):
            pass
 
    def get_log_to_file(self, *, lock: bool = True) -> bool:
        """Return True when file logging is enabled.
 
        Keyword Arguments:
            lock (bool): When True the instance lock is acquired before reading the flag. Default: True
 
        Returns:
            True if writes to the log file are enabled, False otherwise.
        """
        if lock:
            with self._file_lock_file_lock_file_lock:
                return self._log_to_file
        return self._log_to_file
 
    def get_mode(self, *, lock: bool = True) -> str:
        """Return the current file open mode ('w' or 'a').
 
        When `lock` is True the instance lock is acquired prior to
        reading the value.
        """
        if lock:
            with self._file_lock_file_lock_file_lock:
                return self._mode
        return self._mode
 
    def get_merged(self, *, lock: bool = True) -> bool:
        """Return the merged flag (True when streams share a file).
 
        The optional `lock` parameter controls whether the instance
        lock is held while reading the value.
        """
        if lock:
            with self._file_lock_file_lock_file_lock:
                return self.merged
        return self.merged
 
    def get_merge_stdin(self, *, lock: bool = True) -> bool:
        """Return the merge_stdin flag (True when stdin is merged into the shared log file).
 
        Keyword Arguments:
            lock (bool): When True the instance lock is held while reading the value. Default: True
 
        Returns:
            True if stdin is merged into the shared log file, False otherwise.
        """
        if lock:
            with self._file_lock_file_lock_file_lock:
                return self.merge_stdin
        return self.merge_stdin
 
    def get_encoding(self, *, lock: bool = True) -> str:
        """Return the configured text encoding for file writes.
 
        When `lock` is True the instance lock is held while the value
        is read.
        """
        if lock:
            with self._file_lock_file_lock_file_lock:
                return self.encoding
        return self.encoding
 
    def get_prefix(self, *, lock: bool = True) -> Optional[CONST.Prefix]:
        """Return a safe copy of the current `Prefix` configuration.
 
        The returned `CONST.Prefix` is a fresh object to avoid exposing
        internal references. If no prefix is configured None is
        returned. When `lock` is True the instance lock is held while
        copying the object.
        """
        if self.prefix:
            if lock:
                with self._file_lock_file_lock_file_lock:
                    _prefix = CONST.Prefix()
                    _prefix.std_in = self.prefix.std_in
                    _prefix.std_out = self.prefix.std_out
                    _prefix.std_err = self.prefix.std_err
                    return _prefix
            _prefix = CONST.Prefix()
            _prefix.std_in = self.prefix.std_in
            _prefix.std_out = self.prefix.std_out
            _prefix.std_err = self.prefix.std_err
            return _prefix
        return None
 
    def get_override(self, *, lock: bool = True) -> bool:
        """Return True when override mode ('w') is active.
 
        This convenience maps internal mode strings to a boolean.
        When `lock` is True the instance lock is held for the check.
        """
        _value: Dict[str, bool] = {
            "w": True,
            "a": False
        }
        if lock:
            with self._file_lock_file_lock_file_lock:
                mode: str = self.get_mode(lock=False).lower()
                if mode in _value:
                    return _value[mode]
                self.rogger.log_error(
                    "Unsupported mode",
                    stream=CONST.RAW_STDERR
                )
 
                raise ValueError("Unsupported mode")
        mode: str = self.get_mode(lock=False).lower()
        if mode in _value:
            return _value[mode]
        self.rogger.log_error(
            "Unsupported mode",
            stream=CONST.RAW_STDERR
        )
        raise ValueError("Unsupported mode")
 
    def get_filepath(self, *, lock: bool = True) -> Optional[CONST.FileInfo]:
        """Return the internal `FileInfo` reference (may be None).
 
        Note: the returned object may be shared; callers that need to
        mutate it should take care to hold the instance lock or use
        `copy()` to obtain an independent view.
        """
        if lock:
            with self._file_lock_file_lock_file_lock:
                return self.filefile
        return self.filefile
 
    def get_flush_size(self, *, lock: bool = True) -> int:
        """Return the configured buffer flush threshold in bytes."""
        if lock:
            with self._file_lock_file_lock_file_lock:
                return self.flush_size
        return self.flush_size
 
    def get_max_size(self, *, lock: bool = True) -> int:
        """Return the configured maximum file size in bytes."""
        if lock:
            with self._file_lock_file_lock_file_lock:
                return self.max_size
        return self.max_size
 
    def get_folder_prefix(self, *, lock: bool = True) -> Optional[CONST.StdMode]:
        """Return the configured folder prefix (StdMode) or None."""
        if lock:
            with self._file_lock_file_lock_file_lock:
                return self.folder_prefix
        return self.folder_prefix
 
    def update(self, file_data: Optional['FileInstance'], *, lock: bool = True) -> None:
        """Public method to copy configuration from another instance.
 
        This method acquires the lock by default and delegates to the
        `_update` implementation which performs the actual field
        assignments.
        """
        if lock:
            with self._file_lock_file_lock_file_lock:
                self._update(file_data)
                return
        self._update(file_data)
 
    def copy(self, *, lock: bool = True) -> "FileInstance":
        """Return a shallow copy of this instance's configuration.
 
        The returned `FileInstance` is a new object populated from the
        current instance. By default the instance lock is acquired to
        provide a consistent snapshot.
        """
        if lock:
            with self._file_lock_file_lock_file_lock:
                return self._copy()
        return self._copy()
 
    def write(self, message: str) -> None:
        """Append `message` to the internal buffer (thread-safe).
 
        The message is encoded and counted toward `flush_size`. When the
        buffer reaches the configured threshold a background flush is
        triggered (performed synchronously inside `_flush_buffer()` but
        with I/O outside the main lock to minimize blocking).
        """
        # append under lock, then decide whether to flush
        with self._file_lock_file_lock_file_lock:
            self._buffer.append(message)
            should = self._should_flush()
        try:
            self.rogger.log_debug(
                f"write: appended {len(message)} chars, should_flush={should}",
                stream=CONST.RAW_STDOUT
            )
        except (AttributeError, OSError, ValueError):
            pass
        if should:
            self.rogger.log_debug(
                "Flushing buffer",
                stream=CONST.RAW_STDOUT
            )
            self._flush_buffer()
 
    def flush(self):
        """Flush any buffered log lines to disk immediately.
 
        This is a blocking call that performs disk I/O; callers should
        avoid calling it too frequently. Errors raised by the underlying
        I/O are propagated as OSError or ValueError when appropriate.
        """
        try:
            self.rogger.log_debug(
                "flush: manual flush requested",
                stream=CONST.RAW_STDOUT
            )
        except (AttributeError, OSError, ValueError):
            pass
        self._flush_buffer()
 
    def _set_prefix(self, prefix: Optional[CONST.Prefix]) -> None:
        """Set the internal `Prefix` object from an external one.
 
        This internal setter copies boolean flag values from `prefix`
        into a fresh `CONST.Prefix()` instance. The caller is
        responsible for holding any required locks; this routine does
        not perform locking itself.
        """
        if not prefix:
            self.prefix = None
            self.rogger.log_debug(
                "Prefixes set to None",
                stream=CONST.RAW_STDOUT
            )
            return
        # internal setter assumes caller handles locking
        self.rogger.log_debug(
            "Setting prefixes",
            stream=CONST.RAW_STDOUT
        )
        self.prefix = CONST.Prefix()
        self.prefix.std_in = prefix.std_in
        self.prefix.std_out = prefix.std_out
        self.prefix.std_err = prefix.std_err
        self.rogger.log_debug(
            f"Prefixes set: {self.prefix}",
            stream=CONST.RAW_STDOUT
        )
 
    def _set_folder_prefix(self, folder_prefix: Optional[CONST.StdMode]) -> None:
        """Configure the per-stream folder prefix.
 
        If `folder_prefix` is a valid `CONST.StdMode` value present in
        `CONST.CORRECT_FOLDER`, it is stored; otherwise the stored
        `folder_prefix` is cleared (set to None).
        """
        if folder_prefix and folder_prefix in CONST.CORRECT_FOLDER:
            self.folder_prefix = folder_prefix
            self.rogger.log_debug(
                f"Folder prefix: {self.folder_prefix}",
                stream=CONST.RAW_STDOUT
            )
            return
        self.folder_prefix = None
 
    def _set_mode(self, mode: str, *, lock: bool = True) -> None:
        """Set the file mode to 'w' (overwrite) or 'a' (append).
 
        The `lock` parameter indicates whether this function should
        acquire `self._file_lock` before updating the internal mode.
        Invalid input is ignored.
        """
        if lock:
            with self._file_lock_file_lock_file_lock:
                if mode in ("w", "a"):
                    self._mode = mode
                return
        if mode in ("w", "a"):
            self._mode = mode
 
    def _set_max_size(self, max_size_mb: int) -> None:
        """Configure the maximum file size used for rotation.
 
        `max_size_mb` is interpreted as megabytes when reasonable; the
        function attempts to coerce the input to an integer. Negative
        or too-small values are corrected with warnings. The resulting
        internal `self.max_size` is stored in bytes.
        """
        # Treat parameter as a count of megabytes (MB)
        try:
            _resp: int = int(max_size_mb)
        except (ValueError, TypeError):
            _resp = CONST.DEFAULT_LOG_MAX_FILE_SIZE
        if _resp < 0:
            warn("Max provided size cannot be negative, converting to positive")
            _resp = abs(_resp)
        if _resp < 1:
            warn("Max provided size is smaller than 1 MB, using default max file size")
            _resp = CONST.DEFAULT_LOG_MAX_FILE_SIZE
        # Convert MB to bytes (if the provided number looks like MB)
        if _resp < CONST.MB1:
            self.max_size = _resp * CONST.MB1
        else:
            self.max_size = _resp
 
    def _set_flush_size(self, flush_size_kb: int) -> None:
        """Configure the flush threshold for buffered writes.
 
        `flush_size_kb` is interpreted as kilobytes when reasonable; the
        function coerces input to int, normalises negative values and
        ensures the internal `self.flush_size` is stored in bytes.
        """
        # Treat parameter as a count of kilobytes (KB)
        try:
            _resp = int(flush_size_kb)
        except (ValueError, TypeError):
            _resp = CONST.DEFAULT_LOG_BUFFER_FLUSH_SIZE
        if _resp < 0:
            warn("Flush size cannot be negative, converting to positive")
            _resp = abs(_resp)
        if _resp < 1:
            warn("Flush size is smaller than 1 KB, using default flush size")
            _resp = CONST.DEFAULT_LOG_BUFFER_FLUSH_SIZE
        # Convert KB to bytes
        if _resp < CONST.KB1:
            self.flush_size = _resp * CONST.KB1
        else:
            self.flush_size = _resp
 
    def _set_filepath_child(self, file_path: Union[str, Path, CONST.FileInfo]) -> None:
        """Internal routine to set the instance's file reference.
 
        This method closes any previously-open file, clears internal
        state, and opens the provided `file_path`. The `file_path` may
        be a string/Path (in which case a new `FileInfo` is created)
        or an existing `CONST.FileInfo` instance which may be re-opened
        if necessary.
        """
        self._close_file(lock=False)
        # ensure the internal reference is cleared
        self.filefile = None
        if isinstance(file_path, (str, Path)):
            _path = Path(file_path)
            self.filefile = self._open_file(_path)
        elif isinstance(file_path, CONST.FileInfo):
            self.filefile = file_path
            if self.filefile:
                if self.filefile.path and not self.filefile.descriptor:
                    self.filefile = self._open_file(self.filefile.path)
 
    def _update(self, file_data: Optional['FileInstance']) -> None:
        """Copy configuration values from another FileInstance.
 
        This helper updates the receiver to match the provided
        `file_data`. The method is intended to be called while holding
        the caller's lock; it delegates to public setters with the
        `lock=False` flag to avoid deadlocks.
        """
        if not file_data:
            return
        self.set_filepath(file_data.get_filepath(), lock=False)
        self.set_override(file_data.get_override(), lock=False)
        self.set_merged(file_data.get_merged(), lock=False)
        self.set_encoding(file_data.get_encoding(), lock=False)
        self.set_prefix(file_data.get_prefix(), lock=False)
        self.set_max_size(file_data.get_max_size(), lock=False)
        self.set_flush_size(file_data.get_flush_size(), lock=False)
        self.set_folder_prefix(file_data.get_folder_prefix(), lock=False)
        self.set_log_to_file(file_data.get_log_to_file(), lock=False)
        self.set_merge_stdin(file_data.get_merge_stdin(), lock=False)
 
    def _copy(self) -> "FileInstance":
        """Return a shallow copy of this FileInstance configuration.
 
        The returned `FileInstance` will share the same `FileInfo`
        reference but will otherwise have the same configuration
        values (mode, encoding, prefix, sizes). The copy is useful for
        creating per-stream views of shared configuration.
        """
        tmp = FileInstance(None)
        tmp.set_filepath(self.get_filepath(), lock=False)
        tmp.set_override(self.get_override(), lock=False)
        tmp.set_merged(self.get_merged(), lock=False)
        tmp.set_encoding(self.get_encoding(), lock=False)
        tmp.set_prefix(self.get_prefix(), lock=False)
        tmp.set_flush_size(self.get_flush_size(), lock=False)
        tmp.set_max_size(self.get_max_size(), lock=False)
        tmp.set_folder_prefix(self.get_folder_prefix(), lock=False)
        tmp.set_log_to_file(self.get_log_to_file(), lock=False)
        tmp.set_merge_stdin(self.get_merge_stdin(), lock=False)
        return tmp
 
    def _get_current_date(self) -> datetime:
        """Return the current UTC datetime used for naming files.
 
        Centralising the time provider makes it easier to test
        filename generation and ensures all timestamps use UTC.
        """
        return datetime.now(timezone.utc)
 
    def _get_filename(self) -> str:
        """Construct a timestamped log filename.
 
        The filename format is driven by `CONST.FILE_LOG_DATE_FORMAT` and
        uses the current UTC time returned by `_get_current_date()`.
        """
        return self._get_current_date().strftime(f"{CONST.FILE_LOG_DATE_FORMAT}.log")
 
    def _should_flush(self) -> bool:
        """Check whether the in-memory buffer has reached the flush threshold.
 
        Returns:
            True if the total encoded size of buffered lines meets or exceeds `flush_size`, False otherwise.
        """
        _tmp: int = 0
        for line in self._buffer:
            _tmp += len(line.encode(self.encoding))
        return _tmp >= self.flush_size
 
    def _refresh_written_bytes(self) -> None:
        """Add the sizes of buffered lines to `file.written_bytes`.
 
        This method is called after a successful write to update the
        persisted byte counter. It encodes lines using the configured
        encoding and falls back to 'utf-8' on lookup errors. The in-
        memory buffer is cleared after accounting.
        """
        if not self.filefile:
            return
        for line in self._buffer:
            try:
                self.filefile.written_bytes += len(line.encode(self.encoding))
            except LookupError:
                self.filefile.written_bytes += len(line.encode('utf-8'))
        self._buffer.clear()
 
    def _should_rotate(self) -> bool:
        """Check whether the current log file has exceeded the maximum size threshold.
 
        Returns:
            True if the file's `written_bytes` exceeds `max_size` or no file is open, False otherwise.
        """
        if self.filefile:
            return self.filefile.written_bytes > self.max_size
        return True
 
    def _rotate_file(self) -> None:
        """Rotate the current file if the bytes threshold is exceeded.
 
        When rotation is needed the current descriptor is closed and a
        fresh `FileInfo` is opened at a newly-created path returned by
        `_create_log_path()`.
        """
        if self._should_rotate():
            self._close_file()
            log_path: Path = self._create_log_path()
            self.filefile = self._open_file(log_path)
 
    def _create_log_path(self, base_override: Optional[Path] = None) -> Path:
        """Build the timestamped log file path and create the parent directories.
 
        The path is organised as `<root>/logs/<year>/<month>/<day>/[<stream>/]<timestamp>.log`.
        If `base_override` is provided it is used as the root; otherwise the instance's
        current file path or the package directory is used as the fallback.
 
        Keyword Arguments:
            base_override (Optional[Path]): Override root directory for the log path. Default: None
 
        Returns:
            The resolved Path for the new log file.
        """
        # If a base_override is provided (for example when opening from
        # `_open_file` with a user-supplied directory), prefer it. Otherwise
        # fall back to the instance-configured path or package default.
        base = None
        now = self._get_current_date()
        if base_override is not None:
            _root: Path = base_override
        else:
            if self.filefile and isinstance(self.filefile.path, Path):
                candidate = self.filefile.path
                if candidate.suffix == '.log':
                    candidate.parent.mkdir(parents=True, exist_ok=True)
                    return candidate
                base = candidate
 
            now = self._get_current_date()
            _root: Path = Path(__file__).parent
            if base is not None:
                _root = base
            elif self.filefile and self.filefile.path:
                _root = self.filefile.path
 
        if _root.suffix == "" and CONST.LOG_FOLDER_BASE_NAME != _root.name:
            _root = _root / CONST.LOG_FOLDER_BASE_NAME
        elif _root.suffix != "" and CONST.LOG_FOLDER_BASE_NAME != _root.parent:
            _root = _root.parent / CONST.LOG_FOLDER_BASE_NAME / _root.name
 
        year_dir = _root / str(now.year)
        month_dir = year_dir / f"{now.month:02d}"
        day_dir = month_dir / f"{now.day:02d}"
        if self.folder_prefix and self.folder_prefix in CONST.CORRECT_FOLDER:
            day_dir = day_dir / CONST.CORRECT_FOLDER[self.folder_prefix]
 
        # Snapshot the _log_to_file flag under lock, then perform mkdir
        # outside the lock to avoid blocking other callers on filesystem I/O.
        _should_create = False
        with self._file_lock_file_lock_file_lock:
            _should_create = bool(self._log_to_file)
 
        if _should_create:
            day_dir.mkdir(parents=True, exist_ok=True)
        filename = self._get_filename()
        self.rogger.log_debug(
            f"Determined file name: {filename}, determined file path: {day_dir}",
            stream=CONST.RAW_STDOUT
        )
        return day_dir / filename
 
    def _looks_like_directory(self, dir_path: Path) -> bool:
        """Heuristic check to decide whether a path refers to a directory.
 
        Returns True when the path has no file extension, already exists as a
        directory, or ends with a path separator. Used to determine whether a
        caller-supplied path should be treated as a folder (triggering
        automatic log-file naming) or as an explicit file path.
 
        Arguments:
            dir_path (Path): Path to evaluate.
 
        Returns:
            True if the path appears to represent a directory, False otherwise.
        """
        if not str(dir_path):
            return False
        if dir_path.suffix == "":
            return True
        # If it already exists and is a directory, fine
        if dir_path.exists():
            return dir_path.is_dir()
 
        # If the parent exists and is writable, assume it *could* be a directory
        str_path: str = str(dir_path)
        looks_like_it = str_path.endswith(os.sep) or str_path.endswith(
            "/") or str_path.endswith("\\")
        self.rogger.log_debug(
            f"Looks_like_it: {looks_like_it}",
            stream=CONST.RAW_STDOUT
        )
        return looks_like_it
 
    def _open_file(self, file_path: Path) -> CONST.FileInfo:
        """Open or create the log file at `file_path` and return a populated `FileInfo`.
 
        If `file_path` looks like a directory (as determined by `_looks_like_directory`)
        a timestamped filename is generated via `_create_log_path`. The parent directory
        is created if it does not exist. The file descriptor is opened outside the
        instance lock; the resulting `FileInfo` is populated atomically under the lock.
 
        Arguments:
            file_path (Path): Destination path for the log file, or a directory.
 
        Returns:
            A `CONST.FileInfo` with `path`, `descriptor`, and `written_bytes` populated.
        """
        _node: CONST.FileInfo = CONST.FileInfo()
        _node.path = Path(file_path)
        if self._looks_like_directory(_node.path):
            _node.path = self._create_log_path(base_override=_node.path)
        # Ensure parent directory exists before attempting to open the file.
        _node.path.parent.mkdir(parents=True, exist_ok=True)
 
        # Snapshot whether we should open the descriptor under lock, then
        # perform the potentially blocking open() outside the lock.
        should_open = False
        with self._file_lock_file_lock_file_lock:
            should_open = bool(self._log_to_file)
 
        descriptor = None
        if should_open:
            self.rogger.log_debug(
                "File is not open, attempting to open",
                stream=CONST.RAW_STDOUT
            )
            try:
                descriptor = open(
                    _node.path,
                    self._mode,
                    encoding=self.encoding,
                    newline="\n"
                )
            except (OSError, ValueError):
                self.rogger.log_error(
                    f"Failed to open file: {_node.path}",
                    stream=CONST.RAW_STDERR
                )
                # Opening failed; keep descriptor as None. Caller will handle.
                descriptor = None
 
        # Assign descriptor under the lock to keep state updates atomic.
        with self._file_lock_file_lock_file_lock:
            _node.descriptor = descriptor
        if file_path.exists():
            _node.written_bytes = file_path.stat().st_size
        else:
            _node.written_bytes = 0
        self.rogger.log_debug(
            f"Written bytes: {_node.written_bytes}",
            stream=CONST.RAW_STDOUT
        )
        return _node
 
    def _close_file_inner(self) -> bool:
        """Close the underlying file descriptor without acquiring locks.
 
        This inner helper is used by `_close_file()`; it performs the
        actual descriptor close and clears the reference. It is safe to
        call when already closed. Errors during close are suppressed
        because this is a best-effort operation.
        """
        if self.filefile:
            self.rogger.log_debug(
                "File is open, closing",
                stream=CONST.RAW_STDOUT
            )
            descriptor = getattr(self.filefile, "descriptor", None)
            if descriptor:
                try:
                    descriptor.close()
                except (OSError, ValueError):
                    # ignore errors closing the descriptor (IO and value errors)
                    pass
                # clear descriptor reference so subsequent checks see it's closed
                try:
                    self.filefile.descriptor = None
                except AttributeError:
                    # unlikely, but guard against missing attribute
                    pass
        return True
 
    def _close_file(self, *, lock: Optional[bool] = True) -> bool:
        """Close the current file descriptor, optionally acquiring a lock.
 
        When `lock` is True the instance lock `self._file_lock` is
        acquired before closing the file. Returns True on completion.
        """
        # Snapshot and clear the descriptor under lock, then perform the
        # actual close outside the lock to avoid blocking other callers.
        descriptor = None
        if lock:
            with self._file_lock_file_lock_file_lock:
                if self.filefile:
                    descriptor = getattr(self.filefile, "descriptor", None)
                    try:
                        self.filefile.descriptor = None
                    except AttributeError:
                        pass
        else:
            if self.filefile:
                descriptor = getattr(self.filefile, "descriptor", None)
 
        if descriptor:
            try:
                descriptor.close()
            except (OSError, ValueError):
                # ignore close errors
                pass
        return True
 
    def _flush_buffer(self) -> None:
        """Internal: detach pending buffer and write to disk.
 
        Implements the swap-buffer pattern: capture and clear the in-memory
        buffer while holding the lock, perform I/O outside the lock, then
        update counters and rotate under lock.
        """
        # Swap-buffer pattern: capture and detach the in-memory buffer under
        # lock, then perform I/O outside the lock. If the file descriptor is
        # missing, perform the open outside the lock as well to avoid blocking
        # other callers.
        self.rogger.log_debug("Flushing buffer", stream=CONST.RAW_STDOUT)
        with self._file_lock_file_lock_file_lock:
            if not self._buffer:
                return
            to_write = self._buffer
            self._buffer = []
 
            if not self._log_to_file:
                return
 
            # Determine whether we need to open a descriptor. Do not perform
            # the actual open while holding the lock.
            needs_open = False
            descriptor = None
            if self.filefile:
                descriptor = getattr(self.filefile, "descriptor", None)
                if descriptor and not getattr(descriptor, "closed", False):
                    needs_open = False
                else:
                    needs_open = True
            else:
                needs_open = True
 
        # If necessary, open the file outside the lock
        if needs_open:
            try:
                self.rogger.log_debug(
                    f"_flush_buffer: needs_open=True, to_write_lines={len(to_write)}",
                    stream=CONST.RAW_STDOUT
                )
            except (AttributeError, OSError, ValueError):
                pass
            try:
                log_path: Path = self._create_log_path()
                self.filefile = self._open_file(log_path)
            except (OSError, ValueError):
                # If we can't open the file, skip writing but keep buffer
                # detached (to avoid blocking writers). We'll attempt again
                # on the next flush.
                return
        try:
            self.rogger.log_debug(
                f"_flush_buffer: performing write, lines={len(to_write)}",
                stream=CONST.RAW_STDOUT
            )
        except (AttributeError, OSError, ValueError):
            pass
        # perform actual write outside the lock
        try:
            descriptor = None
            if self.filefile:
                descriptor = getattr(self.filefile, "descriptor", None)
            if descriptor and not getattr(descriptor, "closed", False):
                descriptor.writelines(to_write)
                descriptor.flush()
        except (ValueError, OSError):
            try:
                self.rogger.log_warning(
                    "_flush_buffer: write failed, attempting reopen",
                    stream=CONST.RAW_STDERR
                )
            except (AttributeError, OSError, ValueError):
                pass
            # try reopening and write again once
            try:
                log_path: Path = self._create_log_path()
                self.filefile = self._open_file(log_path)
            except (OSError, ValueError):
                return
            descriptor = None
            if self.filefile:
                descriptor = getattr(self.filefile, "descriptor", None)
            if descriptor and not getattr(descriptor, "closed", False):
                try:
                    descriptor.writelines(to_write)
                    descriptor.flush()
                except (ValueError, OSError):
                    try:
                        self.rogger.log_error(
                            "_flush_buffer: retry write failed, giving up on this flush",
                            stream=CONST.RAW_STDERR
                        )
                    except (AttributeError, OSError, ValueError):
                        # give up on this flush attempt
                        pass
        else:
            try:
                # account bytes written for logging purposes
                _bytes = 0
                for line in to_write:
                    try:
                        _bytes += len(line.encode(self.encoding))
                    except LookupError:
                        _bytes += len(line.encode('utf-8'))
                self.rogger.log_debug(
                    f"_flush_buffer: write successful, approx_bytes={_bytes}",
                    stream=CONST.RAW_STDOUT
                )
            except (AttributeError, OSError, ValueError):
                pass
        # update counters and rotate under lock
        with self._file_lock_file_lock_file_lock:
            if not self.filefile:
                return
            for line in to_write:
                try:
                    self.filefile.written_bytes += len(line.encode(self.encoding))
                except LookupError:
                    self.filefile.written_bytes += len(line.encode('utf-8'))
            # perform rotation if needed
            self._rotate_file()