[fix] Added handling of incomplete transactions (#3)

Signed-off-by: Aman Jagotra <aman.jagotra@beyondirr.tech>

Author: AmanJagotra

cas2json/__init__.py

@@ -1,4 +1,4 @@
-from cas2json.cams.processor import parse_cams_pdf
+from cas2json.cams import parse_cams_pdf
 from cas2json.nsdl.processor import parse_nsdl_pdf
 
 __all__ = ["parse_cams_pdf", "parse_nsdl_pdf"]

cas2json/cams/__init__.py

@@ -0,0 +1,52 @@
+import io
+
+from cas2json.cams.processor import CASProcessor
+from cas2json.enums import FileType, FileVersion
+from cas2json.exceptions import CASParseError
+from cas2json.parser import cas_pdf_to_text
+from cas2json.types import CASData
+
+
+def parse_cams_pdf(filename: str | io.IOBase, password: str, sort_transactions=True) -> CASData:
+    """
+    Parse CAMS or KFintech CAS pdf and returns processed data.
+
+    Parameters
+    ----------
+    filename : str | io.IOBase
+        The path to the PDF file or a file-like object.
+    password : str
+        The password to unlock the PDF file.
+    sort_transactions : bool
+        Whether to sort transactions by date and re-compute balances.
+    """
+
+    partial_cas_data = cas_pdf_to_text(filename, password)
+    if partial_cas_data.file_type not in [FileType.CAMS, FileType.KFINTECH]:
+        raise CASParseError("Not a valid CAMS file")
+
+    if partial_cas_data.file_version == FileVersion.DETAILED:
+        processed_data = CASProcessor().process_detailed_version(partial_cas_data.document_data)
+    elif partial_cas_data.file_version == FileVersion.SUMMARY:
+        processed_data = CASProcessor().process_summary_version(partial_cas_data.document_data)
+    else:
+        raise CASParseError("Unknown CAS file type")
+
+    if sort_transactions:
+        for scheme in processed_data.schemes:
+            transactions = scheme.transactions
+            sorted_transactions = sorted(transactions, key=lambda x: x.date)
+            if transactions != sorted_transactions:
+                balance = scheme.open
+                for transaction in sorted_transactions:
+                    balance += transaction.units or 0
+                    transaction.balance = balance
+                scheme.transactions = sorted_transactions
+
+    return CASData(
+        statement_period=processed_data.statement_period,
+        schemes=processed_data.schemes,
+        file_version=partial_cas_data.file_version,
+        investor_info=partial_cas_data.investor_info,
+        file_type=partial_cas_data.file_type,
+    )

cas2json/cams/helpers.py

@@ -1,17 +1,18 @@
+import logging
 import re
 from decimal import Decimal
 
-from dateutil import parser as date_parser
-
 from cas2json import patterns
-from cas2json.enums import CASFileType, TransactionType
-from cas2json.flags import MULTI_TEXT_FLAGS, TEXT_FLAGS
-from cas2json.types import TransactionData
-from cas2json.utils import formatINR
+from cas2json.constants import MISCELLANEOUS_KEYWORDS
+from cas2json.enums import TransactionType
+from cas2json.flags import TEXT_FLAGS
+
+logger = logging.getLogger(__name__)
+logger.setLevel(logging.DEBUG)
 
 
 def get_transaction_type(description: str, units: Decimal | None) -> tuple[TransactionType, Decimal | None]:
-    """Get transaction type from the description text."""
+    """Get transaction type from the description text and units."""
 
     description = description.lower()
     # Dividend
@@ -54,83 +55,17 @@ def get_transaction_type(description: str, units: Decimal | None) -> tuple[Trans
             return (TransactionType.SWITCH_OUT_MERGER if "merger" in description else TransactionType.SWITCH_OUT, None)
         return (TransactionType.REDEMPTION, None)
 
-    print("Warning: Error identifying transaction. Please report the issue with the transaction description")
-    print(f"Txn description: {description} :: Units: {units}")
-    return (TransactionType.UNKNOWN, None)
+    for keyword in MISCELLANEOUS_KEYWORDS:
+        if keyword in description:
+            return (TransactionType.MISC, None)
 
-
-def get_transaction_values(values: str) -> tuple[str | None, str | None, str | None, str | None]:
-    """
-    Extract transaction values in the order of amount, units, nav, and balance from the given string.
-    """
-    values = re.findall(patterns.AMT, values.strip())
-    units = nav = balance = amount = None
-    if len(values) >= 4:
-        # Normal entry
-        amount, units, nav, balance, *_ = values
-    elif len(values) == 3:
-        # Zero unit entry
-        amount, nav, balance = values
-        units = "0.000"
-    elif len(values) == 2:
-        # Segregated Portfolio Entries
-        units, balance = values
-    elif len(values) == 1:
-        # Tax entries
-        amount = values[0]
-    return amount, units, nav, balance
+    logger.warning(f"Error identifying transaction. Description: {description} :: Units: {units}")
+    return (TransactionType.UNKNOWN, None)
 
 
 def get_parsed_scheme_name(scheme: str) -> str:
+    """Helper to clean scheme names."""
     scheme = re.sub(r"\((formerly|erstwhile).+?\)", "", scheme, flags=TEXT_FLAGS).strip()
     scheme = re.sub(r"\((Demat|Non-Demat).*", "", scheme, flags=TEXT_FLAGS).strip()
     scheme = re.sub(r"\s+", " ", scheme).strip()
     return re.sub(r"[^a-zA-Z0-9_)]+$", "", scheme).strip()
-
-
-def detect_cas_type(parsed_lines: list[str]) -> CASFileType:
-    """Detect the type of CAS statement (detailed or summary) from the parsed lines."""
-    text = "\u2029".join(parsed_lines)
-    if m := re.search(patterns.CAS_TYPE, text, MULTI_TEXT_FLAGS):
-        match = m.group(1).lower().strip()
-        if match == "statement":
-            return CASFileType.DETAILED
-        elif match == "summary":
-            return CASFileType.SUMMARY
-    return CASFileType.UNKNOWN
-
-
-def parse_transaction(line: str) -> list[TransactionData]:
-    """
-    Parse a transaction line and return a list of TransactionData objects.
-    """
-    transactions: list[TransactionData] = []
-    parsed_transactions = re.findall(patterns.TRANSACTIONS, line, MULTI_TEXT_FLAGS)
-    if not parsed_transactions:
-        return transactions
-
-    for txn in parsed_transactions:
-        date, details, *_ = txn
-        if not details or not details.strip() or not date:
-            continue
-        description_match = re.match(patterns.DESCRIPTION, details.strip(), MULTI_TEXT_FLAGS)
-        if not description_match:
-            continue
-        description, values, *_ = description_match.groups()
-        amount, units, nav, balance = get_transaction_values(values)
-        description = description.strip()
-        units = formatINR(units)
-        txn_type, dividend_rate = get_transaction_type(description, units)
-        transactions.append(
-            TransactionData(
-                date=date_parser.parse(date).date(),
-                description=description,
-                type=txn_type.name,
-                amount=formatINR(amount),
-                units=units,
-                nav=formatINR(nav),
-                balance=formatINR(balance),
-                dividend_rate=dividend_rate,
-            )
-        )
-    return transactions