Support for summary output in csv format

Author: codereverser

casparser/cli.py

@@ -16,7 +16,7 @@
 from . import read_cas_pdf
 from .enums import CASFileType
 from .exceptions import ParserException
-from .parsers.utils import is_close, cas2json, cas2csv
+from .parsers.utils import is_close, cas2json, cas2csv, cas2csv_summary
 
 CONTEXT_SETTINGS = dict(help_option_names=["-h", "--help"])
 
@@ -95,9 +95,7 @@ def print_summary(data, tablefmt="fancy_grid", output_filename=None, include_zer
             if scheme["close"] < 1e-3 and not include_zero_folios:
                 continue
 
-            calc_close = scheme["open"] + sum(
-                [x["units"] for x in scheme["transactions"] if x["units"] is not None]
-            )
+            calc_close = scheme.get("close_calculated", "open")
             valuation = scheme["valuation"]
 
             # Check is calculated close (i.e. open + units from all transactions) is same as
@@ -230,8 +228,15 @@ def cli(output, summary, password, include_all, force_pdfminer, filename):
             include_zero_folios=include_all,
             output_filename=None if output_ext in (".csv", ".json") else output,
         )
+
     if output_ext in (".csv", ".json"):
-        conv_fn = cas2json if output_ext == ".json" else cas2csv
+        if output_ext == ".csv":
+            if summary or data["cas_type"] == CASFileType.SUMMARY.name:
+                conv_fn = cas2csv_summary
+            else:
+                conv_fn = cas2csv
+        else:
+            conv_fn = cas2json
         with open(output, "w", newline="", encoding="utf-8") as fp:
             fp.write(conv_fn(data))
         click.echo("File saved : " + click.style(output, bold=True))

casparser/parsers/utils.py

@@ -26,6 +26,51 @@ def cas2json(data: CASParserDataType) -> str:
     return json.dumps(data, cls=CASDataEncoder)
 
 
+def cas2csv_summary(data: CASParserDataType) -> str:
+    with io.StringIO() as csv_fp:
+        header = [
+            "amc",
+            "folio",
+            "advisor",
+            "registrar",
+            "pan",
+            "scheme",
+            "isin",
+            "amfi",
+            "open",
+            "close",
+            "value",
+            "date",
+            "transactions",
+        ]
+        writer = csv.DictWriter(csv_fp, fieldnames=header)
+        writer.writeheader()
+        current_amc = None
+        for folio in data.get("folios", []):
+            if current_amc != folio.get("amc", ""):
+                current_amc = folio["amc"]
+            for scheme in folio["schemes"]:
+                row = {
+                    "amc": current_amc.replace("\n", " "),
+                    "folio": folio["folio"],
+                    "advisor": scheme["advisor"],
+                    "registrar": scheme["rta"],
+                    "pan": folio["PAN"],
+                    "scheme": scheme["scheme"].replace("\n", " "),
+                    "isin": scheme["isin"],
+                    "amfi": scheme["amfi"],
+                    "open": scheme["open"],
+                    "close": scheme["close"],
+                    "value": scheme["valuation"]["value"],
+                    "date": scheme["valuation"]["date"],
+                    "transactions": len(scheme["transactions"]),
+                }
+                writer.writerow(row)
+        csv_fp.seek(0)
+        csv_data = csv_fp.read()
+        return csv_data
+
+
 def cas2csv(data: CASParserDataType) -> str:
     with io.StringIO() as csv_fp:
         header = [

casparser/process/cas_detailed.py

@@ -131,13 +131,15 @@ def process_detailed_text(text):
                     "amfi": amfi,
                     "open": Decimal(0.0),
                     "close": Decimal(0.0),
+                    "close_calculated": Decimal(0.0),
                     "valuation": {"date": None, "value": 0, "nav": 0},
                     "transactions": [],
                 }
         if not curr_scheme_data:
             continue
         if m := re.search(OPEN_UNITS_RE, line):
             curr_scheme_data["open"] = Decimal(m.group(1).replace(",", "_"))
+            curr_scheme_data["close_calculated"] = curr_scheme_data["open"]
             continue
         if m := re.search(CLOSE_UNITS_RE, line):
             curr_scheme_data["close"] = Decimal(m.group(1).replace(",", "_"))
@@ -165,6 +167,8 @@ def process_detailed_text(text):
                 nav = Decimal(m.group(5).replace(",", "_"))
                 balance = Decimal(m.group(6).replace(",", "_").replace("(", "-"))
             txn_type, dividend_rate = get_transaction_type(desc, units)
+            if units is not None:
+                curr_scheme_data["close_calculated"] += units
             curr_scheme_data["transactions"].append(
                 {
                     "date": date,

casparser/types.py

@@ -49,6 +49,7 @@ class SchemeType(TypedDict, total=False):
     amfi: Optional[str]
     open: Union[Decimal, float]
     close: Union[Decimal, float]
+    close_calculated: Union[Decimal, float]
     valuation: SchemeValuationType
     transactions: List[TransactionDataType]
 
@@ -72,4 +73,5 @@ class CASParserDataType(TypedDict):
     statement_period: StatementPeriod
     folios: List[FolioType]
     investor_info: InvestorInfoType
+    cas_type: str
     file_type: FileType

tests/test_mupdf.py

@@ -50,6 +50,22 @@ def test_cli(self, tmpdir):
         assert result.exit_code != 1
         assert "File saved" in result.output
 
+        fpath = tmpdir.join("output.csv")
+        result = runner.invoke(
+            cli,
+            [
+                self.kfintech_file_name,
+                "-p",
+                self.kfintech_password,
+                "-o",
+                fpath.strpath,
+                "-s",
+                "fancy_grid",
+            ],
+        )
+        assert result.exit_code != 1
+        assert "File saved" in result.output
+
         result = runner.invoke(
             cli, [self.kfintech_file_name, "-p", self.kfintech_password, "-s", "html"]
         )

tests/test_process.py

@@ -48,7 +48,9 @@ def test_transaction_type(self):
         )
 
     def test_isin_search(self):
-        isin, amfi = isin_search("Axis Long Term Equity Fund - Direct Growth", "KFINTECH" ,"128TSDGG")
+        isin, amfi = isin_search(
+            "Axis Long Term Equity Fund - Direct Growth", "KFINTECH", "128TSDGG"
+        )
         assert isin == "INF846K01EW2"
         assert amfi == "120503"