Python两文件循环传递JSON数据实现方案咨询

第一步：修正 pythonone.py，实现逐个生成并输出JSON #

原代码存在两个核心问题：随机数据仅初始化一次，导致循环内所有JSON内容完全重复；循环结束后才输出一次，无法实现逐个传递。修改后的代码如下：

import json
import time
from faker import Faker

fake = Faker()

def myrandomdata(min_val, max_val):
    return fake.random_int(min=min_val, max=max_val)

def main():
    for i in range(5):
        # 每次循环重新生成随机数据，保证JSON内容唯一
        random_ins_id = str(myrandomdata(20000, 10000000))
        random_inv_item_id = str(myrandomdata(20000, 10000000))
        random_inv_org_id = str(myrandomdata(1000, 100000))
        random_loc_id = str(myrandomdata(20000, 100000))
        qty = myrandomdata(1, 100)
        loc_type_id = myrandomdata(0, 4)
        
        # 构造当前循环的JSON对象
        json_obj = {
            'ID': random_ins_id,
            'QTY': qty,
            'EXT_REF': random_loc_id,
            'INV_ITEM_ID': random_inv_item_id,
            'ORG_ID': random_inv_org_id,
            'SERIAL_NUMBER': loc_type_id
        }
        
        # 转换为JSON字符串并立即输出，确保内容实时传递
        print(json.dumps(json_obj))
        time.sleep(3)

if __name__ == "__main__":
    main()

第二步：编写 pythontwo.py，实现逐个接收并处理JSON #

从标准输入逐行读取pythonone.py输出的JSON，解析后处理，直到输入流结束（即pythonone.py循环完成）：

import json
import sys

def process_my_data(json_data):
    # 这里替换为你的实际处理逻辑，示例仅打印关键字段
    print(f"已处理：ID={json_data['ID']}, 数量={json_data['QTY']}")

def main():
    # 逐行读取输入内容，每次处理一个JSON对象
    for line in sys.stdin:
        line = line.strip()
        if not line:
            continue
        try:
            json_obj = json.loads(line)
            process_my_data(json_obj)
        except json.JSONDecodeError:
            print(f"跳过无效JSON：{line}", file=sys.stderr)

if __name__ == "__main__":
    main()

第三步：运行方式 #

在命令行中使用管道连接两个脚本，实现输出与输入的实时逐个传递：

python pythonone.py | python pythontwo.py

运行后，pythonone.py每3秒生成一个JSON并输出，pythontwo.py会立即接收并处理，直到pythonone.py的5次循环完成后自动终止。

内容的提问来源于stack exchange，提问作者ditil