hangshuo652
106 lines
2.7 KiB
Python
106 lines
2.7 KiB
Python
from data.diff_result import FieldResult
|
|
from decimal import Decimal, InvalidOperation, ROUND_DOWN
|
|
|
|
DEFAULT_TOLERANCE = 0.01
|
|
|
|
|
|
def compare_field(
|
|
name: str,
|
|
cobol_val: str,
|
|
java_val: str,
|
|
field_type: str = "decimal",
|
|
tolerance: float = DEFAULT_TOLERANCE
|
|
) -> FieldResult:
|
|
result = FieldResult(field_name=name,
|
|
cobol_value=cobol_val,
|
|
java_value=java_val)
|
|
|
|
if field_type in ("decimal", "numeric"):
|
|
return _compare_numeric(result, cobol_val, java_val, tolerance)
|
|
|
|
if field_type == "date":
|
|
return _compare_date(result, cobol_val, java_val)
|
|
|
|
if field_type in ("string", "alpha"):
|
|
return _compare_string(result, cobol_val, java_val)
|
|
|
|
return _compare_generic(result, cobol_val, java_val)
|
|
|
|
|
|
def _compare_numeric(fr: FieldResult, c: str, j: str, tol: float) -> FieldResult:
|
|
c_val = _parse_number(c)
|
|
j_val = _parse_number(j)
|
|
|
|
if c_val is None and j_val is None:
|
|
fr.status = "PASS"
|
|
return fr
|
|
|
|
if c_val is None:
|
|
fr.status = "NOT_SET"
|
|
fr.suggestion = "cobol_parse_error"
|
|
return fr
|
|
|
|
if j_val is None:
|
|
fr.status = "MISMATCH"
|
|
fr.suggestion = "java_missing_init: null/None where COBOL has value"
|
|
return fr
|
|
|
|
if c_val == j_val:
|
|
fr.status = "PASS"
|
|
return fr
|
|
|
|
diff = abs(c_val - j_val)
|
|
if isinstance(c_val, Decimal):
|
|
diff = abs(float(c_val - j_val))
|
|
|
|
if diff <= tol:
|
|
fr.status = "TOLERATED"
|
|
fr.tolerance_applied = tol
|
|
else:
|
|
fr.status = "MISMATCH"
|
|
|
|
return fr
|
|
|
|
|
|
def _compare_date(fr: FieldResult, c: str, j: str) -> FieldResult:
|
|
c_norm = _normalize_date(c)
|
|
j_norm = _normalize_date(j)
|
|
fr.status = "PASS" if c_norm == j_norm else "MISMATCH"
|
|
return fr
|
|
|
|
|
|
def _compare_string(fr: FieldResult, c: str, j: str) -> FieldResult:
|
|
c_clean = c.strip() if c else ""
|
|
j_clean = j.strip() if j else ""
|
|
fr.status = "PASS" if c_clean == j_clean else "MISMATCH"
|
|
return fr
|
|
|
|
|
|
def _compare_generic(fr: FieldResult, c: str, j: str) -> FieldResult:
|
|
fr.status = "PASS" if c == j else "MISMATCH"
|
|
return fr
|
|
|
|
|
|
def _parse_number(val: str):
|
|
if val is None or val == "None":
|
|
return None
|
|
s = str(val).strip()
|
|
if s in ("", "\x00", "\x00\x00\x00\x00\x00"):
|
|
return Decimal("0")
|
|
s = s.replace("\x00", "")
|
|
try:
|
|
return Decimal(s)
|
|
except InvalidOperation:
|
|
return None
|
|
|
|
|
|
def _normalize_date(val: str, default: str = "1970-01-01") -> str:
|
|
if not val:
|
|
return default
|
|
s = val.strip()
|
|
if len(s) == 8 and s.isdigit():
|
|
return f"{s[0:4]}-{s[4:6]}-{s[6:8]}"
|
|
if len(s) == 10 and s[4] == '-':
|
|
return s
|
|
return s
|