edi_reader/edi_parser.py at main · bodharma/edi_reader · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
# MIT License
# Copyright (c) 2026 bodharma
# https://github.com/bodharma/edi_reader

import sys
from pprint import pprint
from template import template
from loguru import logger


class EDI:
    def __init__(self, data):
        self.element_delimiter = None
        self.subelement_delimiter = None
        self.segment_delimiter = None
        self.data = data.split('\n')

    def find_segment_position(self, segment):
        positions_list = [self.data.index(el) for el in self.data if f'{segment}' in el]
        if len(positions_list) == 1:
            position = positions_list[-1]
        else:
            logger.debug(f"We have {len(positions_list)} of {segment} segments in the file. \nParsing not implemented\nExiting...")
            sys.exit(0)
        return position

    def isa_parser(self):
        isa = template.ISA.copy()
        out = isa.copy()
        isa_position = self.find_segment_position('ISA')

        isa_segment = self.data[isa_position]

        self.element_delimiter = isa_segment[-3:-2]
        self.subelement_delimiter = isa_segment[-2:-1]
        self.segment_delimiter = isa_segment[-1]

        isa_element_list = isa_segment.split(self.element_delimiter)
        for key, value in isa.items():
            out[key] = isa_element_list[value]

        return out

    def gs_parser(self):
        gs_list = []
        gs_position = self.find_segment_position('GS')

        gs_segment = self.data[gs_position]
        gs_segment_list = gs_segment.split(self.segment_delimiter)
        # TODO: this should be moved out to splitter function
        for segment in gs_segment_list:
            if len(segment) > 0:
                gs_element_list = segment.split(self.element_delimiter)

                user_defined_elements_list = []
                qualifiers_list = []

                for element in gs_element_list:
                    if len(element) == 2:
                        if element == "ZZ":
                            logger.debug("User defined")
                            user_defined_elements_list.append(element)
                        else:
                            logger.debug(f"{element} is qualifier, read the standarts guide to get more info ")
                            qualifiers_list.append(element)


                gs_document_type = gs_element_list[1]
                gs_sender = gs_element_list[2]
                gs_receiver = gs_element_list[3]
                gs_version = gs_element_list[-1]
                gs_id = gs_element_list[6]
                gs_list.append({
                    "gs_document_type": gs_document_type,
                    "gs_sender": gs_sender,
                    "gs_receiver": gs_receiver,
                    "segment_id": gs_id,
                    "gs_version": gs_version,
                    "qualifiers_list": qualifiers_list,
                    "user_defined_elements_list": user_defined_elements_list
                })
            else:
                continue
        return gs_list

    def st_parser(self):
        st_position = self.find_segment_position('ST')
        st_segment = self.data[st_position]
        return st_segment

    def ge_parser(self):
        ge_position = self.find_segment_position('GE')
        ge_segment = self.data[ge_position]
        ge_segment_list = ge_segment.split(self.segment_delimiter)
        parsed_segment_list = []
        #TODO: this should be moved out to splitter function
        for segment in ge_segment_list:
            if len(segment) > 0:
                ge_element_list = segment.split(self.element_delimiter)
                ge_id = ge_element_list[2]
                parsed_segment_list.append(
                    {
                        "segment_id": ge_id
                    }
                )
            else:
                continue
        return parsed_segment_list

    def iea_parser(self):
        iea_position = self.find_segment_position('IEA')
        iea_segment = self.data[iea_position]
        iea_segment_list = iea_segment.split(self.segment_delimiter)
        parsed_segment_list = []
        # TODO: this should be moved out to splitter function
        for segment in iea_segment_list:
            if len(segment) > 0:
                iea_element_list = segment.split(self.element_delimiter)
                parsed_segment_list.append(iea_element_list)
            else:
                continue
        return parsed_segment_list

    def st_se_generator(self):
        return {"ST-SE data": self.st_parser()}

    def gs_ge_generator(self):
        gs_ge_template = {
            "GS data": self.gs_parser(),
            "GE data": self.ge_parser(),
        }
        gs_ge_template.update(self.st_se_generator())
        return {"GS-GE data": gs_ge_template}

    def isa_iea_generator(self):
        isa_iea_template = {
            "ISA data": self.isa_parser(),
            "IEA data": self.iea_parser(),
        }
        isa_iea_template.update(self.gs_ge_generator())
        return isa_iea_template


if __name__ == '__main__':
    f = open('EDI Test.txt', 'r')
    data = f.read()

    edi = EDI(data)
    out = edi.isa_iea_generator()
    pprint(out)