.gitignore

@@ -12,4 +12,3 @@ ghlib/
 *.txt
 
 !version.txt
-!tests/test_inputs/*

Hold Reconciler.spec

@@ -5,11 +5,11 @@ block_cipher = None
 
 
 a = Analysis(
-    ['hold_reconciler.py'],
-    pathex=['\\leafnow.com\shared\Business Solutions\Griff\Code\HoldReconciler'],
+    ['reconcile_holds.py'],
+    pathex=[],
     binaries=[],
-    datas=[('.\\config_logger.toml', '.'), ('.\\config_reports.toml', '.')],
-    hiddenimports=['reports.*','memory.*','helpers.*'],
+    datas=[('config.toml', '.'), ('requirements.txt', '.')],
+    hiddenimports=['openpyxl'],
     hookspath=[],
     hooksconfig={},
     runtime_hooks=[],

config_reports.toml

@@ -0,0 +1,31 @@
+output_columns = [
+        "contract_number",
+        "vendor_name",
+        "AppNum",         # OB only
+        "DateBooked",     # OB only
+        "Document Number",# GP Only
+        "Resolution",
+        "Notes"
+        # 'Source' added for 'no match'
+    ]
+
+[gp_filters]
+    # These regex will be combined and with ORs and used to filer
+    # the document number column of the GP report 
+    doc_num_filters = [
+        "p(oin)?ts",
+        "pool",
+        "promo",
+        "o(ver)?f(und)?",
+        "m(ar)?ke?t",
+        "title",
+        "adj",
+        "reg free",
+        "cma"
+    ]
+    po_filter = "^(?!.*cma(\\s|\\d)).*$"
+  
+[shared_columns]
+contract_number = { GP = "Transaction Description", OB = "Contract"}
+onhold_amount = { GP = "Current Trx Amount", OB = "CurrentOnHold" }
+vendor_name = { GP = "Vendor Name", OB = "DealerName"} 

hold_reconciler.py

@@ -5,7 +5,7 @@ saved as an excel file with todays date.
 """
 # Custom module for reconciliation
 from helpers import setup_logging, find_most_recent_file
-from reports import OnBaseReport, GreatPlainsReport, ReconciledReports
+from reports import OnBaseReport, GreatPlainsReport
 
 import pandas as pd
 from pandas import DataFrame
@@ -15,8 +15,19 @@ import logging
 from tomllib import load
 import logging.config
 from datetime import datetime as dt
+from openpyxl import load_workbook, Workbook
+import pathlib
 from pathlib import Path
 
+"""
+[ ] Pull in past reconciliations to check against
+[ ] Record reconciled transaction (connect with VBA)
+[ ] Check GP against the database
+[ ] Check OB against the database
+[X] Add resolution column to error sheets 
+[ ] Add sheet for problem contractas already seen and 'resolved'
+"""
+
 setup_logging()
 logger = logging.getLogger(__name__)
 logger.info(f"Logger started with level: {logger.level}")
@@ -93,13 +104,28 @@ def main() -> int:
     obr: OnBaseReport = OnBaseReport(ob_df, reports_config)
     gpr: GreatPlainsReport = GreatPlainsReport(gp_df, reports_config)
 
-    rec_output: ReconciledReports = obr.reconcile(gpr)
+    overdue: DataFrame = obr.get_overdue()
    
-    output_name: Path = Path(f"Reconciled Holds [{dt.now().strftime('%m-%d-%Y')}].xlsx")
-    output_base: Path = Path(reports_config["output_path"])
-    output_path: Path = Path(output_base, output_name)
+    no_match, amt_mismatch = obr.reconcile(gpr)
 
-    rec_output.save_reports(output_path)
+    # Write the results to a new Excel file
+    output_name: Path = Path(f"Reconciled Holds [{dt.now().strftime('%m-%d-%Y')}].xlsx")
+    output_path: Path = Path("./Work", output_name)
+    with pd.ExcelWriter(output_path, mode='w') as writer:
+        no_match.to_excel(writer, sheet_name="No Match", 
+                          index=False, freeze_panes=(1,3)
+                          )
+        amt_mismatch.to_excel(writer, sheet_name="Amount Mismatch", 
+                            index=False, freeze_panes=(1,3)
+                            )
+        overdue.to_excel(writer, sheet_name="Overdue", index=False)
+
+    wb: Workbook = load_workbook(output_path)
+    for sheet in ["No Match", "Amount Mismatch"]:
+        ws = wb[sheet]
+        ws.column_dimensions['A'].hidden = True
+        ws.column_dimensions['B'].hidden = True
+    wb.save(output_path)
 
     return 0
 

memory.py

@@ -8,59 +8,51 @@ resolved holds.
 *Last Updated: version 1.3
 """
 from helpers import drop_unnamed, setup_logging
-from ghlib.database.database_manager import SQLiteManager, select_fields_statement
+from ghlib.database.database_manager import SQLiteManager
 
 from pandas import DataFrame, Series, read_sql_query, read_excel, concat
-from numpy import NaN
 from logging import getLogger
 from dataclasses import dataclass
 from hashlib import md5
-from typing import TypeAlias
 
 setup_logging()
 logger = getLogger(__name__)
 
-col_hash: TypeAlias = str
 
-def hash_cols(row: Series, cols_to_hash: list[str]) -> col_hash:
+def hash_cols(row: Series, cols_to_hash: list[str]) -> str:
     md5_hash = md5()
-    md5_hash.update((''.join(str(row[col]) for col in cols_to_hash)).encode('utf-8'))
+    md5_hash.update((''.join(row[col] for col in cols_to_hash)).encode('utf-8'))
     return md5_hash.hexdigest()
 
-def create_identifier(df: DataFrame) -> DataFrame:
-    for id in ["ID_OB","ID_GP"]:
-        df[id].fillna("x", inplace=True)
-    df["Indentifier"] = df.apply(lambda row: 
-        hash_cols(row, ["ID_OB","ID_GP"]), axis=1
-    )
-    for id in ["ID_OB","ID_GP"]:
-        df[id].replace('x',NaN, inplace=True)
-    return df
      
 def save_rec(resolved_dataframes: list[DataFrame]):
     """
+    #TODO Actually handle this...
     """
+    #raise NotImplementedError("You were too lazy to fix this after the rewrite. FIX PLZ!")
     sqlManager: SQLiteManager = SQLiteManager("OnHold.db")
     with sqlManager.get_session() as session:
+        conn = session.connection()
 
         rdf: DataFrame
         for rdf in resolved_dataframes:
             cols: list[str] = rdf.columns.to_list()
-            logger.debug(f"{cols=}")
             if "onhold_amount" in cols:
-                logger.debug("Found 'onhold_amount' in rdf: no_match dataframe")
+                logger.debug(f"Found 'onhold_amount' in rdf: no_match dataframe")
                 # Split the on_hold col to normalize with amount mismatch
                 rdf["onhold_amount_GP"] = rdf.apply(lambda row:
-                    row["onhold_amount"] if row["Source"] == "GP" else None
-                , axis=1)
+                    row.onhold_amount if row.Source == "GP" else None
+                )
                 rdf["onhold_amount_OB"] = rdf.apply(lambda row:
-                    row["onhold_amount"] if row["Source"] == "OB" else None
-                , axis=1 )
+                    row.onhold_amount if row.Source == "OB" else None
+                )
             else:
-                logger.debug("No 'onhold_amount' col found in rdf: amount_mismatch dataframe")
-            
+                logger.debug(f"No 'onhold_amount' col found in rdf: amount_mismatch dataframe")
             # Create a unified column for index 
-            rdf = create_identifier(rdf)
+            rdf["Indentifier"] = rdf.apply(lambda row: 
+                hash_cols(row, ["ID_OB","ID_GP"]), axis=1
+            )
+
 
             rec_cols: list[str] = [
                 "Indentifier",
@@ -70,20 +62,11 @@ def save_rec(resolved_dataframes: list[DataFrame]):
                 "Resolution"
             ]
             
-            rdf = rdf[rec_cols]
-            rdf.set_index("Indentifier", inplace=True, drop=True)
-            rdf.drop_duplicates(inplace=True)
-            rdf = rdf.dropna(axis=0, how="all", subset=["HideNextMonth", "Resolution"])
-            logger.debug(f"Saving resolutions to db:\n{rdf}")
 
-            rdf.to_sql('Resolutions', 
-                con=session.connection(),
-                if_exists="append"    
-            )
     
          
 
-def get_prev_reconciled(identfiers: list[col_hash]) -> DataFrame|None:
+def get_prev_reconciled(contracts: list[str]) -> DataFrame:
     """
     Get a DataFrame of previously reconciled contracts from an SQLite database.
 
@@ -99,26 +82,23 @@ def get_prev_reconciled(identfiers: list[col_hash]) -> DataFrame|None:
     # Create a temp table to hold this batches contract numbers
     # this table will be cleared when sqlManager goes out of scope
     temp_table_statement = """
-    CREATE TEMPORARY TABLE CUR_IDENT (Indentifier VARCHAR(32));
+    CREATE TEMPORARY TABLE CUR_CONTRACTS (contract_number VARCHAR(11));
     """
     sqlManager.execute(temp_table_statement)
 
     # Insert the current contracts into the temp table
-    insert_idents = f"""
-    INSERT INTO CUR_IDENT (Indentifier) VALUES
-    {', '.join([f"('{cn}')" for cn in identfiers])};
+    insert_contracts = f"""
+    INSERT INTO CUR_CONTRACTS (contract_number) VALUES
+    {', '.join([f"('{cn}')" for cn in contracts])};
     """
-
-    logger.debug(f"{insert_idents=}")
-
-    sqlManager.execute(insert_idents)
+    sqlManager.execute(insert_contracts)
 
     # Select previously resolved contracts
     res_query = """
     SELECT r.*
     FROM Resolutions r
-    JOIN CUR_IDENT i
-    ON r.Indentifier = i.Indentifier;
+    JOIN CUR_CONTRACTS t
+    ON r.contract_number = t.contract_number;
     """
     resolved: DataFrame = sqlManager.execute(res_query, as_dataframe=True)
     return resolved

reports.py

@@ -1,54 +1,17 @@
-from pandas import DataFrame, merge, to_datetime, NaT, concat, ExcelWriter
-from openpyxl import Workbook, load_workbook
-from abc import ABC
+from pandas import DataFrame, merge, to_datetime, NaT, concat, Series
+from numpy import concatenate
+from abc import ABC, abstractmethod
 from logging import getLogger
 import re
+from typing import Literal
 import datetime
 from copy import deepcopy
-from dataclasses import dataclass
+
 from helpers import CN_REGEX, drop_unnamed
-from memory import get_prev_reconciled, hash_cols, col_hash, create_identifier
-from pathlib import Path
+from memory import get_prev_reconciled
 
 logger = getLogger(__name__)
 
-@dataclass
-class ReconciledReports:
-    no_match: DataFrame
-    amt_mismatch: DataFrame
-    prev_rec: DataFrame
-    gp_filtered: DataFrame
-    ob_overdue: DataFrame
-    
-    def save_reports(self, output_path: Path):
-        with ExcelWriter(output_path, mode='w') as writer:
-            self.no_match.drop_duplicates(inplace=True)
-            self.no_match.to_excel(writer, sheet_name="No Match", 
-                                index=False, freeze_panes=(1,3)
-                                )
-            self.amt_mismatch.drop_duplicates(inplace=True)
-            self.amt_mismatch.to_excel(writer, sheet_name="Amount Mismatch", 
-                                index=False, freeze_panes=(1,3)
-                                )
-            self.ob_overdue.to_excel(writer, sheet_name="Overdue", 
-                                index=False
-                                )
-            self.prev_rec.to_excel(writer, sheet_name="Previously Reconciled", 
-                                index=False, freeze_panes=(1,3)
-                                )
-            self.gp_filtered.to_excel(writer, sheet_name="Filtered from GP", 
-                                index=False, freeze_panes=(1,0)
-                                )
-
-        wb: Workbook = load_workbook(output_path)
-        for sheet in ["No Match", "Amount Mismatch"]:
-            ws = wb[sheet]
-            ws.column_dimensions['A'].hidden = True
-            ws.column_dimensions['B'].hidden = True
-        for sheet in ["Filtered from GP", "Previously Reconciled"]:
-            wb[sheet].sheet_state = "hidden"          
-        wb.save(output_path)
-        wb.close()
 
 class HoldReport(ABC):
 
@@ -58,8 +21,9 @@ class HoldReport(ABC):
         self.config = reports_config
         drop_unnamed(dataframe)
         self.df = dataframe
-        self.df = self._add_work_columns(self.df)
+        self.prev_rec = None
         self._normalize()
+        self._previsouly_resolved()
 
 
     def _normalize(self):
@@ -86,60 +50,55 @@ class HoldReport(ABC):
         self.df["Source"] = self.source
 
 
-    @staticmethod
-    def _remove_prev_recs(contract_match, no_match) -> \
-        tuple[DataFrame, DataFrame, DataFrame]:
+    def _previsouly_resolved(self):
         """
         """
+        current_contracts: list[str] = self.df["contract_number"]
 
-        idents: list[col_hash] = create_identifier(contract_match)["Indentifier"].to_list()
-        idents.extend(create_identifier(no_match)["Indentifier"].to_list())
-        logger.debug(f"{idents=}")
-        # Get previsouly reced
-        prev_recs: DataFrame|None = get_prev_reconciled(idents)
-
-        if prev_recs is None:
+        prev_recd: DataFrame = get_prev_reconciled(contracts=current_contracts)
+        if not prev_recd:
             logger.info("No previously reconciled!")
-            return DataFrame(), contract_match, no_match
-
-        dfs = []
-        for df in [contract_match, no_match]:
-            start_size = df.shape[0]
-            logger.debug(f"Report DF: \n{df}")
-            logger.debug(f"prev_rec: \n{prev_recs}")
-        
-            df = merge(
-                df,
-                prev_recs,
+            self.df = self._add_work_columns(self.df)
+            return
+        self.prev_rec = prev_recd
+
+        start_size = self.df.shape[0]
+        logger.debug(f"Report DF: \n{self.df}")
+        logger.debug(f"prev_rec: \n{prev_recd}")
+      
+        source_id = f"ID_{self.source}"
+        self.df[source_id] = self.df["ID"]
+        self.df = merge(
+            self.df,
+            prev_recd,
             how="left",
-                on= "Indentifier",
+            on= source_id,
             suffixes=("_cur", "_prev")
         )
-
-            df = HoldReport._created_combined_col("HideNextMonth", df, ["prev", "cur"])
-            df = HoldReport._created_combined_col("Resolution", df, ["prev", "cur"])
-            df["ID_OB"] = df["ID_OB_cur"]
-            df["ID_GP"] = df["ID_GP_cur"]
+        #self.df.to_excel(f"merged_df_{self.source}.xlsx")
         
         # Drop anything that should be ignored
-            df = df[df["HideNextMonth"] != True]
-            logger.info(f"Prev res added:\n{df}")
+        self.df = self.df[self.df["Hide Next Month"] != True]
+        logger.info(f"Prev res added:\n{self.df}")
 
         col_to_drop = []
-            for c in df.keys().to_list():
-                if "_prev" in c in c or "_cur" in c:
+        for c in self.df.keys().to_list():
+            logger.debug(f"{c=}")
+            if "_prev" in c or "ID_" in c:
+                logger.debug(f"Found '_prev' in {c}")
                 col_to_drop.append(c)
-                
+            else:
+                logger.debug(f"{c} is a good col!")
+        #col_to_drop.extend([c for c in self.df.keys().to_list() if '_prev' in c])
         logger.debug(f"{col_to_drop=}")
-            df.drop(
+        self.df.drop(
             columns= col_to_drop,
             inplace=True
         )
         # Restandardize
-            end_size = df.shape[0]
+        self.df.rename(columns={"contract_number_cur": "contract_number"}, inplace=True)
+        end_size = self.df.shape[0]
         logger.info(f"Reduced df by {start_size-end_size}")
-            dfs.append(df)
-        return prev_recs, dfs[0], dfs[1]
 
     def _remove_full_matches(self, other: 'HoldReport'):
         """
@@ -152,7 +111,7 @@ class HoldReport(ABC):
         other.df: DataFrame = other.df[~(other.df["ID"].isin(self.df["ID"]))]
         self.df = filter_id_match
         self.combined_missing: DataFrame = concat([self.df, other.df], ignore_index=True)
-        #self.combined_missing.to_excel("ALL MISSING.xlsx")
+        self.combined_missing.to_excel("ALL MISSING.xlsx")
         logger.debug(f"Combined Missing:\n{self.combined_missing}")
         logger.info(f"Payments with errors: {self.combined_missing.shape[0]}")
 
@@ -168,7 +127,7 @@ class HoldReport(ABC):
         return target_df
            
 
-    def _requires_rec(self,  other: 'HoldReport') -> tuple[DataFrame, DataFrame]:
+    def _requires_rec(self,  other: 'HoldReport') -> DataFrame:
         """
         To be run after full matches have been re
         """
@@ -181,11 +140,9 @@ class HoldReport(ABC):
             suffixes=('_'+self.source, '_'+other.source)
         )
 
-        contract_match = create_identifier(contract_match)
-
         #contract_match.to_excel("CONTRACT_MATCH.xlsx")
 
-        for col in ["vendor_name", "HideNextMonth", "Resolution"]:
+        for col in ["vendor_name", "Resolution", "Notes"]:
             self._created_combined_col(col, contract_match, (self.source, other.source)) 
 
         logger.debug(f"_requires_rec | contract_match:\n{contract_match.columns} ({contract_match.shape})")
@@ -202,10 +159,7 @@ class HoldReport(ABC):
             row["ID"] if row["Source"] == other.source else None                                          
         , axis=1)
 
-        no_match = create_identifier(no_match)
-
         logger.debug(f"_requires_rec | no_match:\n{no_match.columns} ({no_match.shape})")
-        self.prev_recs, contract_match, no_match = self._remove_prev_recs(contract_match, no_match)
 
         return contract_match, no_match      
 
@@ -222,28 +176,19 @@ class HoldReport(ABC):
                 df[col] = ''
         return df
 
-    def reconcile(self, other: 'HoldReport') -> ReconciledReports:
+    def reconcile(self, other: 'HoldReport') -> tuple[DataFrame]:
         """
         """
-        assert self.source != other.source, f"Reports to reconcile must be from different sources.\
-            ({self.source} , {other.source})."
         self._remove_full_matches(other)
-        
-        if self.source == "OB":
-            over_due: DataFrame = self.overdue
-            filtered_gp: DataFrame = other.filtered
-        elif self.source == "GP":
-            over_due: DataFrame = other.overdue
-            filtered_gp: DataFrame = self.filtered
-        
+        all_prev_reced = concat([self.prev_rec, other.prev_rec],ignore_index=True)
         logger.debug(f"Removed matches:\n{self.df}")
 
+        
         amount_mismatch, no_match = self._requires_rec(other)
 
         logger.debug(f"reconcile | no_match unaltered\n{no_match.columns} ({no_match.shape})")
         logger.debug(f"reconcile | am_mm unaltered:\n{amount_mismatch.columns} ({amount_mismatch.shape})")
         
-        # Formatting
         columns: list[str] = ["ID_GP", "ID_OB"]
         columns.extend(self.config["output_columns"])
 
@@ -264,36 +209,19 @@ class HoldReport(ABC):
         ]
         logger.info(f"no_match: {no_match.shape[0]}")
         logger.info(f"am_mm: {amount_mismatch.shape[0]}")
-
-        
-
-        reconciled: ReconciledReports = ReconciledReports(
-            no_match=no_match,
-            amt_mismatch=amount_mismatch,
-            prev_rec=self.prev_recs,
-            gp_filtered=filtered_gp,
-            ob_overdue = over_due
-        )
-        return reconciled
+        return no_match, amount_mismatch
     
 
 class OnBaseReport(HoldReport):
 
     source = "OB"
 
-    def __init__(self, dataframe: DataFrame, reports_config: dict) -> None:
-        self.overdue = self._get_overdue(dataframe)
-        super().__init__(dataframe, reports_config)
-
-    @staticmethod
-    def _get_overdue(dataframe: DataFrame) -> DataFrame:
+    def get_overdue(self) -> DataFrame:
         """
         """
-        dataframe["InstallDate"] = to_datetime(dataframe["InstallDate"])
-        dataframe["InstallDate"].fillna(NaT, inplace=True)
-        overdue: DataFrame = dataframe[dataframe["InstallDate"].dt.date\
-            < datetime.date.today()]
-        return  overdue
+        self.df["InstallDate"] = to_datetime(self.df["InstallDate"])
+        self.df["InstallDate"].fillna(NaT, inplace=True)
+        return  self.df[self.df["InstallDate"].dt.date < datetime.date.today()]
 
 
 class GreatPlainsReport(HoldReport):
@@ -302,7 +230,7 @@ class GreatPlainsReport(HoldReport):
 
     def __init__(self, dataframe: DataFrame, report_config: dict) -> None:
 
-        self.filtered: DataFrame = self._filter(
+        self._filter(
             gp_report_df= dataframe,
             doc_num_filters= report_config["gp_filters"]["doc_num_filters"],
             good_po_num_regex=  report_config["gp_filters"]["po_filter"]
@@ -311,8 +239,7 @@ class GreatPlainsReport(HoldReport):
 
     @staticmethod
     def _filter(gp_report_df: DataFrame, 
-                doc_num_filters: list[str], good_po_num_regex: str
-        ) -> DataFrame:
+                doc_num_filters: list[str], good_po_num_regex: str) -> DataFrame:
 
         GOOD_PO_NUM = re.compile(good_po_num_regex, re.IGNORECASE)
 
@@ -330,15 +257,15 @@ class GreatPlainsReport(HoldReport):
         )
 
         # Get the rows that DO NOT fit the keep_mask
-        dropped_posotives: DataFrame = gp_report_df[~keep_mask]
+        rows_to_drop = gp_report_df[~keep_mask].index
         # Drop the rows to filter
-        gp_report_df.drop(dropped_posotives.index, inplace=True)
+        gp_report_df.drop(rows_to_drop, inplace=True)
 
         # Create a filter to remove rows that meet this requirment
         # Making this a negative in the keep mask is more trouble than
         # it's worth
         remove_mask = gp_report_df["Document Number"].str.contains(bad_doc_num)
-        dropped_negatives: DataFrame = gp_report_df[remove_mask]
-        gp_report_df.drop(dropped_negatives.index, inplace=True)
+        rows_to_drop = gp_report_df[remove_mask].index
+        gp_report_df.drop(rows_to_drop, inplace=True)
 
-        return concat([dropped_posotives,dropped_negatives], ignore_index=False)
+        return gp_report_df

src/__init__.py

@@ -1,6 +0,0 @@
-from typing import TypeVar, Literal
-from enum import Enum
-
-class ReportSource(Enum):
-    OB = "OB"
-    GP = "GP"

src/config.py

@@ -1,186 +0,0 @@
-from tomllib import load as t_load
-from json import load as j_load
-from pathlib import Path
-from dataclasses import dataclass
-from typing import TypedDict
-from re import Pattern, compile
-
-from src import ReportSource
-
-
-Regex = str | Pattern
-
-
-class ReportConfigError(Exception):
-    """
-    Exception stemming from a report configuration
-    """
-    pass
-
-
-class SharedColumn(TypedDict, total=True):
-    """
-    Excel/Dataframe column that is shared between both GP & OB
-    """
-    standard: str
-    gp: str
-    ob: str
-
-
-class PathsConfig:
-    """
-    Configuration holding the paths to:
-     - input_directory: Where to search for new report files
-     - gp/ob_glob: regex used to find new OB & GP files in the report location
-     - db_path: path to an SQLite database if any
-    """
-        
-    def __init__(self, in_dir: str, out_dir: str, 
-        input_regex_dict: dict[str:str] , db_path: str = None) -> None:
-
-        self.input_directory: Path  = Path(in_dir)
-        self.output_directory: Path = Path(out_dir)
-
-        self.gp_glob: str = r"*.xlsx"
-        self.ob_glob: str = r"*.xlsx"
-
-        if db_path is not None:
-            self.db_path: Path = Path(db_path)
-        
-        try: 
-            self.gp_glob: str = input_regex_dict["GP"]
-            self.ob_glob: str = input_regex_dict["OB"]
-        except KeyError:
-            # Defaulting to newest of any xlsx file!
-            # TODO investigate warning
-            pass # will remain as *.xlsx
-
-    def get_most_recent(self, report_type: ReportSource = None) -> Path|None| tuple[Path|None, Path|None]:
-        
-        report_files = []
-        report_types = [ReportSource.OB, ReportSource.GP] if report_type is None else [report_type]
-        rt: ReportSource
-        for rt in report_types:
-            match rt:
-                case rt.OB:
-                    file_glob: str = self.ob_glob
-                case rt.GP:
-                    file_glob: str = self.gp_glob
-                case _:
-                    raise NotImplementedError(\
-                        f"No regex pattern for report type: {rt}"
-                    )
-
-            files = self.input_directory.glob(file_glob)
-            
-            # Find the most recently created file
-            most_recent_file = None
-            most_recent_creation_time = None
-
-            file: Path
-            for file in files:
-                creation_time = file.stat().st_ctime
-                if most_recent_creation_time is None or creation_time > most_recent_creation_time:
-                    most_recent_file = file
-                    most_recent_creation_time = creation_time
-            report_files.append(most_recent_file)
-
-        if len(report_files) > 1:
-            return report_files
-        
-        return report_files[0]           
-    
-    def has_database(self) -> tuple[bool, bool]:
-        """
-        Returns whether the config has a SQlite database path and
-        whether that path exists
-        """
-        has_db: bool = isinstance(self.db_path, Path)
-        exists: bool = self.db_path.exists() if has_db else False
-        return has_db, exists
-
-
-@dataclass
-class ReportConfig:
-
-    # Paths to work with
-    # - input/output
-    # - input discovery regexes
-    # - SQLite database path
-    paths: PathsConfig 
-
-    use_mssql: bool
-
-    # Work columns are included in finsished columns
-    work_columns: list[str]
-    finished_columns: list[str]
-
-    filters: dict[str:list[Pattern]|Pattern]
-
-    # Columns featured in both reports
-    # unified col name -> origin report -> origin col name
-    # e.g. contract_number -> GP -> Transaction Description
-    shared_columns: list[SharedColumn]
-
-    @staticmethod
-    def from_file(config_path: str|Path) -> 'ReportConfig':
-
-        config_path = Path(config_path) if isinstance(config_path, str) else config_path
-       
-        with open(config_path, "rb") as config_file:
-            match config_path.suffix:
-                case ".toml":
-                    c_dict: dict = t_load(config_file)
-                case ".json":
-                    c_dict: dict= j_load(config_file)
-                case _:
-                    raise NotImplementedError(f"Only json and toml configs are supported not: {config_path.suffix}")
-        
-        try:
-
-            path_config: PathsConfig = PathsConfig(
-                in_dir = c_dict["input_directory"],
-                out_dir= c_dict["output_directory"],
-                input_regex_dict= c_dict["input_glob_pattern"],
-                db_path= c_dict["database_path"]
-            )
-            
-            use_mssql = False #TODO no yet implemented
-
-            work_columns = c_dict["work_columns"]
-            finished_column = c_dict["finished_column"]
-            
-            # Create filter dict with compiled regex
-            filters_dict : dict = c_dict["filters"]
-            filters: dict[str:list[Pattern]|Pattern] =  {}
-            k: str
-            v: Regex|list[Regex]
-            for k, v in filters_dict.items():
-
-                if not isinstance(v, Regex) and not isinstance(v, list):
-                    raise ReportConfigError(f"Filter items must be a valid regex pattern or a list of valid patterns!\
-                        {v} ({type(v)}) is not valid!")
-                
-                # Convert the strings to regex patterns
-                if isinstance(v, list):
-                    filters[k] = [
-                        r if isinstance(r, Pattern)
-                        else compile(r)
-                        for r in v 
-                    ]
-                else:
-                    filters[k] = compile(v) if isinstance(v, Pattern) else v
-               
-            shared_columns: list[SharedColumn] = c_dict["shared_columns"]
-        
-        except KeyError as ke:
-            raise ReportConfigError(f"Invalid report config!\n{ke}")
-
-        return ReportConfig(
-            paths= path_config,
-            use_mssql= use_mssql,
-            work_columns= work_columns,
-            finished_columns= finished_column,
-            filters= filters,
-            shared_columns= shared_columns,
-        )

src/configs/report_config_template.json

@@ -1,33 +0,0 @@
-{
-  "input_directory": "/path/to/input/folder",
-  "input_glob_pattern": {
-    "GP": "*GP*.xlsx",
-    "OB": "*OB*.xlsx"
-  },
-  "output_directory": "/path/to/output",
-  "interactive_inputs": false,
-  "use_mssql": false,
-  "database_path": "./onhold.db",
-  "work_columns": [
-    "Col_A",
-    "Col_B"
-  ],
-  "finished_column": [
-    "Notes",
-    "Conctract Number"
-  ],
-  "filters": {
-    "filter_name": [
-      "\\d{7}",
-      "\\w+"
-    ],
-    "other_filter": "(OB|GP)$"
-  },
-  "shared_columns": [
-    {
-      "standardized_name": "contract_number",
-      "GP": "Transactoin Description",
-      "OB": "ContractNumber"
-    }
-  ]
-}

src/configs/reports_config.toml

@@ -1,72 +0,0 @@
-####  Paths: using '' makes the string 'raw' to avoid escape characters
-
-# Path to the directory to search for input report files
-input_directory = '../Reports'
-# Regex used to discover newest files
-input_glob_pattern = { GP = "*GP*.xlsx", OB = '*OB*.xlsx'}
-# Path to the directory to save the reconcilation work report
-output_directory = '../Output'
-# Fallback to interactive?
-interactive_inputs = false # NOT YET IMPLEMENTED
-
-
-#### DB
-
-# Whether to try using a mssql database
-# NOT YET IMPLEMENTED!
-use_mssql = false
-# Path to the SQLite database used to view/save reconcilations
-database_path = './onhold_reconciliation.db'
-
-
-### Finished rec details
-
-# Columns to add to all 'work' sheets
-# also saved 'Reconcilations' database
-work_columns = [
-    "HideNextMonth", # Boolean column for user to indicate if this contract should be ignored next month
-    "Resolution" # Text field describing the disprecany and how it may be resolved
-]
-# Columns to keep on reconcilation 'work' sheets
-finished_column = [
-        "contract_number",
-        "vendor_name",
-        "AppNum",           # OB only
-        "Document Number",  # GP Only
-        "DateBooked",       # OB only
-        "Document Date",    # GP Only
-        # 'Source' added for 'no match'
-    ]
-
-# Any regex filters that might be needed 
-[filters]
-# Use label to distinguish a regex set
-doc_num_filters = [
-        "p(oin)?ts",
-        "pool",
-        "promo",
-        "o(ver)?f(und)?",
-        "m(ar)?ke?t",
-        "title",
-        "adj",
-        "reg fee",
-        "rent",
-        "cma"
-    ]
-po_filter = ["^(?!.*cma(\\s|\\d)).*$"]
-
-# Columns that are featured & expected on both OB & GP
-[[shared_columns]]
-standardized_name = "contract_number" # The name you'd like to use to standardize them
-GP = "Transaction Description" # Column name used in GP
-OB = "Contract" # Column name used in GP
-
-[[shared_columns]]
-standardized_name = "onhold_amount"
-GP = "Current Trx Amount"
-OB = "CurrentOnHold"
-
-[[shared_columns]]
-standardized_name = "vendor_name" 
-GP = "Vendor Name" 
-OB = "DealerName"

src/configs/reports_config_template.toml

@@ -1,40 +0,0 @@
-####  Paths: using '' makes the string 'raw' to avoid escape characters
-
-# Path to the directory to search for input report files
-input_directory = '/path/to/input/folder'
-# Regex used to discover newest files
-input_glob_pattern = { GP = "*GP*.xlsx", OB = '*OB*.xlsx'}
-# Path to the directory to save the reconcilation work report
-output_directory = '/path/to/output'
-# Fallback to interactive?
-interactive_inputs = false # NOT YET IMPLEMENTED
-
-
-#### DB
-
-# Whether to try using a mssql database
-# NOT YET IMPLEMENTED!
-use_mssql = false
-# Path to the SQLite database used to view/save reconcilations
-database_path = './onhold.db'
-
-
-### Finished rec details
-
-# Columns to add to all 'work' sheets
-# also saved 'Reconcilations' database
-work_columns = ["Col_A", "Col_B" ]
-# Columns to keep on reconcilation 'work' sheets
-finished_column = [ "Notes", "Conctract Number" ]
-
-# Any regex filters that might be needed 
-[filters]
-# Use label to distinguish a regex set
-filter_name = [ '\d{7}', '\w+']
-other_filter = '(OB|GP)$'
-
-# Columns that are featured & expected on both OB & GP
-[[shared_columns]]
-standardized_name = "contract_number" # The name you'd like to use to standardize them
-GP = "Transactoin Description" # Column name used in GP
-OB = "ContractNumber" # Column name used in GP

tests/context.py

@@ -1,5 +0,0 @@
-import os
-import sys
-sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
-
-import src

tests/test_config.py

@@ -1,73 +0,0 @@
-import unittest
-from pathlib import Path
-from re import Pattern, compile
-from .context import src
-from src import config
-from src import ReportSource
-
-class TestReportConfig(unittest.TestCase):
-
-    def test_from_file(self):
-        # Provide the path to your config file
-        config_file = Path(r"tests\test_inputs\TEST_reports_config.toml")
-
-        # Call the static method from_file to create an instance of ReportConfig
-        report_config = config.ReportConfig.from_file(config_file)
-
-        # Assert the values of the attributes in the created instance
-        self.assertEqual(report_config.paths.input_directory, Path(r"tests\test_inputs"))
-        self.assertEqual(report_config.paths.gp_glob, r'*GP*.xlsx')
-        self.assertEqual(report_config.paths.ob_glob, r"*OB*.xlsx")
-        self.assertEqual(report_config.paths.output_directory, Path(r"tests\test_outputs"))
-        self.assertEqual(report_config.use_mssql, False)
-        self.assertEqual(report_config.paths.db_path, Path("./onhold_reconciliation.db"))
-        self.assertEqual(report_config.work_columns, ["HideNextMonth", "Resolution"])
-        self.assertEqual(report_config.finished_columns, [
-            "contract_number",
-            "vendor_name",
-            "AppNum",
-            "Document Number",
-            "DateBooked",
-            "Document Date",
-        ])
-        self.assertEqual(report_config.filters["doc_num_filters"], [
-            compile(r"p(oin)?ts",),
-            compile(r"pool",),
-            compile(r"promo",),
-            compile(r"o(ver)?f(und)?",),
-            compile(r"m(ar)?ke?t",),
-            compile(r"title",),
-            compile(r"adj",),
-            compile(r"reg fee",),
-            compile(r"rent",),
-            compile(r"cma",),
-        ])
-        self.assertEqual(report_config.filters["po_filter"], [compile(r"^(?!.*cma(\s|\d)).*$")])
-        self.assertEqual(report_config.shared_columns[0]["standardized_name"], "contract_number")
-        self.assertEqual(report_config.shared_columns[0]["GP"], "Transaction Description")
-        self.assertEqual(report_config.shared_columns[0]["OB"], "Contract")
-        self.assertEqual(report_config.shared_columns[1]["standardized_name"], "onhold_amount")
-        self.assertEqual(report_config.shared_columns[1]["GP"], "Current Trx Amount")
-        self.assertEqual(report_config.shared_columns[1]["OB"], "CurrentOnHold")
-        self.assertEqual(report_config.shared_columns[2]["standardized_name"], "vendor_name")
-        self.assertEqual(report_config.shared_columns[2]["GP"], "Vendor Name")
-        self.assertEqual(report_config.shared_columns[2]["OB"], "DealerName")
-
-    def test_get_newest(self):
-        # Provide the path to your config file
-        config_file = Path(r"tests\test_inputs\TEST_reports_config.toml")
-
-        # Call the static method from_file to create an instance of ReportConfig
-        report_config = config.ReportConfig.from_file(config_file)
-
-        newest_ob: Path = report_config.paths.get_most_recent(report_type=ReportSource.OB)
-        self.assertEqual(newest_ob.name, "April 2023 OB.xlsx")
-        newest_gp: Path = report_config.paths.get_most_recent(report_type=ReportSource.GP)
-        self.assertEqual(newest_gp.name, "April GP.xlsx")
-
-        nob, ngp = report_config.paths.get_most_recent()
-        self.assertEqual(nob.name, "April 2023 OB.xlsx")
-        self.assertEqual(ngp.name, "April GP.xlsx")
-
-if __name__ == '__main__':
-    unittest.main()

tests/test_inputs/TEST_reports_config.toml

@@ -1,72 +0,0 @@
-####  Paths: using '' makes the string 'raw' to avoid escape characters
-
-# Path to the directory to search for input report files
-input_directory = 'tests\test_inputs'
-# Regex used to discover newest files
-input_glob_pattern = { GP = "*GP*.xlsx", OB = '*OB*.xlsx'}
-# Path to the directory to save the reconcilation work report
-output_directory = 'tests\test_outputs'
-# Fallback to interactive?
-interactive_inputs = false # NOT YET IMPLEMENTED
-
-
-#### DB
-
-# Whether to try using a mssql database
-# NOT YET IMPLEMENTED!
-use_mssql = false
-# Path to the SQLite database used to view/save reconcilations
-database_path = './onhold_reconciliation.db'
-
-
-### Finished rec details
-
-# Columns to add to all 'work' sheets
-# also saved 'Reconcilations' database
-work_columns = [
-    "HideNextMonth", # Boolean column for user to indicate if this contract should be ignored next month
-    "Resolution" # Text field describing the disprecany and how it may be resolved
-]
-# Columns to keep on reconcilation 'work' sheets
-finished_column = [
-        "contract_number",
-        "vendor_name",
-        "AppNum",           # OB only
-        "Document Number",  # GP Only
-        "DateBooked",       # OB only
-        "Document Date",    # GP Only
-        # 'Source' added for 'no match'
-    ]
-
-# Any regex filters that might be needed 
-[filters]
-# Use label to distinguish a regex set
-doc_num_filters = [
-        "p(oin)?ts",
-        "pool",
-        "promo",
-        "o(ver)?f(und)?",
-        "m(ar)?ke?t",
-        "title",
-        "adj",
-        "reg fee",
-        "rent",
-        "cma"
-    ]
-po_filter = ['^(?!.*cma(\s|\d)).*$']
-
-# Columns that are featured & expected on both OB & GP
-[[shared_columns]]
-standardized_name = "contract_number" # The name you'd like to use to standardize them
-GP = "Transaction Description" # Column name used in GP
-OB = "Contract" # Column name used in GP
-
-[[shared_columns]]
-standardized_name = "onhold_amount"
-GP = "Current Trx Amount"
-OB = "CurrentOnHold"
-
-[[shared_columns]]
-standardized_name = "vendor_name" 
-GP = "Vendor Name" 
-OB = "DealerName"

version.txt

@@ -1 +1 @@
-2.1
+2.0