Python并行脚本通过JSON共享数据触发JSONDecodeError报错

json.decoder.JSONDecodeError: Expecting value: line 1 column 1 (char 0) 触发原因是Selenium脚本读取JSON文件时，文件内容为空，不符合JSON语法规范。
问题出在跨脚本读写文件没有做并发保护：常规写文件逻辑是先清空目标文件原有内容，再逐字节写入新内容，整个过程不是瞬时完成的。Selenium脚本轮询时只要检测到文件变动（此时刚好是文件被清空、新内容还没写入的时间点）就立刻读取，必然读到空文件触发解析崩溃，和你写的JSON内容本身对错没关系。

分两端改，改完不管是现在的测试脚本还是后续接Flask都能稳定跑。

1. 所有写JSON的逻辑改用原子写入 #

不管是你现在用来测试的定时写文件脚本，还是后续开发的Flask webhook接收脚本，不要直接以w模式打开目标JSON文件写入，先写临时文件，写完再整体替换目标文件。这个替换操作是系统级原子的，读文件的脚本永远只能读到完整的旧内容或者完整的新内容，碰不到空文件、写了一半的残缺文件。
原子写的参考实现：

import json
import os
import tempfile

def save_trade_signal(signal: dict, target_file: str = "trade_signal.json"):
    # 先把数据序列化为合法JSON字符串
    json_str = json.dumps(signal, ensure_ascii=False)
    file_dir = os.path.dirname(os.path.abspath(target_file))
    # 在目标文件同目录创建临时文件，避免跨分区替换失败
    with tempfile.NamedTemporaryFile(
        mode="w",
        dir=file_dir,
        delete=False,
        encoding="utf-8"
    ) as tmp_fp:
        tmp_fp.write(json_str)
        tmp_fp.flush()
        os.fsync(tmp_fp.fileno()) # 强制把缓存内容刷入磁盘，避免内容丢失
        tmp_file_path = tmp_fp.name
    # 原子替换目标文件
    os.replace(tmp_file_path, target_file)

2. Selenium轮询侧加容错和重复校验 #

轮询逻辑不要检测到文件变动就直接解析，加两层防护：一是解析逻辑加异常捕获，碰到非法JSON直接跳过本轮等待下次检测，不要直接崩溃；二是加内容哈希校验，避免重复处理同一个信号，也能过滤掉无效的空内容。
轮询逻辑参考实现：

import json
import os
import time
import hashlib

SIGNAL_FILE = "trade_signal.json"
POLL_INTERVAL = 0.2 # 200毫秒轮询一次，响应足够快，CPU占用也低
last_processed_md5 = None

# 初始化读取已有文件，避免启动后重复处理旧信号
if os.path.exists(SIGNAL_FILE):
    with open(SIGNAL_FILE, "r", encoding="utf-8") as fp:
        init_content = fp.read()
    last_processed_md5 = hashlib.md5(init_content.encode("utf-8")).hexdigest()

while True:
    time.sleep(POLL_INTERVAL)
    if not os.path.exists(SIGNAL_FILE):
        continue
    try:
        # 先读全量文件内容
        with open(SIGNAL_FILE, "r", encoding="utf-8") as fp:
            content = fp.read()
        # 空内容直接跳过
        if not content.strip():
            continue
        # 计算内容哈希，和已处理的哈希比对，相同说明是老信号直接跳过
        current_md5 = hashlib.md5(content.encode("utf-8")).hexdigest()
        if current_md5 == last_processed_md5:
            continue
        # 解析JSON
        trade_signal = json.loads(content)
        # 以下替换成你自己的Selenium下单逻辑
        print("收到新交易信号，执行下单：", trade_signal)
        
        # 处理完成后更新已处理标记
        last_processed_md5 = current_md5
    except json.JSONDecodeError:
        # 碰到JSON解析错误直接等下一轮，不用终止程序
        continue
    except Exception as e:
        # Selenium操作等其他异常打日志后重试，不要让循环退出
        print(f"执行异常，1秒后重试：{str(e)}")
        time.sleep(1)
        continue

补充说明 #

你之前尝试Flask和Selenium整合到单脚本跑不通是正常的，Flask默认启动的是单线程阻塞服务，启动后会阻塞主线程，Selenium的逻辑要么没法正常运行，要么和Flask的请求上下文冲突。现在双脚本解耦的方案稳定性更高，后续排查问题、单独重启某一部分都更方便，不用强行改单脚本。

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