Python3如何无需重启脚本更新users列表并避免文件重复写入?
问题与解决方案:保持users变量与文件同步且避免重复写入
问题概述
脚本启动时读取users.txt到users列表,后续将新用户追加到列表并调用写入函数时,会出现两个核心问题:
- 写入函数会把整个
users列表追加到文件,导致文件内容重复; - 若
users.txt被外部修改,内存中的users变量无法自动更新(当前场景以内部添加为主,扩展场景需考虑)。
现有代码
初始读取与写入函数:
import os import time # 1. Read users.txt if os.path.isfile('users.txt'): with open('users.txt', 'r') as file: users = [line.rstrip('\n') for line in file] # 2. Function for writing to users.txt def write_to_users(): try: with open('users.txt', 'a') as file: for item in users: file.write('{}\n'.format(item)) except FileNotFoundError: print('Current path:', os.getcwd()) print('Isdir reports:', os.path.isdir(os.getcwd())) print('Isfile reports:', os.path.isfile('users.txt')) raise
后续添加用户的代码:
users.append(author) write_to_users() time.sleep(1)
解决方案
方案一:修改写入逻辑,只写入新增用户(最简方案)
核心思路:每次仅写入新增的单个用户,而非整个列表;添加前检查用户是否已存在,从源头避免重复。
修改后的写入函数:
def write_to_users(new_user): try: with open('users.txt', 'a') as file: file.write(f'{new_user}\n') except FileNotFoundError: print('Current path:', os.getcwd()) print('Isdir reports:', os.path.isdir(os.getcwd())) print('Isfile reports:', os.path.isfile('users.txt')) raise
添加用户的代码改为:
# 先检查用户是否已存在,避免重复添加 if author not in users: users.append(author) write_to_users(author) time.sleep(1)
该方案保证users列表始终最新,同时彻底解决文件内容重复问题。
方案二:封装用户管理逻辑(更健壮)
将用户的加载、添加、同步逻辑封装到类中,统一管理内存列表与文件,扩展性更强,适合长期维护的场景。
import os import time class UserManager: def __init__(self, file_path='users.txt'): self.file_path = file_path self.users = self._load_users() def _load_users(self): """从文件加载用户列表,自动过滤空行""" if os.path.isfile(self.file_path): with open(self.file_path, 'r') as file: return [line.rstrip('\n') for line in file if line.strip()] return [] def add_user(self, username): """添加用户:检查有效性与重复 -> 更新内存列表 -> 写入文件""" username = username.strip() if not username or username in self.users: return False # 空用户名或已存在,添加失败 self.users.append(username) self._write_single_user(username) return True def _write_single_user(self, username): """写入单个用户到文件""" try: with open(self.file_path, 'a') as file: file.write(f'{username}\n') except FileNotFoundError: print('Current path:', os.getcwd()) print('Isdir reports:', os.path.isdir(os.getcwd())) print('Isfile reports:', os.path.isfile(self.file_path)) raise def refresh(self): """重新从文件加载用户,同步内存列表(应对外部修改)""" self.users = self._load_users() # 使用示例 user_manager = UserManager() # 添加用户 if user_manager.add_user(author): print(f"用户 {author} 已成功添加") else: print(f"用户 {author} 已存在或无效") time.sleep(1) # 若需要同步外部修改的文件,调用refresh # user_manager.refresh()
方案三:定时刷新列表(应对外部文件修改)
如果脚本运行期间users.txt可能被其他程序修改,可通过线程定时重新读取文件,更新内存列表。注意添加锁避免并发冲突:
import os import time import threading users = [] user_lock = threading.Lock() def load_users(): """加载用户列表""" if os.path.isfile('users.txt'): with open('users.txt', 'r') as file: return [line.rstrip('\n') for line in file if line.strip()] return [] def refresh_users_periodically(interval=60): """定时刷新用户列表,interval为间隔秒数""" global users while True: with user_lock: users = load_users() time.sleep(interval) # 初始化加载 users = load_users() # 启动刷新线程(守护线程,随主程序退出) threading.Thread(target=refresh_users_periodically, daemon=True).start() # 添加用户的逻辑(加锁保证线程安全) def add_user_safely(username): username = username.strip() with user_lock: if username not in users: users.append(username) # 写入单个用户 with open('users.txt', 'a') as file: file.write(f'{username}\n') return True return False # 使用示例 if add_user_safely(author): print(f"用户 {author} 已添加") else: print(f"用户 {author} 已存在") time.sleep(1)
总结
- 仅内部添加用户:方案一足以解决问题,简单高效;
- 需要长期维护或扩展功能:方案二的类封装更适合;
- 存在外部修改文件的场景:方案三结合定时刷新与锁机制保证同步。
内容的提问来源于stack exchange,提问作者zin
相关产品推荐
相关产品推荐

