python报错runtimeerror:dictionary changed size during iteration

Question

这是最近写的库的一部分内容：

MoldelName = "tableOut"

import os,sys
import io
import re
import time
import random
#from colorama import Fore,Back,Style,coloramaInit

outOverwritten = False # 输出时是否覆盖输出,如否当出现覆盖时抛出异常
block_interval = 20 # 每格的横轴宽度(字符)
out_sep_time   = 0  # 每次刷新屏幕的时间间隔
autoFlush = True

"""flushType类型:
real 实时:
    实时刷新所有屏幕内容,会清空屏幕
    适合用于导出输出后查看
segm 分段:
    分段输出内容,较快且易于观察实时输出
    适合实时查看输出的情况
"""
flushType = "real"
flush_typeReal_pringList = []

stdout = {}

class FlushTypeError(Exception):
    ErrText = "\"flushType\" %s is not different."

class NoInitError(Exception):
    ErrText = f"No initialization has been performed, please call the \"init\" function to initialize {MoldelName}!"
    is_init = False

class InternalError(Exception):
    ErrText_stdout = "\tError in variable \"stdout\", unknown type \"%s\".\n\t\tPerhaps due to incorrect type input when customizing the \"stdout\" variable."

class OutputRepetition(Exception):
    ErrText = "Output overwritten, please increase timestamp accuracy or specify the 'outOverwritten' variable as True."

class Order:
    def __init__(self):
        """
        序号类型:
            stamp      时间戳:
             ┣time.time时间戳
             ┗如为这个类型,self.accuracy即为时间戳精度
            num        数字序列:
             ┣从self.num_start开始
             ┗如为这个类型,self.num_start即为序列起始序号
            rand       随机数:
             ┣random.randint随机数
             ┣如为这个类型,self.rand_range即为随机范围,
             ┗do_rand_no_repet即可使随机数不重复(见下文)
            randForNum 随机数:
             ┣同rand类型
             ┣如为这个类型,在rand类型的基础上(do_rand_no_repet不起效),
             ┗会在被重复使用的数字后添加序号
        """
        self.type = "stamp" # 序号类型0
        self.accuracy = 1 # 正整数代表小数点后位数，负数代表小数点前位数
        self.num_start = 1 # 序列起始序号
        self.rand_range = (1000,9999) # 随机范围
        """
        is_rand_down_do操作类型:
            raise    抛出错误
            whileNum 重新使用所有随机数并在后面添加序号
            while    重新使用所有随机数但不在后面添加序号
        """
        self.is_rand_down_do = "raise" # 当随机数用完时进行的操作
        self.do_rand_no_repet = True # 如为True可使随机数不重复,但当所有数字即将用完时会产生延时,用完后根据self.is_rand_down_do确定操作
        
        self.num_list = self.num_start
        self.rand_usenumlist = []
        self.rand_usedownnumdir = {}
        self.randForNum_usenumdir = {}
        
    def getOrder(self):
        if self.type == "stamp":
            return round(time.time(),self.accuracy)
        elif self.type == "num":
            self.num_list += 1
            return self.num_list - 1
        elif self.type == "rand":
            r = random.randint(*self.rand_range)
            if not self.do_rand_no_repet:
                return r
            else:
                while (r in self.rand_usenumlist):
                    if len(self.rand_usenumlist) >= len(list(range(*self.randrange))) + 1:
                        self.doRandDown()
                    r = random.randint(*self.rand_range)
                self.rand_usenumlist.append(r)
                if len(self.rand_usedownnumdir):
                    self.rand_usedownnumdir[r] += 1
                    r = f"{r} ({self.rand_usedownnumdir[r]})"
                return r
        elif self.type == "randForNum":
            r = random.randint(*self.rand_range)
            if r in self.randForNum_usenumdir:
                self.randForNum_usenumdir[r] += 1
            else:
                self.randForNum_usenumdir[r] = 1
            return f"{r} ({self.randForNum_usenumdir[r]})"
        else:
            return "No Orders"
    
    def doRandDown(self):
        if self.is_rand_down_do == "raise":
            class RandNumUsageError(Exception):ErrText = "Random numbers used up!"
            raise RandNumUsageError(RandNumUsageError.ErrText)
        elif self.is_rand_down_do == "whileNum":
            print("Wait for rand-whileNum Mod doing...",end = "")
            for i in self.randrange:
                self.rand_usedownnumdir[i] = 1
            print("\r",end = "")
        elif self.is_rand_down_do == "while":
            self.rand_usenumlist = []

order = Order()

class Std_:
    def __init__(self,type):
        self.type = type
    def write(self,text,):
        if NoInitError.is_init == False:
            raise NoInitError(NoInitError.ErrText)
        timestamp = order.getOrder()
        try:
            stdout[timestamp].append({"type":self.type,"text":text})
        except:
            stdout[timestamp] = [{"type":self.type,"text":text}]
        if autoFlush:
            flush()
class Stdinfo(Std_):pass
class Stderr(Std_):pass
class Stdwarn(Std_):pass
class Stddat(Std_):pass

stdinfo = Stdinfo("info")
stdwarn = Stdwarn("warn")
stderr  = Stderr("err")
stddat  = Stddat("dat")

class Stderr_take(io.TextIOBase):
    def __init__(self):
        self.content = ""
    def write(self,text):
        stderr.write(text)
stderr_take = Stderr_take()

#sys.stdout = Stdout()

class DataStat:
    def __init__(self):
        self.init_time = 0
        self.flush_average_time = 0
ds = DataStat()
def quit(info = True,tr = True):
    global os,sys,time,io,re,random,stdout
    if flushType == "segm":
        print("\r" + end)
    if info:
        print("╔" + ("═" * (interval + 4) * len(showlist)) + "╗\n║Info:" + (" " * ((interval + 4) * len(showlist) - 5)) + f"║\n║\tinit_time:{ds.init_time}" + (" " * ((interval + 4) * len(showlist) - 17 - len(str(ds.init_time)))) + f"║\n║\tflush_average_time:{ds.flush_average_time}" + (" " * ((interval + 4) * len(showlist) - 26 - len(str(ds.flush_average_time)))) + "║")
        print("╚" + ("═" * (interval + 4) * len(showlist)) + "╝\nWait for quit...")
    del os,sys,time,io,re,random,stdout
    if tr:
        return ds

def cls():
    os.system('cls' if os.name == 'nt' else 'clear')
    if out_sep_time:
        time.sleep(out_sep_time)

def cutstring(t,l):
    t = t.replace("\n","\\n").replace("\t","    ").replace("\r","\\r")
    tl = [t[i:i + l] for i in range(0,len(t),l)]
    if len(tl[-1]) < l:
        tl[-1] = tl[-1] + " " * (l - len(tl[-1]))
    return tl

def gettextblock(k,v):
    interval = block_interval
    text = [0 for _ in range(5)]
    if len(v) > len(showlist) and not outOverwritten:
        raise OutputRepetition(OutputRepetition.ErrText)
    for i in v:
        if "order" in showlist:
            text[0] = str(k) + " " * (interval - len(str(k)) + 3)
        if "info" in showlist and i["type"] == "info":
            text[1] = cutstring(i["text"],interval + 3)
        if "warn" in showlist and i["type"] == "warn":
            text[2] = cutstring(i["text"],interval + 3)
        if "err" in showlist and i["type"] == "err":
            text[3] = cutstring(i["text"],interval + 3)
        if "dat" in showlist and i["type"] == "dat":
            text[4] = cutstring(i["text"],interval + 3)
        if i["type"] not in showlist:
            raise InternalError(InternalError.ErrText_stdout % i["type"])
    #print(text)
    
    """
    textmaxline = max(
                   (len(text[1]) if type(text[1]) == list else 0),
                   (len(text[2]) if type(text[2]) == list else 0),
                   (len(text[3]) if type(text[3]) == list else 0),
                   (len(text[4]) if type(text[4]) == list else 0)
                  )
    """
    #优化后的textmaxline算法
    textmaxline = max((len(i) if type(i) == list else 0) for i in text[1:])
    #修复bug，使没有输出的格中填充空格
    text = [([" " * (interval + 3) for _ in range(textmaxline)] if i == 0 else i) for i in text]
    #print(textmaxline)
    blocktext = []
    #return
    for i in range(textmaxline):
        blocktext_il = [text[0]] if not i else [" " * (interval + 3) for _ in range(textmaxline)]
        for o in range(1,5):
            #if type(text[o]) == list:
                if len(text[o]) > i:
                    blocktext_il.append(text[o][i])
        # 修复第二行后前面多出一列的bug(对timestamp及每行有多个显示内容时基本无效)
        if len(blocktext_il) >= len(showlist):
            blocktext_il = blocktext_il[len(blocktext_il) - len(showlist):]
        blocktext.append(blocktext_il)
        del blocktext_il# 释放资源
    
    timestamp_recut = cutstring(blocktext[0][0],interval + 4)
    for i in range(len(timestamp_recut)):
        if len(blocktext) >= i:
            blocktext[i][0] = timestamp_recut[i]
    
    if len(blocktext) > len(timestamp_recut):
        for i in range(len(timestamp_recut),len(blocktext)):
            blocktext[i][0] = " " * (interval + 4)
    #print(blocktext)
    #return
    del text,timestamp_recut # 释放资源
    
    #print("blocktext:")
    #print(blocktext)
    #return
    text = "\n".join(["┃" + "│".join(i) + "┃" for i in blocktext])
    #print("text:")
    #print(text)
    return text

def flush(segmType_k = None):
    stdoutl = stdout
    if NoInitError.is_init == False:
        raise NoInitError(NoInitError.ErrText)
    flush_start = time.time()
    if flushType == "real":
        text = []
        for k,v in stdoutl.items():
            text.append(gettextblock(k,v))
        cls()
        print("\n".join([head,*("".join([i,("\n" if text.index(i) != len(text) - 1 else ""),(sep if text.index(i) != len(text) - 1 else "")]) for i in text),end]))
    elif flushType == "segm":
        if "title" not in flush_typeReal_pringList:
            cls()
            print(head,end = "")
            flush_typeReal_pringList.append("title")
        else:
            if segmType_k:
                print("\n" + gettextblock(segmType_k,stdoutl[segmType_k]) + "\n" + sep,end = "")
                flush_typeReal_pringList.append(segmType_k)
            for k,v in stdoutl.items():
                if k not in flush_typeReal_pringList:
                    print("\n" + gettextblock(k,v) + "\n" + sep,end = "")
                    flush_typeReal_pringList.append(k)
    else:
        raise FlushTypeError(FlushTypeError.ErrText % flushType)
    if ds.flush_average_time:
        ds.flush_average_time = (ds.flush_average_time + time.time() - flush_start) / 2
    else:
        ds.flush_average_time = time.time() - flush_start
"""
当datwrite之前的没有输出，dat会输出到前面而不是最后一列
"""


def init(timeshow = True,info = True,err = True,warn = True,dat = True,takeStderr = True):
    global end,head,sep,showlist,interval
    if NoInitError.is_init:return
    
    if takeStderr:
        global Sys_stderr
        Sys_stderr = sys.stderr
        sys.stderr = stderr_take
    
    init_start = time.time()
    interval = block_interval
    fit = "┏"
    title = ""
    sep = "┣"
    end = "┗"
    showlist = []
    titletextlen = []
    titlelen = 0
    if timeshow:
        titlelen += 1
        titletextlen.append(5)
        showlist.append("order")
        title += "order" + " " * (interval - 1)
    if info:
        titlelen += 1
        titletextlen.append(4)
        showlist.append("info")
        title += "│info" + " " * (interval - 1)
    if warn:
        titlelen += 1
        titletextlen.append(4)
        showlist.append("warn")
        title += "│warn" + " " * (interval - 1)
    if err:
        titlelen += 1
        titletextlen.append(4)
        showlist.append("err")
        title += "│err " + " " * (interval - 1)
    if dat:
        titlelen += 1
        titletextlen.append(3)
        showlist.append("dat")
        title += "│dat " + " " * (interval - 1)
    
    for i in range(titlelen):
        fit += "━" * (interval - 1 + titletextlen[i]) + ("┯" if titlelen - 1 != i else "━")
    fit += "┓"
    
    for i in range(titlelen):
        sep += "━" * (interval - 1 + titletextlen[i]) + ("┿" if titlelen - 1 != i else "━")
    sep += "┫"
    
    for i in range(titlelen):
        end += "━" * (interval - 1 + titletextlen[i]) + ("┷" if titlelen - 1 != i else "━")
    end += "┛"
    
    head = "".join([fit,"\n┃",title,"┃\n",sep])
    #coloramaInit()
    cls()
    print(head + "\n" + end,end = "\n")
    
    NoInitError.is_init = True
    ds.init_time = time.time() - init_start

"""
# 测试用
stdout = {"1":[
               {"type":"info","text":"1infotext..........." * 2},
               {"type":"warn","text":"1warntext..........." * 2},
               {"type":"err","text":"1errtext............" * 2}
              ],
          "2":[
               {"type":"info","text":"2infotext..........." * 2},
               {"type":"warn","text":"2warntext..........." * 2},
               {"type":"dat","text":"2dattext............" * 2}
              ]
         }
"""
#block_interval = 11
"""
stdout[3] = [{"type":"info","text":"adtjkvd" * 3},
             {"type":"dat","text":"adtjkvd" * 3}]
"""
if __name__ == "__main__":
    init()
    order.accuracy = 6
    stderr.write("nihao" * 2)
    stdinfo.write("nihao " * 5)
    flush()
"""
win = tk.Tk()
win.title("Python_Std")

win.mainloop()"""

这是调用时的文件：

import tableOut as psr,random as r
from threading import Thread as td
psr.order.accuracy = 4
psr.order.type = "num"
psr.order.rand_range = (10,10)
psr.block_interval = 20
#psr.outOverwritten = True
#psr.autoFlush = False
psr.out_sep_time = 0.001
psr.flushType = "segm"
psr.init(takeStderr = 1)
def getstr():
    t = ""
    for i in range(r.randint(10,29)):
        t += chr(r.randint(65,122))
    return t
raisevar = 0
def raiseErr():
    global raisevar
    raisevar += 1
    t = 1 / 0
    print(t)
for i in range(30):
    # 17.5s
    psr.stdinfo.write("infowrite " + str(i) + getstr())
    #psr.stdwarn.write("warnwrite " + str(i) + getstr())
    #psr.stddat.write("datwrite " + str(i) + getstr())
    psr.stderr.write("errwrite" + str(i) + getstr())
    #psr.flush()
    if i == 1:
        ttd = td(target = raiseErr)
        ttd.start()
    """
    print("infowrite " + str(i) + getstr())
    print("warnwrite " + str(i) + getstr())
    print("datwrite " + str(i) + getstr())
    print("errwrite" + str(i) + getstr())"""

psr.quit()
print(raisevar)

该调用的文件是想测试将sys.stderr重定向到自定义类并写入stdout字典，可是报错runtimeerror:dictionary changed size during iteration，请问如何解决该问题？（我尝试在flush函数中使用临时字典可是还是报错，由于重定向到自定义类了，但是该类报错导致错误在程序最后才输出了部分内容，没有报错行数及位置，虽然已知flush函数没有问题，所以不知道哪里有问题）

Answer 1

引自免费微信小程序：皆我百晓生

你遇到的问题是因为在迭代字典的同时试图修改了字典的大小，这在Python中是不允许的。具体来说，当你尝试在遍历字典的过程中添加或删除键值对时，就会触发这个错误。在你的代码中，这个问题出现在flush函数中，你试图在迭代stdout字典的同时修改了它的内容（通过添加新的键值对）。

解决这个问题的一个方法是使用字典的copy来避免在迭代过程中直接修改原始字典。你可以创建一个字典的副本，然后在这个副本上进行修改，最后将修改后的副本赋值给原始字典。另外，由于你的flush函数需要遍历所有的键值对，你也可以考虑使用字典的items()方法，这样可以避免直接修改字典的大小。这是一个修改后的flush函数示例：

def flush(segmType_k = None):
    if NoInitError.is_init == False:
        raise NoInitError(NoInitError.ErrText)
    stdoutl = dict(stdout)  # 创建字典的副本
    if flushType == "real":
        text = []
        for k, v in stdoutl.items():  # 使用items()方法遍历字典的键值对
            text.append(gettextblock(k,v))
        cls()
        print("\n".join([head,*("".join([i,("\n" if text.index(i) != len(text) - 1 else ""),(sep if text.index(i) != len(text) - 1 else "")]) for i in text],end))
    elif flushType == "segm":
        if "title" not in flush_typeReal_pringList:
            cls()
            print(head,end = "")
            flush_typeReal_pringList.append("title")
        else:
            if segmType_k:
                print("\n" + gettextblock(segmType_k,stdoutl[segmType_k]) + "\n" + sep,end = "")
                flush_typeReal_pringList.append(segmType_k)
            for k in flush_typeReal_pringList:  # 只遍历已知的键
                if k in stdoutl:  # 检查键是否在字典中
                    print("\n" + gettextblock(k,stdoutl[k]) + "\n" + sep,end = "")
    else:
        raise FlushTypeError(FlushTypeError.ErrText % flushType)
    # 其他代码...

另外，你的代码中有一些地方可能需要进一步优化和改进。例如，你的getstr()函数每次调用都会生成一个新的随机字符串，这可能并不是你想要的。你可能需要更精细地控制这个函数的调用，以避免不必要地生成字符串。另外，你在代码中使用了全局变量（如raisevar），这可能会导致代码难以理解和维护。你可以考虑使用函数参数或类属性来替代全局变量。