feat(output): add drop_secondary_tail flag for external records

This commit is contained in:
2026-09-13 14:58:08 +08:00
parent 4b275075c9
commit 27b8c5ed97
2 changed files with 39 additions and 2 deletions
+4 -2
View File
@@ -125,7 +125,8 @@ def output_json(records, outpath, roles=None, fd_fields=None, field_to_fd=None,
def output_input_files(records, outdir, stem, roles, fd_fields, field_to_fd, open_dir, def output_input_files(records, outdir, stem, roles, fd_fields, field_to_fd, open_dir,
term_types=None, data_fields=None, select_info=None): term_types=None, data_fields=None, select_info=None,
drop_secondary_tail=True):
term_types = term_types or ['normal'] * len(records) term_types = term_types or ['normal'] * len(records)
input_fds = {} input_fds = {}
for fd_name, fds_set in fd_fields.items(): for fd_name, fds_set in fd_fields.items():
@@ -162,7 +163,8 @@ def output_input_files(records, outdir, stem, roles, fd_fields, field_to_fd, ope
normals.append(fd_rec) normals.append(fd_rec)
# 丢弃次要 FD 的最后 2 条记录,触发 EOF/不匹配路径 # 丢弃次要 FD 的最后 2 条记录,触发 EOF/不匹配路径
if fd_idx > 0 and normals: # (黑盒外部数据需保留全部记录,传 drop_secondary_tail=False
if drop_secondary_tail and fd_idx > 0 and normals:
normals = normals[:-2] normals = normals[:-2]
if normals: if normals:
+35
View File
@@ -46,3 +46,38 @@ def test_output_input_files_basic():
with tempfile.TemporaryDirectory() as tmp: with tempfile.TemporaryDirectory() as tmp:
output_input_files(records, tmp, "TESTPGM", roles, {}, {}, {}) output_input_files(records, tmp, "TESTPGM", roles, {}, {}, {})
assert os.path.isdir(tmp) assert os.path.isdir(tmp)
def test_output_input_files_keeps_secondary_tail_when_disabled():
"""黑盒模式: drop_secondary_tail=False 时次要 FD 不砍尾。"""
records = [{"WS-A": str(i)} for i in range(5)]
roles = {"WS-A": "input"}
fd_fields = {"F1": {"WS-A"}, "F2": {"WS-A"}}
field_to_fd = {"WS-A": "F1"}
open_dir = {"F1": "INPUT", "F2": "INPUT"}
select_info = {"F1": {"assign": "IN1"}, "F2": {"assign": "IN2"}}
data_fields = [{"name": "WS-A", "pic": "X(1)",
"pic_info": {"type": "alphanumeric", "length": 1}}]
with tempfile.TemporaryDirectory() as tmp:
output_input_files(records, Path(tmp), "T", roles, fd_fields, field_to_fd,
open_dir, data_fields=data_fields, select_info=select_info,
drop_secondary_tail=False)
data = json.loads((Path(tmp) / "T_F2.json").read_text(encoding="utf-8"))
assert len(data) == 5
def test_output_input_files_drops_secondary_tail_by_default():
"""默认行为不变: 次要 FD 砍掉最后 2 条。"""
records = [{"WS-A": str(i)} for i in range(5)]
roles = {"WS-A": "input"}
fd_fields = {"F1": {"WS-A"}, "F2": {"WS-A"}}
field_to_fd = {"WS-A": "F1"}
open_dir = {"F1": "INPUT", "F2": "INPUT"}
select_info = {"F1": {"assign": "IN1"}, "F2": {"assign": "IN2"}}
data_fields = [{"name": "WS-A", "pic": "X(1)",
"pic_info": {"type": "alphanumeric", "length": 1}}]
with tempfile.TemporaryDirectory() as tmp:
output_input_files(records, Path(tmp), "T", roles, fd_fields, field_to_fd,
open_dir, data_fields=data_fields, select_info=select_info)
data = json.loads((Path(tmp) / "T_F2.json").read_text(encoding="utf-8"))
assert len(data) == 3