编程猫云变量云列表修改器V1.1.0
用于修改和读取编程猫kitten4、kitten3、kittenN、Nemo作品的云变量和云列表，使用python当作后端
总体介绍
警告：本程序仅可用于学习和扩展编程猫图形化程序编辑器的用途，禁止用于不道德或者不合法的行为！任何使用本程序所造成的后果，由您自行承担！概述
这个库可以修改和读取编程猫 kitten4、kitten3、kittenN、Nemo 作品的云变量和云列表，使用python当作后端
适用场景
编程猫作品（除Coco作品）的聊天室、PK系统等需要大量存储数据，云列表都存不下，或需要后端出题的场景。
主要特性
- 支持长连接，几天都不会断；
- 支持忽视消息的功能和解除忽视消息的功能；
- 支持监听原生websocket事件和共有云变量共有云列表的各种事件；
- 一行简单的代码即可修改共有云变量或者共有云列表；
- 支持一键查看共有云变量和共有云列表的所有数据；
- 支持获取在线人数。
相比于V1.0.0版本的改进
V1.1.0相比于V1.0.0版本，改进了以下功能：- 不再使用 playwright 库获取登录凭证，采用更加简单、资源占用更少的方法，告别 playwright 库下载慢、系统资源消耗高的各种问题；
- 修改云变量和云列表时支持同时修改多个云变量和云列表，没有了每次修改需要间隔 0.5 秒的规则；
- 新增锁功能，告别了多线程之间的数据竞争。
V1.1.0 相比于 V1.0.0 版本，没有新增任何功能，因为 V1.1.0 是兼容版本。
本版本您可以理解为 V1.0.0 的升级兼容版本，所有 API 的名称都没有变，您使用此库 V1.0.0 版本编写的程序99.9%可以在 V1.1.0 版本可以运行。 如果您想体验最新版，尝试最新版的新功能，您可以下载 V2.0.0 版本，但是 V2.0.0 版本包含破坏性变更，您写的 V1.x 版本的程序可能需要修改。 V2.0版本还不知道什么时候上线。安装
本库暂未上线 pypi ，所以不能使用 pip 安装。
需要依赖： websocket （不是websockets ） 、 requests
下载下来后，与程序文件放在同目录下，使用此命令导入库：
import bcmcloud快速开始1
作品：图片渲染器
此示例是让编程猫作品接收后端发送的图片数据，渲染到前端。
代码：
import time
import numpy as np
import bcmcloud, threading
from PIL import Image
import colorsys
def encode_image_to_custom_format(image_path):
"""
将图片编码为自定义格式
Args:
image_path: 图片文件路径
Returns:
[图片高度, 图片宽度, 图片数据列表]
图片数据列表中的每个字符串包含126个字符（除最后一个可能不足）
"""
try:
# 打开图片并转换为RGB模式
img = Image.open(image_path).convert('RGB')
width, height = img.size
# 使用numpy获取像素数据，避免getdata的弃用警告，得到形状为(height*width, 3)的数组
pixels = np.array(img).reshape(-1, 3)
# 定义编码字符集
ones_digits = "0123456789ABCDEF" # 个位字符集（0-15）
tens_digits = "0123456789ABCDEFGHIJKLMNO" # 十位字符集（0-24）
def value_to_custom_hex(value, max_val):
"""
将值转换为自定义HEX编码
Args:
value: 要编码的值
max_val: 最大值（用于限制范围）
Returns:
2位自定义HEX字符串
"""
# 确保值在有效范围内并转为整数
value = max(0, min(int(round(value)), max_val))
# 计算个位和十位
ones = value % 16
tens = value // 16
# 确保十位不超过24（tens_digits的范围）
tens = min(tens, 24)
# 转换为自定义HEX
return f"{tens_digits[tens]}{ones_digits[ones]}"
# 编码所有像素
encoded_pixels = []
for r, g, b in pixels:
# 转换为HSL（colorsys返回的H在0-1，S和L在0-1）
h, l, s = colorsys.rgb_to_hls(r / 255.0, g / 255.0, b / 255.0)
# 转换为目标范围
h_val = round(h * 360) % 360 # 0-360
s_val = round(s * 100) # 0-100
l_val = round(l * 100) # 0-100
# 编码为自定义HEX
h_hex = value_to_custom_hex(h_val, 360)
s_hex = value_to_custom_hex(s_val, 100)
l_hex = value_to_custom_hex(l_val, 100)
# 拼接为6位字符串，顺序改为H、S、L，符合标准HSL解析顺序
encoded_pixels.append(f"{h_hex}{s_hex}{l_hex}")
encoded_str = "".join(encoded_pixels)
# 每126个字符（21个像素）分组
group_size = 126 # 21像素 * 6字符
encoded_groups = [encoded_str[i:i+group_size] for i in range(0, len(encoded_str), group_size)]
return [int(height), int(width), encoded_groups]
except Exception as e:
print(f"处理图片时出错: {e}")
return None
def work():
while not bcmcloud_worker.ready:#等待准备就绪
time.sleep(1)
"""
server_state 变量:
1: 服务器就绪响应
2: 客户端就绪响应
3: 服务器发送完成
"""
bcmcloud_worker.var_upd("server_state", 1)
result = encode_image_to_custom_format("./send.png")
hex_chunks = result[2]
h = result[0]
w = result[1]
while not bcmcloud_worker.cloud_vars["server_state"] == 2: # 等待客户端准备就绪
bcmcloud_worker.var_upd("server_state", 1)
time.sleep(2)
bcmcloud_worker.var_upd("hex_len", len(hex_chunks)) # 图片数据列表项数
bcmcloud_worker.var_upd("total_x", w) # 图片宽
bcmcloud_worker.var_upd("total_y", h) # 图片长
bcmcloud_worker.var_upd("server_state", 1)
while not bcmcloud_worker.cloud_vars["server_state"] == 2:
time.sleep(0.1)
begin_index = 0
while len(hex_chunks) > begin_index: # 发送图片数据
for i in range(1, min(len(hex_chunks)-begin_index+1, 201)):
bcmcloud_worker.list_replace("img_data", i, hex_chunks[begin_index + i - 1])
bcmcloud_worker.var_upd("server_state", 1)
while not bcmcloud_worker.cloud_vars["server_state"] == 2:
time.sleep(0.1)
begin_index += 200 # 服务器一次发送200项
bcmcloud_worker.var_upd("server_state", 3) # 图片数据发送完成
while not bcmcloud_worker.cloud_vars["server_state"] == 2: # 等待客户端回应
time.sleep(0.1)
bcmcloud_worker.var_upd("server_state", 1)
print("发送完成。")
threading.Thread(target=work).start() # 重新启动线程以便下次再次发送图片
config = \
{
"phone_number": "你的手机号",
"password": "你的密码",
"work": "304104073"
}
bcmcloud_worker = bcmcloud.codemao_cloud(config)
threading.Thread(target=work).start()
bcmcloud_worker.run()
效果（如果不动了就重新加载网页再看）：
快速开始2
作品：有后端的聊天室
代码：
import bcmcloud
import time,threading,math
# 主程序
msg_list = []
msg_list.append("默认消息，忽略即可$系统")
#消息格式："消息$用户名"
def send_msg(start_id,end_id,json,client_sendMsg):
global bcmcloud_worker
max_end_time = time.time()+25#设置与客户端交流的最大时间
for i in range(start_id,end_id):
if i%2==1:
msg = "get_msg_done&"+client_sendMsg[1]+"&"+msg_list[i]
else:
msg = "get_msg_done&"+client_sendMsg[1]+"&"+msg_list[i]+" "
bcmcloud_worker.list_replace("event_ls",json["nth"],msg)
while True:
time.sleep(1)
temp = bcmcloud_worker.list_get("event_ls")
if temp[json["nth"]-1] == "next":
break
if max_end_time < time.time():
bcmcloud_worker.list_replace("event_ls",json["nth"],"empty")
return
msg = "get_msg_done&"+client_sendMsg[1]+"&get_ok"
bcmcloud_worker.list_replace("event_ls",json["nth"],msg)
time.sleep(10)
#a.view_list("event_ls",json.get("nth"))
return
def handle_listEvent(json):
"""
下面是一个用户请求聊天消息的示例，此聊天室存储请求的云数据是一个列表，用户会找到最前一个空闲的云列表项，来发送下面的信息：
A: 用户发送: get_msg&2A69818C827B3656&1
意思是会话 id 为 2A69818C827B3656 的用户请求 get_msg(获取聊天信息的意思) 并且是获取第 1 页的
B: 返回消息: get_msg_done&2A69818C827B3656&默认消息，忽略即可$系统
意思为会话 id 为 2A69818C827B3656 的用户收到了 1 条消息，为 get_msg_done(成功获取了一条消息的意思) ，且消息为 默认消息，忽略即可$系统
其中 “默认消息，忽略即可” 为那个用户发的文字， “系统”为用户id(通常情况下为训练师编号,只是这条消息比较特殊)
C: 用户发送: next
意思是用户已经接收了这条消息，请服务器继续发送下一条消息。接下来会一直执行B步骤和C步骤，如果聊天消息足够的情况下，最多发送10条消息。
D: 返回消息: get_msg_done&2A69818C827B3656&get_ok
意思是服务器已经发送完会话 id 为 2A69818C827B3656 的消息了
E: 用户发送: empty
意思是客户端已接收完毕，修改该云列表项为空闲状态。接收完毕
注: 此程序有超时功能，如果用户在25秒内没有接收消息，服务器会强制修改该云列表项为空闲状态，并结束此获取评论会话(当前其他的会话和以后的会话不会影响)
"""
global bcmcloud_worker
user_method = json.get("value").split("&")
#a.noview_list("event_ls",json.get("nth"))
#考虑到目前noview和view函数有bug，因此不建议使用
if user_method[0] == "get_msg":
client_sendMsg:str = json.get("value")
client_sendMsg = client_sendMsg.split("&")
room_id = client_sendMsg[1]
page = client_sendMsg[2]
end_id = int(page)*10
start_id = end_id-10
if end_id > len(msg_list):
end_id = len(msg_list)
if client_sendMsg[0] == "get_msg":
kkk = threading.Thread(target=send_msg,args=(start_id,end_id,json,client_sendMsg,))
kkk.daemon = True
kkk.start()
elif user_method[0] == "send_msg":
client_sendMsg:str = json.get("value")
client_sendMsg = client_sendMsg.split("&")
room_id = client_sendMsg[1]
user_id = client_sendMsg[2]
message = client_sendMsg[3]
msg_list.append(message+"$"+str(user_id))
if len(msg_list) > 1000:
msg_list.pop(0)
client_msg = "send_msg_done&"+str(room_id)+"&send_ok"
bcmcloud_worker.list_replace("event_ls",json["nth"],client_msg)
def cloud_work():#每隔10秒修改最大页数
global bcmcloud_worker
while not bcmcloud_worker.ready:#等待准备就绪
time.sleep(1)
while True:
temp = math.ceil(len(msg_list)/10)
bcmcloud_worker.var_upd("max_page",temp)
time.sleep(10)
if __name__ == "__main__":
config = \
{
"phone_number": "你的手机号",
"password": "你的密码",
"work": "257509128"
}
bcmcloud_worker = bcmcloud.codemao_cloud(config)
threading.Thread(target=cloud_work).start()
#绑定列表名称为event_ls的replace(替换)事件，处理函数为handle_listEvent
bcmcloud_worker.bind("event_ls-replace", handle_listEvent, "list_name")
bcmcloud_worker.run()
效果（如果不动了就重新加载网页再看）：
常用API
需要注意：本版本虽然没有了需要间隔0.5秒才能发送的规则，但是实际上底层还是每隔1秒检测一次有没有内容要发送。举个例子：
- 13:00:01.02 - 云变量a消息：设置为"0"
- 13:00:01.30 - 云变量b消息：设置为"-1"
- 13:00:01.30 - 云变量c消息：设置为"-2"
- 13:00:01.30 - 云列表d消息：末尾追加"-3"
- 13:00:01.30 - 云列表e消息：末尾追加"-4"
- 13:00:01.30 - 云列表d消息：末尾追加"-5"
- 13:00:01.90 - 云变量a消息：设置为"1"
- 13:00:01.95 - 云变量a消息：设置为"2"
那么最终结果：
| 云变量a | 云变量b | 云变量c | 云列表d | 云列表e |
| 13:00:01成功设置为"0"，13:00:04成功设置为"2" | 13:00:04成功设置为"-1" | 13:00:04成功设置为"-2" | 13:00:02成功设置为["-3", "-5"] | 13:00:03成功设置为["-4"] |
为什么修改速度不一？
因为编程猫底层对于云变量修改有限制，经测试最短 0.5 秒一次？但是 kitten 编辑器的标准为 1 秒一次，所以此库也一样，对于发送云变量修改请求，频率为每秒最多 1 次。
但是有一些修改请求可以打包在一起发送，比如所有云变量的修改可以打包在一起发送，单个云列表的修改操作也可以打包在一起发送（两个云列表的话就要分开发送）。
本编程猫云变量修改器是先处理云列表修改请求（如果有两个云列表，至于哪个先处理，那就凭运气了，实在要说似乎是创建顺序决定的？目前我不清楚），再处理云变量修改请求。
本库在有云变量或者云列表修改时，会直接处理请求，不会阻塞调用的线程，也不会再等待一会儿查找有没有更多的修改请求可以一起打包发送的。在发送线程空闲时，检测修改请求的频率为 0.01 秒一次。
所以就会出现上面的 13:00:01 秒只处理了 a 设置为 "0" 的请求，处理完这个请求会休息 1 秒；过完 1 秒后，继续查看有没有修改请求，发现有两个云列表的修改请求，随机抽取一个云列表， 将所有该列表的修改请求整合起来一起发送；休息 1 秒，再处理下一个列表；接着，再处理云变量修改请求，打包一起发送
def bind(self,name:str,func,bind_type:str = "websocket") -> None:
"""
绑定一个事件
Args:
:param name: 事件名称
:param func: 事件处理函数
:param bind_type: 如果设置为字符串“list_name”或“var_name”则参数name指的是云列表或云变量名称改变事件，如果设置为\
“websocket_name”则参数name指的是websocket消息事件，如果不写这个参数，则默认参数为websocket消息事件\
建议普通人使用云列表或云变量名称改变事件
"""def var_get(self,name:str) -> int|str:
"""
获取某个云变量的值
Args:
:param name: 云变量名称
实际上bcmcloud.cloud_vars["xxx"]也可以获取，但是注意，这个字典不可以修改，只能读取
"""def list_get(self,name:str) -> list:
"""
获取某个云列表的所有元素
Args:
:param name: 云列表名称
实际上bcmcloud.cloud_lists["xxx"]也可以获取，但是注意，这个字典不可以修改，只能读取
"""
def var_upd(self,var_name:str,value:int|str) -> None:
"""
修改云变量值
Args:
:param var_name: 云变量名称
:param value: 要修改的值,由于kitten自身有bug,不能设置修改的值为列表
"""
def list_append(self,list_name:str,value:str|int) -> None:
"""
云列表末尾追加
Args:
:param list_name: 云列表名称
:param value: 要修改的值
"""
def list_replace(self,list_name:str,replace_index:int,value:str|int) -> None:
"""
云列表替换,注意下标是从1开始的,注意两次相同列表项的修改需要间隔至少0.5秒
Args:
:param list_name: 云列表名称
:param replace_index: 要替换的值的下标,注意下标是从1开始的
:param value: 要修改的值
"""
def list_del(self,list_name:str,del_index:int) -> None:
"""
云列表删除项，注意下标是从1开始的
Args:
:param list_name: 云列表名称
:param del_index: 要删除的值的下标
:param value: 要修改的值
"""
def list_insert(self,list_name:str,insert_index:int,value:str|int) -> None:
"""
云列表插入项,注意下标是从1开始的
Args:
:param list_name: 云列表名称
:param insert_index: 插入下标
:param value: 要修改的值
"""
def list_len(self,list_name:str) -> int:
"""
返回云列表项数
Args:
:param list_name: 云列表名称
实际上len(bcmcloud.cloud_lists["xxx"])也可以获取该列表项数
"""
常见问题
1、
Exception in thread Thread-1 (work):
Traceback (most recent call last):
File "C:\Users\x\AppData\Local\Programs\Python\Python313\Lib\threading.py", line 1043, in _bootstrap_inner
self.run()
~~~~~~~~^^
File "C:\Users\x\AppData\Local\Programs\Python\Python313\Lib\threading.py", line 994, in run
self._target(*self._args, **self._kwargs)
~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "i:\work\py\bcm\example1.py", line 94, in work
bcmcloud_worker.var_upd("xxx", 1)
~~~~~~~~~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^
File "i:\work\py\bcm\bcmcloud.py", line 558, in var_upd
this_cvid = str(self.name_to_cvid[var_name])
~~~~~~~~~~~~~~~~~^^^^^^^^^^
KeyError: 'xxx'1、当作品新增了云变量或者云列表后，必须要更新发布作品后本程序才可以访问新增的云列表和云变量
2、请检查前后端云变量名称是否匹配
2、
开始登录...
登录成功！即将开始监听云变量(列表)
云变量连接错误：Handshake status 401 Unauthorized -+-+- {'date': 'Wed, 11 Feb 2026 08:20:29 GMT', 'transfer-encoding': 'chunked', 'connection': 'keep-alive', 'strict-transport-security': 'max-age=15724800; includeSubDomains'} -+-+- None
编程猫云变量(列表)连接已关闭，关闭代码：None，关闭消息：None
检测到此程序和编程猫云变量(列表)没有任何通讯
可能是因为当前作品id为不存在的作品或者您的作品没有用到云变量
同时需要注意您的作品id一定要是编程猫社区发布作品的作品id而不是kitten编辑器上的开始登录...
登录成功！即将开始监听云变量(列表)
云变量连接错误：Handshake status 400 Bad Request -+-+- {'date': 'Wed, 11 Feb 2026 08:22:16 GMT', 'transfer-encoding': 'chunked', 'connection': 'keep-alive', 'strict-transport-security': 'max-age=15724800; includeSubDomains'} -+-+- None
编程猫云变量(列表)连接已关闭，关闭代码：None，关闭消息：None
检测到此程序和编程猫云变量(列表)没有任何通讯
可能是因为当前作品id为不存在的作品或者您的作品没有用到云变量1、当前作品id为不存在的作品或者您的作品没有用到云变量
2、您的作品id一定要是编程猫社区发布作品的作品id而不是kitten编辑器上的作品id
3、检查作品id是否输入正确
4、检查作品id前后以及中间有没有多余的空格或者其他字符
目前就发现有这两个问题，如果您有其他的问题，欢迎提交！