hangshuo652
90 lines
2.7 KiB
Python
90 lines
2.7 KiB
Python
from dataclasses import dataclass
|
|
|
|
|
|
EBCDIC_TO_ASCII = {
|
|
0xC1: 'A', 0xC2: 'B', 0xC3: 'C', 0xC4: 'D', 0xC5: 'E',
|
|
0xC6: 'F', 0xC7: 'G', 0xC8: 'H', 0xC9: 'I', 0xD1: 'J',
|
|
0xD2: 'K', 0xD3: 'L', 0xD4: 'M', 0xD5: 'N', 0xD6: 'O',
|
|
0xD7: 'P', 0xD8: 'Q', 0xD9: 'R', 0xE2: 'S', 0xE3: 'T',
|
|
0xE4: 'U', 0xE5: 'V', 0xE6: 'W', 0xE7: 'X', 0xE8: 'Y',
|
|
0xE9: 'Z', 0xF0: '0', 0xF1: '1', 0xF2: '2', 0xF3: '3',
|
|
0xF4: '4', 0xF5: '5', 0xF6: '6', 0xF7: '7', 0xF8: '8',
|
|
0xF9: '9', 0x40: ' ', 0x4B: '.', 0x6B: ',', 0x5A: '!',
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class CobolIRField:
|
|
raw_hex: str
|
|
decoded_value: str
|
|
encoding: str
|
|
field_type: str
|
|
length: int
|
|
scale: int
|
|
signed: bool
|
|
|
|
|
|
@dataclass
|
|
class JavaIRField:
|
|
raw_value: str
|
|
decoded_value: str
|
|
field_type: str
|
|
nullable: bool
|
|
|
|
|
|
@dataclass
|
|
class IRRecord:
|
|
field_name: str
|
|
cobol: CobolIRField | None = None
|
|
java: JavaIRField | None = None
|
|
|
|
|
|
class Normalizer:
|
|
def normalize_encoding(self, raw: bytes, encoding: str) -> str:
|
|
if encoding == "EBCDIC":
|
|
return self._ebcdic_to_ascii(raw)
|
|
return raw.decode("ascii", errors="replace")
|
|
|
|
def normalize_comp3(self, raw: bytes) -> str:
|
|
if not raw:
|
|
return "0"
|
|
nibbles = []
|
|
for b in raw:
|
|
nibbles.append((b >> 4) & 0x0F)
|
|
nibbles.append(b & 0x0F)
|
|
sign = nibbles.pop()
|
|
value = 0
|
|
for n in nibbles:
|
|
value = value * 10 + n
|
|
if sign in (0x0D, 0x0B):
|
|
value = -value
|
|
return str(value)
|
|
|
|
def normalize_date(self, date_str: str) -> str:
|
|
s = date_str.strip()
|
|
if len(s) == 8 and s.isdigit():
|
|
return f"{s[0:4]}-{s[4:6]}-{s[6:8]}"
|
|
return s
|
|
|
|
def to_ir_record(self, field_name, raw_hex, decoded_value,
|
|
encoding, field_type, length=0, scale=0, signed=False) -> IRRecord:
|
|
return IRRecord(
|
|
field_name=field_name,
|
|
cobol=CobolIRField(
|
|
raw_hex=raw_hex, decoded_value=decoded_value,
|
|
encoding=encoding, field_type=field_type,
|
|
length=length, scale=scale, signed=signed))
|
|
|
|
def to_null_ir(self, field_name, side="java") -> IRRecord:
|
|
if side == "java":
|
|
return IRRecord(field_name=field_name,
|
|
cobol=None, java=JavaIRField(raw_value="", decoded_value="", field_type="null", nullable=True))
|
|
return IRRecord(field_name=field_name,
|
|
cobol=None, java=JavaIRField(raw_value="", decoded_value="", field_type="null", nullable=True))
|
|
|
|
def _ebcdic_to_ascii(self, raw: bytes) -> str:
|
|
result = []
|
|
for b in raw:
|
|
result.append(EBCDIC_TO_ASCII.get(b, chr(b) if 32 <= b < 127 else '?'))
|
|
return ''.join(result)
|