code refactor

Author: codereverser

casparser/__init__.py

@@ -1,8 +1,4 @@
-try:
-    from .parsers.mupdf import read_cas_pdf
-except (ImportError, ModuleNotFoundError):
-    from .parsers.pdfminer import read_cas_pdf
-
+from .parsers import read_cas_pdf
 from .types import CASParserDataType
 from .__version__ import __version__
 

casparser/cli.py

@@ -7,10 +7,7 @@
 
 from .__version__ import __version__
 
-try:
-    from .parsers.mupdf import read_cas_pdf
-except ImportError:
-    from .parsers.pdfminer import read_cas_pdf
+from . import read_cas_pdf
 from .encoder import CASDataEncoder
 from .exceptions import ParserException
 
@@ -109,12 +106,7 @@ def cli(output, summary, password, force_pdfminer, filename):
         click.echo("No output file provided. Printing summary")
         summary = True
     try:
-        if force_pdfminer:
-            from .parsers.pdfminer import read_cas_pdf as read_cas_pdf_pm
-
-            data = read_cas_pdf_pm(filename, password)
-        else:
-            data = read_cas_pdf(filename, password)
+        data = read_cas_pdf(filename, password, force_pdfminer=force_pdfminer)
     except ParserException as exc:
         click.echo("Error parsing pdf file :: " + click.style(str(exc), bold=True, fg="red"))
         sys.exit(1)

casparser/parsers/__init__.py

@@ -0,0 +1,38 @@
+import io
+import json
+from typing import Union
+
+from casparser.encoder import CASDataEncoder
+from casparser.process import process_cas_text
+
+
+def read_cas_pdf(filename: Union[str, io.IOBase], password, output="dict", force_pdfminer=False):
+    """
+    Parse CAS pdf and returns line data.
+
+    :param filename: CAS pdf file (CAMS or Kfintech)
+    :param password: CAS pdf password
+    :param output: Output format (json,dict)  [default: dict]
+    :param force_pdfminer: Force pdfminer parser even if mupdf is detected
+    """
+    if force_pdfminer:
+        from .pdfminer import cas_pdf_to_text
+    else:
+        try:
+            from .mupdf import cas_pdf_to_text
+        except (ImportError, ModuleNotFoundError):
+            from .pdfminer import cas_pdf_to_text
+
+    partial_cas_data = cas_pdf_to_text(filename, password)
+
+    processed_data = process_cas_text("\u2029".join(partial_cas_data.lines))
+    # noinspection PyProtectedMember
+    processed_data.update(
+        {
+            "file_type": partial_cas_data.file_type.name,
+            "investor_info": partial_cas_data.investor_info._asdict(),
+        }
+    )
+    if output == "dict":
+        return processed_data
+    return json.dumps(processed_data, cls=CASDataEncoder)

casparser/parsers/mupdf.py

@@ -1,20 +1,14 @@
-from collections import namedtuple
 import io
-import json
 from operator import itemgetter
 import re
 from typing import List, Iterator, Union, Any
 
 # noinspection PyPackageRequirements
 import fitz
 
-from casparser.encoder import CASDataEncoder
 from casparser.enums import FileType
 from casparser.exceptions import CASParseError
-from casparser.process import process_cas_text
-from .utils import isclose
-
-InvestorInfo = namedtuple("InvestorInfo", ["name", "email", "address", "mobile"])
+from .utils import isclose, InvestorInfo, PartialCASData
 
 
 def extract_blocks(page_dict):
@@ -138,14 +132,13 @@ def group_similar_rows(elements_list: List[Iterator[Any]]):
     return lines
 
 
-def read_cas_pdf(filename: Union[str, io.IOBase], password, output="dict"):
+def cas_pdf_to_text(filename: Union[str, io.IOBase], password) -> PartialCASData:
     """
     Parse CAS pdf and returns line data.
 
     :param filename: CAS pdf file (CAMS or Kfintech)
     :param password: CAS pdf password
-    :param output: Output format (json,dict)  [default: dict]
-    :return: array of lines from the CAS.
+    :return: partial cas data with FileType, InvestorInfo and lines of data
     """
     file_type: FileType = FileType.UNKNOWN
 
@@ -183,13 +176,4 @@ def read_cas_pdf(filename: Union[str, io.IOBase], password, output="dict"):
                 investor_info = parse_investor_info(page_dict)
             pages.append(sorted_blocks)
         lines = group_similar_rows(pages)
-        processed_data = process_cas_text("\u2029".join(lines))
-        processed_data.update(
-            {
-                "file_type": file_type.name,
-                "investor_info": investor_info._asdict(),
-            }
-        )
-        if output == "dict":
-            return processed_data
-        return json.dumps(processed_data, cls=CASDataEncoder)
+        return PartialCASData(file_type=file_type, investor_info=investor_info, lines=lines)

casparser/parsers/pdfminer.py

@@ -1,6 +1,4 @@
-from collections import namedtuple
 import io
-import json
 import re
 from typing import List, Optional, Iterator, Union
 
@@ -12,13 +10,9 @@
 from pdfminer.pdfpage import PDFPage
 from pdfminer.layout import LTTextBoxHorizontal, LTTextBoxVertical
 
-from casparser.encoder import CASDataEncoder
 from casparser.enums import FileType
 from casparser.exceptions import CASParseError
-from casparser.process import process_cas_text
-from .utils import isclose
-
-InvestorInfo = namedtuple("InvestorInfo", ["name", "email", "address", "mobile"])
+from .utils import isclose, InvestorInfo, PartialCASData
 
 
 def parse_investor_info(layout, width, height) -> InvestorInfo:
@@ -103,13 +97,12 @@ def group_similar_rows(elements_list: List[Iterator[LTTextBoxHorizontal]]):
     return lines
 
 
-def read_cas_pdf(filename: Union[str, io.IOBase], password, output="dict"):
+def cas_pdf_to_text(filename: Union[str, io.IOBase], password) -> PartialCASData:
     """
     Parse CAS pdf and returns line data.
 
     :param filename: CAS pdf file (CAMS or Kfintech)
     :param password: CAS pdf password
-    :param output: Output format (json,dict)  [default: dict]
     :return: array of lines from the CAS.
     """
     file_type: Optional[FileType] = None
@@ -159,14 +152,4 @@ def read_cas_pdf(filename: Union[str, io.IOBase], password, output="dict"):
             pages.append(text_elements)
 
         lines = group_similar_rows(pages)
-        processed_data = process_cas_text("\u2029".join(lines))
-        processed_data.update(
-            {
-                "file_type": file_type.name,
-                "investor_info": investor_info._asdict(),
-            }
-        )
-
-        if output == "dict":
-            return processed_data
-        return json.dumps(processed_data, cls=CASDataEncoder)
+        return PartialCASData(file_type=file_type, investor_info=investor_info, lines=lines)

casparser/parsers/utils.py

@@ -1,3 +1,9 @@
+from collections import namedtuple
+
+InvestorInfo = namedtuple("InvestorInfo", ["name", "email", "address", "mobile"])
+PartialCASData = namedtuple("PartialCASData", ["file_type", "investor_info", "lines"])
+
+
 def isclose(a0, a1, tol=1.0e-4):
     """
     Check if two elements are almost equal with a tolerance.

tests/base.py

@@ -3,6 +3,7 @@
 
 import pytest
 
+from casparser import read_cas_pdf
 from casparser.exceptions import CASParseError
 
 
@@ -19,12 +20,8 @@ def setup_class(cls):
         cls.kfintech_password = os.getenv("KFINTECH_CAS_PASSWORD")
 
     def read_pdf(self, filename, password, output="dict"):
-        if self.mode == "pdfminer":
-            from casparser.parsers.pdfminer import read_cas_pdf
-        else:
-            from casparser.parsers.mupdf import read_cas_pdf
-
-        return read_cas_pdf(filename, password, output=output)
+        use_pdfminer = (self.mode == "pdfminer")
+        return read_cas_pdf(filename, password, output=output, force_pdfminer=use_pdfminer)
 
     def test_read_dict(self):
         self.read_pdf(self.cams_file_name, self.cams_password)