"""CsvPath is the main class for the library. most of the magic

happens either here or in individual functions.""" # pylint: disable=C0302

import time

import os

import hashlib

import traceback

from datetime import datetime

from typing import List, Dict, Any, Optional, Callable

from collections.abc import Iterator

from .util.config import Config

from .util.line_monitor import LineMonitor

from .util.log_utility import LogUtility as lout

from .util.printer import Printer

from .util.file_readers import DataFileReader

from .util.line_spooler import LineSpooler, ListLineSpooler

from .modes.mode_controller import ModeController

from .matching.matcher import Matcher

from .scanning.scanner2 import Scanner2 as Scanner

from .util.metadata_parser import MetadataParser

from .managers.errors.error import Error

from .managers.errors.error_comms import ErrorCommunications

from .managers.errors.error_manager import ErrorManager

from .managers.metadata import Metadata

from .util.printer import StdOutPrinter

from .util.line_counter import LineCounter

from .util.exceptions import VariableException, InputException, ParsingException

from .util.exceptions import (

FileException,

FormatException,

ProcessingException,

CsvPathsException,

)

from csvpath.managers.files.lines_and_headers_cacher import LinesAndHeadersCacher

from .matching.util.exceptions import MatchException

from .managers.errors.error_collector import ErrorCollector

from csvpath.util.date_util import DateUtility as daut

class CsvPath(ErrorCollector, Printer): # pylint: disable=R0902, R0904

"""CsvPath represents a csvpath string that contains a reference to

a file, scanning instructions, and rules for matching lines.

"""

# re R0902, R0904: reasonable, but not a priority

def __init__( # pylint: disable=R0913

self,

*,

csvpaths=None,

delimiter=",",

quotechar='"',

skip_blank_lines=True,

print_default=True,

# config=None,

#

# atm, we don't use this error manager reference. instead we make

# an error manager for this csvpath, and if we have a csvpaths we

# add its error_manager as an internal listener for error events

#

# we can refactor this reference away

#

error_manager=None,

project_context="no_project_context",

project="no_project_name",

):

#

# these identifiers are for the logging system. they are optional but useful

#

self.project = project

self.project_context = project_context

"""

#

# passing in the config actually nets us nothing but a little unnecessary complexity that we don't use.

#

# we want to be able to override config.ini specifically for

# this instance, if needed; however, we do want to be able

# to pass in a config object that has been configured in some

# way.

"""

self._config = Config() # config

self.scanner = None

""" @private """

self.matcher = None

""" @private """

#

# a parent CsvPaths may manage a CsvPath instance. if so, it will enable

# the use of named files and named paths, print capture, error handling,

# results collection, reference handling, etc. if a CsvPaths is not present

# the CsvPath instance is responsible for all its own upkeep and does not

# have some of those capabilities.

#

self.csvpaths = csvpaths

""" @private """

#

# there are two logger components one for CsvPath and one for CsvPaths.

# the default levels are set in config.ini. to change the levels pass LogUtility

# your component instance and the logging level. e.g.:

# LogUtility.logger(csvpath, "debug")

#

self._logger = None

""" @private """

self.logger.info("initialized CsvPath")

#

# if we don't have a csvpaths these will both be None

#

self.named_paths_name = None

""" @private """

self.named_file_name = None

""" @private """

#

# all errors come to our manager if they occur during matching. we use

# the CsvPaths manager if possible. Otherwise, we just make our own that

# only knows how to collect errors, not distribute them.

#

self.ecoms = ErrorCommunications(csvpath=self)

""" @private """

self.error_manager = ErrorManager(csvpath=self)

if csvpaths is not None:

self.error_manager.add_internal_listener(csvpaths.error_manager)

#

# modes are set in external comments

#

self.modes = ModeController(self)

""" @private """

#

# captures the number of lines up front and tracks line stats as the

# run progresses

#

self._line_monitor = None

#

# the scanning part of the csvpath. e.g. $test.csv[*]

#

self.scan = None

""" @private """

#

# the matching part of the csvpath. e.g. [yes()]

#

self.match = None

""" @private """

#

# when True the lines that do not match are returned from next()

# and collect(). this effectively switches CsvPath from being an

# create an OR expression in this case. in the default, we say:

# are all of these things true?

# but when collect_when_not_matched is True we ask:

# are any of these things not true?

#

# self._when_not_matched = False

self._headers = None

self.variables: Dict[str, Any] = {}

self.delimiter = delimiter

self.quotechar = quotechar

#

# match expressions -- and others -- can register a callback that

# will be called when CsvPath.flush() is called at the end of a run

# or called intentionally by any user code that is using next().

#

self._flushes: list[Callable[[None], None]] = []

#

# a blank line has no headers. it has no data. physically it is 2 \n with

# nothing but whitespace between them. any data or any delimiters would make

# the line non-blank.

#

self.skip_blank_lines = skip_blank_lines

#

# in the case of a [*] scan where the last line is blank we would miss firing

# last() unless we take steps. instead, we allow that line to match, but we

# do not return a line to the caller of next() and we freeze the variables

# there is room for side effects make changes, but that a reasonable compromise

# between missing last and allowing unwanted changes. we definitely do not

# freeze is_valid or stop, which can be useful signaling, even in an

# inconsistent state.

#

self._freeze_path = False

#

# counts are 1-based

#

self.scan_count = 0

self.match_count = 0

#

# used by stop() and advance(). a stopped CsvPath halts without finishing

# its run. an advancing CsvPath doesn't consider the match part of the

# csvpath and does not incur any side effects as it progresses through the

# rows the advance skips. the skip() function has the same effect as

# advance(1) but without any guarantee that the other match components on

# the line will be considered before skipping ahead. there are likely

# corner cases where an onmatch qualifier or some other constraint will

# trigger match components that would otherwise be skipped so the ability

# to shortcut some of the match should not be relied on for anything

# critical.

#

self.stopped = False

self._advance = 0

#

# the lines var will hold a reference to a LineSpooler instance during a

# run, if lines are being collected by the collect() method. if the user

# is using this CsvPath instance directly the LineSpooler is only there

# as a proxy to the list of lines being collected -- we don't spool to

# disk, at least atm.

#

self.lines = None

""" @private """

#

# set by fail()

#

self._is_valid = True

#

# basic timing for the CsvPath instance only. if the CsvPath is managed

# by a CsvPaths the timings for a run may include time spent by other

# CsvPath instances.

#

self.last_row_time = -1

""" @private """

self.rows_time = -1

""" @private """

self.total_iteration_time = -1

""" @private """

#

# limiting collection means returning fewer headers (values in the

# line, a.k.a columns) then are available. limiting headers returned

# can impact named results, reset_headers(), and other considerations.

#

self._limit_collection_to = None # []

#

# error collecting is at the CsvPath instance by default. CsvPath

# instances that are managed by a CsvPaths have their errors collected

# by their Results as well. Result handles persistence.

#

# errors policies are set in config.ini at CsvPath and CsvPaths levels.

#

self._errors: List[Error] = []

#

# saves the scan and match parts of paths for reference. mainly helpful

# for testing the CsvPath library itself; not used end users. the run

# name becomes the file name of the saved path parts.

#

self._save_scan_dir = None

self._save_match_dir = None

self._run_dir = None

#

# metadata is collected from "outer" csvpath comments. outer comments

# separate from the comments within the match part of the csvpath.

# the keys are words with colons. e.g. ~ name: my new csvpath ~

#

self.metadata: Dict[str, Any] = {}

#

# holds the current match count while we're in the middle of a match

# so that anyone who wants to can increase the match count using

# raise_match_count_if(). it is important to do the raise asap so that

# components that are onmatched have the right match count available.

#

self._current_match_count = 0

#

# printers receive print lines from the print function. the default

# printer prints to standard out. a CsvPath that is managed by a

# CsvPaths has its Results as a printer, as well as having

# the default printer.

#

self.printers = []

""" @private """

if print_default:

self.printers.append(StdOutPrinter())

#

# _function_times_match collects the time a function spends doing its matches()

#

self._function_times_match = {}

#

# _function_times_value collects the time a function spends doing its to_value()

#

self._function_times_value = {}

self._created_at = daut.now()

# self._created_at = datetime.now(timezone.utc)

self._run_started_at = None

self._collecting = False

#

# holds the unmatched lines when lines are being collected and

# _unmatched_available is True. it is analogous to the lines returned

# by collect(), but is the lines not returned by collect().

#

self._unmatched = None

self._cacher = None

@property

def run_dir(self) -> str:

return self._run_dir

@run_dir.setter

def run_dir(self, d: str) -> None:

self._run_dir = d

@property

def logger(self):

if self._logger is None:

self._logger = lout.logger(self)

return self._logger

@logger.setter

def logger(self, ler) -> None:

self._logger = ler

def __del__(self) -> None:

self.csvpaths = None

self.matcher = None

try:

# in a test on windows 0.0.570 we see self has no error_manager attribute

# that is surprising since there is one ^^^^. no idea. this test is cheap tho.

if (

hasattr(self, "error_manager")

and self.error_manager

and self.error_manager.error_metrics

):

self.error_manager.error_metrics.provider.shutdown()

self.error_manager.error_metrics = None

self.error_manager = None

except Exception:

print(traceback.format_exc())

finally:

lout.release_logger(self)

#

# this method saves and reloads the config. if you don't want that use

# CsvPath.config.save_to_config().

#

def add_to_config(self, section, key, value) -> None:

"""@private"""

self.config.add_to_config(section=section, key=key, value=value)

self.config.save_config()

self.config.reload()

@property

def cacher(self) -> LinesAndHeadersCacher:

if self._cacher is None:

if self.csvpaths:

self._cacher = self.csvpaths.file_manager.lines_and_headers_cacher

else:

self._cacher = LinesAndHeadersCacher(

self, line_counter=LineCounter(self)

)

return self._cacher

@property

def data_from_preceding(self) -> bool:

"""@private"""

return self.modes.source_mode.value

@data_from_preceding.setter

def data_from_preceding(self, dfp: bool) -> None:

"""@private"""

self.modes.source_mode.value = dfp

@property

def unmatched(self) -> list[list[Any]]:

"""@private"""

return self._unmatched

@unmatched.setter

def unmatched(self, lines: list[list[Any]]) -> None:

"""@private"""

self._unmatched = lines

@property

def collecting(self) -> bool:

"""@private"""

return self._collecting

@collecting.setter

def collecting(self, c: bool) -> None:

"""@private"""

self._collecting = c

@property

def unmatched_available(self) -> bool:

"""@private"""

return self.modes.unmatched_mode.value

@unmatched_available.setter

def unmatched_available(self, ua: bool) -> None:

"""@private"""

self.modes.unmatched_mode.value = ua

@property

def created_at(self) -> datetime:

"""@private"""

return self._created_at

@property

def run_started_at(self) -> datetime:

"""@private"""

return self._run_started_at

@property

def will_run(self) -> bool:

"""@private"""

return self.modes.run_mode.value

@will_run.setter

def will_run(self, mode) -> None:

"""@private"""

self.modes.run_mode.value = mode

#

# increases the total accumulated time spent doing c.matches() by t

#

def up_function_time_match(self, c, t) -> None:

"""@private"""

if c not in self.function_times_match:

self.function_times_match[c] = 0

st = self.function_times_match[c]

st += t

self.function_times_match[c] = st

@property

def function_times_match(self) -> int:

"""@private"""

return self._function_times_match

#

# increases the total accumulated time spent doing c.to_value() by t

#

def up_function_time_value(self, c, t) -> None:

"""@private"""

if c not in self.function_times_value:

self.function_times_value[c] = 0

st = self.function_times_value[c]

st += t

self.function_times_value[c] = st

@property

def function_times_value(self) -> int:

"""@private"""

return self._function_times_value

def do_i_raise(self) -> bool:

"""@private"""

return self.ecoms.do_i_raise()

@property

def advance_count(self) -> int: # pragma: no cover

"""@private"""

return self._advance

@advance_count.setter

def advance_count(self, lines: int) -> None:

"""@private"""

self._advance = lines

@property

def headers(self) -> List[str]:

"""@private"""

if self._headers is None:

self.get_total_lines_and_headers()

return self._headers

@headers.setter

def headers(self, headers: List[str]) -> None:

"""@private"""

self._headers = headers

@property

def line_monitor(self) -> LineMonitor:

"""@private"""

if self._line_monitor is None:

self.get_total_lines_and_headers()

return self._line_monitor

@line_monitor.setter

def line_monitor(self, lm) -> None:

"""@private"""

self._line_monitor = lm

@property

def AND(self) -> bool: # pylint: disable=C0103

return self.modes.logic_mode.value

@AND.setter

def AND(self, a: bool) -> bool: # pylint: disable=C0103

self.modes.logic_mode.value = a

@property

def OR(self) -> bool: # pylint: disable=C0103

return not self.modes.logic_mode.value

@OR.setter

def OR(self, a: bool) -> bool: # pylint: disable=C0103

self.modes.logic_mode.value = not a

@property

def identity(self) -> str:

"""returns id or name if found in metadata.

the id or name gets into metadata primarily if found

in an "external" comment in the csvpath. "external"

meaning outside the []s. comments are keyword:comment.

we take id, Id, ID and name, Name, NAME.

id is preferred over name. E.g. in:

~ name: my path description: an example id: this value wins ~

the id becomes the identity of the instance.

we prefer in this order: all-lower most, Initial-caps,

ALL-CAPS least

the ordering is relied on in Result and possibly

elsewhere.

"""

ret = None

if not self.metadata:

ret = ""

if "NAME" in self.metadata:

ret = self.metadata["NAME"]

if "Name" in self.metadata:

ret = self.metadata["Name"]

if "name" in self.metadata:

ret = self.metadata["name"]

if "ID" in self.metadata:

ret = self.metadata["ID"]

if "Id" in self.metadata:

ret = self.metadata["Id"]

if "id" in self.metadata:

ret = self.metadata["id"]

return ret

@property

def config(self) -> Config: # pylint: disable=C0116

"""@private"""

if not self._config:

self._config = Config()

return self._config

# ==========================

# Errors

# <thinking> if we have a csvpaths people should look at the result to find errors

# but we give access to metadata, vars, etc. from the csvpath, so we should

# give errors too. that means we need to have our own listener. ultimately we'd

# just be adding pointers, not dup the original error data.

#

def metadata_update(self, mdata: Metadata) -> None:

"""@private"""

if isinstance(mdata, Error):

self.collect_error(mdata)

@property

def errors(self) -> List[Error]: # pylint: disable=C0116

return self._errors

@property

def errors_count(self) -> int: # pylint: disable=C0116

return len(self.errors)

def collect_error(self, error: Error) -> None: # pylint: disable=C0116

"""@private"""

if not self.has_error(error):

self.errors.append(error)

def has_error(self, e: Error) -> bool:

for _ in self.errors:

if _.equals(e):

return True

return False

def has_errors(self) -> bool:

return self.errors_count > 0

@property

def stop_on_validation_errors(self) -> bool:

"""@private"""

return self.modes.validation_mode.stop_on_validation_errors

@property

def fail_on_validation_errors(self) -> bool:

"""@private"""

return self.modes.validation_mode.fail_on_validation_errors

@property

def print_validation_errors(self) -> bool:

"""@private"""

return self.modes.validation_mode.print_validation_errors

@property

def log_validation_errors(self) -> bool:

"""@private"""

return self.modes.validation_mode.log_validation_errors

@property

def raise_validation_errors(self) -> bool:

"""@private"""

return self.modes.validation_mode.raise_validation_errors

@property

def match_validation_errors(self) -> bool:

"""@private"""

return self.modes.validation_mode.match_validation_errors

@property

def collect_validation_errors(self) -> bool:

"""@private"""

return self.modes.validation_mode.collect_validation_errors

@property

def consolidate_printouts(self) -> bool:

return self.modes.print_mode.consolidate_printouts

def add_printer(self, printer) -> None: # pylint: disable=C0116

"""@private"""

if printer not in self.printers:

self.printers.append(printer)

def set_printers(self, printers: List) -> None: # pylint: disable=C0116

"""@private"""

self.printers = printers

@property

def has_default_printer(self) -> bool:

"""@private"""

if not self.printers:

self.printers = []

for p in self.printers:

if isinstance(p, StdOutPrinter):

return True

return False

def print(self, string: str) -> None: # pylint: disable=C0116

"""@private"""

for p in self.printers:

p.print(string)

def print_to(self, name: str, string: str) -> None:

"""@private"""

for p in self.printers:

p.print_to(name, string)

@property

def last_line(self):

"""@private

this method only returns the default printer's last_line"""

if not self.printers or len(self.printers) == 0:

return None

return self.printers[0].last_line

@property

def lines_printed(self) -> int:

"""@private

this method only returns the default printer's lines printed"""

if not self.printers or len(self.printers) == 0:

return -1

return self.printers[0].lines_printed

@property

def is_frozen(self) -> bool:

"""@private

True if the instance is matching on its last row only to

allow last()s to run; in which case, no variable updates

are allowed, along with other limitations."""

return self._freeze_path

@is_frozen.setter

def is_frozen(self, freeze: bool) -> None:

"""@private"""

self._freeze_path = freeze

@property

def explain(self) -> bool:

"""@private

when this property is True CsvPath dumps a match explanation

to INFO. this can be expensive. a 25% performance hit wouldn't

be unexpected.

"""

return self.modes.explain_mode.value

@explain.setter

def explain(self, yesno: bool) -> None:

"""@private"""

self.modes.explain_mode.value = yesno

@property

def collect_when_not_matched(self) -> bool:

"""@private

when this property is True CsvPath returns the lines that do not

match the matchers match components"""

return self.modes.return_mode.collect_when_not_matched

@collect_when_not_matched.setter

def collect_when_not_matched(self, yesno: bool) -> None:

"""@private

when c ollect_when_not_matched is True we return the lines that failed

to match, rather than the default behavior of returning the matches.

"""

self.modes.return_mode.collect_when_not_matched = yesno

def parse(self, csvpath, disposably=False):

"""@private

displosably is True when a Matcher is needed for some purpose other than

the run we were created to do. could be that a match component wanted a

parsed csvpath for its own purposes. when True, we create and return the

Matcher, but then forget it ever existed.

when disposably is False we build the scanner and return that

"""

#

# strip off any comments and collect any metadata

# CsvPaths will do this earlier but it stripped off

# the comments so we won't find them again

#

csvpath = MetadataParser(self).extract_metadata(instance=self, csvpath=csvpath)

self.update_settings_from_metadata()

#

#

#

if disposably is False:

csvpath = self._update_file_path(csvpath)

#

#

#

s, mat = self._find_scan_and_match_parts(csvpath)

#

# a disposable matcher still needs the match part

#

self.match = mat

if disposably:

pass

else:

self.scan = s

self.scanner = Scanner(csvpath=self)

self.scanner.parse(s)

#

# we build a matcher to see if it builds without error.

# in principle we could keep this as the actual matcher.

# atm, tho, just create a dry-run copy. in some possible

# unit tests we may not have a parsable match part.

#

if disposably:

matcher = None

if mat:

matcher = Matcher(csvpath=self, data=mat, line=None, headers=None)

#

# if the matcher was requested for some reason beyond our own needs

# we just return it and forget it existed.

#

return matcher

if self.scanner.filename is None:

raise FileException("Cannot proceed without a filename")

self.get_total_lines_and_headers()

return self

def update_settings_from_metadata(self) -> None:

"""@private"""

#

# settings:

# - logic-mode: AND | OR

# - return-mode: matches | no-matches

# - print-mode: default | no-default

# - validation-mode: (no-)print | log | (no-)raise | quiet | (no-)match | (no-)stop

# - run-mode: no-run | run

# - unmatched-mode: no-keep | keep

# - source-mode: preceding | origin

# - files-mode: all | no-data | no-unmatched | no-printouts | data

# | unmatched | errors | meta | vars | printouts

#

self.modes.update()

#

# if we find "use-delimiter" or "use-quotechar" we need to update ourselves. this is primarily for

# flightpath server (and other similar non-programmatic uses). these could be modes but it doesn't

# feel like we're changing the behavior of the framework so much as just passing a parameter

# declaratively, similar to the integrations and "test-delimiter", "test-quotechar".

#

d = self.metadata.get("use-delimiter")

if d:

v = ["pipe", "bar", "semi", "comma", "colon", "tab", "space"]

if d not in v:

raise ValueError(f"The use-delimiter directive must be one of {v}")

v = {

"pipe": "|",

"bar": "|",

"semi": ";",

"comma": ",",

"colon": ":",

"tab": "\t",

"space": " ",

}

self.delimiter = v[d]

q = self.metadata.get("use-quotechar")

if q:

v = ["quotes", "quote", "single-quote", "singlequote", "single", "tick"]

if q not in v:

raise ValueError(f"The use-quotechar directive must be one of {v}")

v = {

"quotes": '"',

"quote": '"',

"single-quote": "'",

"singlequote": "'",

"single": "'",

"tick": "`",

}

self.quotechar = v[q]

# =====================

# in principle the modes should come through the mode controller like:

# self.modes.transfer_mode.value

# not wading into that today. low value.

#

@property

def transfer_mode(self) -> str:

"""@private"""

return self.metadata.get("transfer-mode")

@property

def source_mode(self) -> str:

"""@private"""

return self.metadata.get("source-mode")

@property

def error_mode(self) -> str:

"""@private"""

return self.metadata.get("error-mode")

@property

def files_mode(self) -> str:

"""@private"""

return self.metadata.get("files-mode")

@property

def validation_mode(self) -> str:

"""@private"""

return self.metadata.get("validation-mode")

@property

def run_mode(self) -> str:

"""@private"""

return self.metadata.get("run-mode")

@property

def logic_mode(self) -> str:

"""@private"""

return self.metadata.get("logic-mode")

@property

def return_mode(self) -> str:

"""@private"""

return self.modes.get("return-mode")

@property

def explain_mode(self) -> str:

"""@private"""

return self.metadata.get("explain-mode")

@property

def print_mode(self) -> str:

"""@private"""

return self.metadata.get("print-mode")

@property

def unmatched_mode(self) -> str:

"""@private"""

return self.metadata.get("unmatched-mode")

# =====================

@property

def transfers(self) -> list[tuple[str, str]]:

"""@private"""

return self.modes.transfer_mode.transfers

@property

def all_expected_files(self) -> list[str]:

"""@private"""

return self.modes.files_mode.all_expected_files

@all_expected_files.setter

def all_expected_files(self, efs: list[str]) -> None:

"""@private"""

self.modes.files_mode.all_expected_files = efs

def _pick_named_path(self, name, *, specific=None) -> str:

"""@private"""

if not self.csvpaths:

raise CsvPathsException("No CsvPaths object available")

np = self.csvpaths.paths_manager.get_named_paths(name)

if not np:

raise CsvPathsException(f"Named-paths '{name}' not found")

if len(np) == 0:

raise CsvPathsException(f"Named-paths '{name}' has no csvpaths")

if len(np) == 1:

return np[0]

if specific is None:

self.logger.warning(

"Parse_named_path %s has %s csvpaths. Using just the first one.",

name,

len(np),

)

return np[0]

for p in np:

# this ends up being redundant to the caller. we do it 1x so it's not

# a big lift and is consistent.

c = CsvPath(csvpaths=self.csvpaths)

MetadataParser(c).extract_metadata(instance=c, csvpath=p)

if c.identity == specific:

return p

self.logger.error(

"Cannot find csvpath identified as %s in named-paths %s", specific, name

)

raise ParsingException(f"Cannot find path '{specific}' in named-paths '{name}'")

def parse_named_path(self, name, *, disposably=False, specific=None):

"""@private

disposably is True when a Matcher is needed for some purpose other than

the run we were created to do. could be that a match component wanted a

parsed csvpath for its own purposes. import() uses this method.

when True, we create and return the Matcher, but then forget it ever existed.

also note: the path must have a name or full filename. $[*] is not enough.

"""

if not self.csvpaths:

raise CsvPathsException("No CsvPaths object available")

path = self._pick_named_path(name, specific=specific)

c = CsvPath(csvpaths=self.csvpaths)

path = MetadataParser(c).extract_metadata(instance=c, csvpath=path)

#

# exp. oddly this seems to be superfluous

# if disposably is False:

# path = c._update_file_path(path)

#

dis = c.parse(path, disposably=disposably)

if disposably is True:

return dis

return None

def _update_file_path(self, data: str):

"""@private

this method replaces a name (i.e. name in: $name[*[][yes()]) with

a file system path, if that name is registered with csvpaths's file

manager. if there is no csvpaths no replace happens. if there is a

csvpaths but the file manager doesn't know the name, no replace

happens.

"""

if data is None:

raise InputException("The csvpath string cannot be None")

if self.csvpaths is None:

return data

name = self._get_name(data)

#

# this will blow up frequently when name is an actual path. ie name == path

# below. since we want file manager to be discriminating we have to catch

# the error and reset the name

#

try:

path = self.csvpaths.file_manager.get_named_file(name)

except ValueError:

path = name

if path is None:

return data

if path == name:

return data

return data.replace(name, path)

def _get_name(self, data: str):

if self.csvpaths is None:

return data

data = data.strip()

if data[0] == "$":

name = data[1 : data.find("[")]

return name

raise FormatException(f"Must start with '$', not {data[0]}")

def _find_scan_and_match_parts(self, data):

if data is None or not isinstance(data, str):

raise InputException("Not a csvpath string")

scan = ""

matches = ""

data = data.strip()

i = data.find("]")

if i < 0:

raise InputException(f"Cannot find the scan part of this csvpath: {data}")

if i == len(data) - 1:

raise InputException(

f"The scan part of this csvpath cannot be last: {data}"

)

scan = data[0 : i + 1]

scan = scan.strip()

ndata = data[i + 1 :]

ndata = ndata.strip()

if ndata == "":

raise InputException(f"There must be a match part of this csvpath: {data}")

if ndata[0] != "[":

raise InputException(f"Cannot find the match part of this csvpath: {data}")

if ndata[len(ndata) - 1] != "]":

raise InputException(f"The match part of this csvpath is incorrect: {data}")

matches = ndata

#

# if we're given directory(s) to save to, save the parts

#

self._save_parts_if(scan, matches)

return scan, matches

def _save_parts_if(self, scan, match):

if self._save_scan_dir and self._run_dir: