You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

Python追加写入JSON文件读取报错,求可更新存储方案

问题根源

用a模式追加JSON时,每次json.dump会输出一段完整的JSON字符串,多次执行后文件里会出现多个首尾相连的JSON结构,而json.load只能解析单个合法的JSON对象/数组,所以触发JSONDecodeError。

可行解决方案

方案1:JSON Lines格式(逐行存储单个JSON对象)

这是最适合追加场景的JSON变种,每行存一个独立的JSON对象,既支持追加写入,也能逐行解析读取。

写入(支持追加)

每次把单个任务转成JSON字符串后加换行写入:

import json

def append_task(task):
    with open('tasks.jsonl', 'a', encoding='utf8') as f:
        json.dump(task, f)
        f.write('\n')  # 每行一个JSON对象

# 测试追加
task1 = ["1Name", "1c:/path/file.ext", "1Some other text", 10, ["a", "b", "c"], ""]
task2 = ["2Name", "2c:/path/file.ext", "2Test text", 20, ["a", "b", "c"], ""]
append_task(task1)
append_task(task2)

读取

逐行读取并解析每一行的JSON:

def read_tasks():
    tasks = []
    with open('tasks.jsonl', 'r', encoding='utf8') as f:
        for line in f:
            line = line.strip()
            if line:  # 跳过空行
                tasks.append(json.loads(line))
    return tasks

print(read_tasks())

删除数据

读取所有任务,过滤掉不需要的,再重新写入覆盖文件:

def delete_task(name):
    tasks = read_tasks()
    # 过滤掉指定名称的任务(这里假设第一个元素是名称)
    filtered_tasks = [t for t in tasks if t[0] != name]
    # 重新写入覆盖
    with open('tasks.jsonl', 'w', encoding='utf8') as f:
        for task in filtered_tasks:
            json.dump(task, f)
            f.write('\n')

# 测试删除名称为"1Name"的任务
delete_task("1Name")

方案2:读写完整JSON数组(适合数据量小的场景)

每次追加前先读取现有数组,把新数据合并进去,再用w模式覆盖写入,保证文件始终是一个合法的JSON数组。

import json

def append_to_json(new_data):
    try:
        # 先读取现有数据
        with open('test.txt', 'r', encoding='utf8') as f:
            existing_data = json.load(f)
    except FileNotFoundError:
        # 文件不存在则初始化空数组
        existing_data = []
    
    # 合并新数据(如果new_data是列表就extend,是单个元素就append)
    if isinstance(new_data, list):
        existing_data.extend(new_data)
    else:
        existing_data.append(new_data)
    
    # 覆盖写入
    with open('test.txt', 'w', encoding='utf8') as f:
        json.dump(existing_data, f, indent=2)

# 测试追加
new_tasks = [["1Name", "1c:/path/file.ext", "1Some other text", 10, ["a", "b", "c"], ""],
             ["2Name", "2c:/path/file.ext", "2Test text", 20, ["a", "b", "c"], ""]]
append_to_json(new_tasks)

# 读取函数不变
def read_json():
    with open('test.txt', 'r', encoding='utf8') as f:
        return json.load(f)

print(read_json())

方案3:用SQLite数据库(适合频繁增删改查的场景)

如果数据量较大、需要复杂查询或频繁修改,直接用Python内置的SQLite数据库更省心,不用自己处理文件格式问题。

import sqlite3
import json

# 初始化数据库表
def init_db():
    conn = sqlite3.connect('tasks.db')
    cursor = conn.cursor()
    # 创建表:id自增,name、path、desc、num、tags(用JSON存列表)、extra
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS tasks (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            name TEXT,
            path TEXT,
            desc TEXT,
            num INTEGER,
            tags TEXT,
            extra TEXT
        )
    ''')
    conn.commit()
    conn.close()

# 插入任务
def add_task(name, path, desc, num, tags, extra):
    conn = sqlite3.connect('tasks.db')
    cursor = conn.cursor()
    # 把tags列表转成JSON字符串存储
    cursor.execute('''
        INSERT INTO tasks (name, path, desc, num, tags, extra)
        VALUES (?, ?, ?, ?, ?, ?)
    ''', (name, path, desc, num, json.dumps(tags), extra))
    conn.commit()
    conn.close()

# 查询所有任务
def get_all_tasks():
    conn = sqlite3.connect('tasks.db')
    cursor = conn.cursor()
    cursor.execute('SELECT * FROM tasks')
    tasks = []
    for row in cursor.fetchall():
        # 把tags的JSON字符串转回列表
        task = list(row[1:])  # 跳过id
        task[4] = json.loads(task[4])
        tasks.append(task)
    conn.close()
    return tasks

# 删除任务(按名称)
def delete_task_by_name(name):
    conn = sqlite3.connect('tasks.db')
    cursor = conn.cursor()
    cursor.execute('DELETE FROM tasks WHERE name = ?', (name,))
    conn.commit()
    conn.close()

# 测试使用
init_db()
add_task("1Name", "1c:/path/file.ext", "1Some other text", 10, ["a", "b", "c"], "")
add_task("2Name", "2c:/path/file.ext", "2Test text", 20, ["a", "b", "c"], "")
print(get_all_tasks())
delete_task_by_name("1Name")
print(get_all_tasks())

内容的提问来源于stack exchange,提问作者Eitan

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.06.24 21:00:37