import socketio

class ExportResourceClient:

    def __init__(self, server_url):
        self.server_url = server_url
        self.sio = socketio.Client()
        # 用于判断当前任务是否完成
        self.current_task_done = False
        self._register_events()

    def _register_events(self):
        """注册事件回调函数"""

        @self.sio.event
        def connect():
            log.info("SocketIO连接成功")

        @self.sio.event
        def export_start(data):
            log.info("开始导出...")

        @self.sio.event
        def export_progress(data):
            log.info(f"进度: {data['msg']}")

        @self.sio.event
        def export_error(data):
            log.info(f"导出失败: {data['msg']}")
            self.current_task_done = True  # 标记任务完成，避免卡住

        @self.sio.event
        def export_info(data):
            log.info(f"信息: {data['msg']}")

        @self.sio.event
        def export_success(data):
            log.info(f"导出成功: {data['msg']}")
            log.info(f"下载地址: {data['url']}")
            self.current_task_done = True  # 标记任务完成

        @self.sio.event
        def task_status(data):
            print(f"任务状态: {'运行中' if data['is_running'] else '已完成'}")

    def connect(self):
        """连接到WebSocket服务器"""
        self.sio.connect(self.server_url)

    def send_task(self, task_name):
        """发送任务事件"""
        self.current_task_done = False  # 任务开始时重置状态
        self.sio.emit(task_name)
        log.info(f"任务 {task_name} 已发送")

    def wait_for_task_completion(self):
        """等待当前任务完成"""
        log.info("等待任务完成...")
        while not self.current_task_done:
            time.sleep(1)

    def start_export_kr_resource(self):
        log.step("start export kr resource event_name:start_export_kr_resource")
        self.send_task("start_export_kr_resource")
        self.wait_for_task_completion()

    def start_export_kr_img_resource(self):
        log.step("start export img resource event_name:start_export_kr_img_resource")
        self.send_task("start_export_kr_img_resource")
        self.wait_for_task_completion()

    def start_export_translation_pack(self):
        log.step("start export translation pack event_name:start_export_translation_pack")
        self.send_task("start_export_translation_pack")
        self.wait_for_task_completion()

    def wait_for_events(self):
        """保持WebSocket连接，等待事件"""
        self.sio.wait()

开发可以支配测试吗

羽毛球无敌

AES 加密时会约定在前面加上 16 个随机的字节，服务解密的时候会先去掉 16 个随机字节，然后解密。因为每次都有会随机字节，所以加密的结果不同，但解密的结果是相同的。

线上上线是如何操作的？不会也是手动部署？应该都是 jenkins + salt 部署？线上部署那一套在测试服部署一下就 OK。客户端包可以直接部署一个 Nginx，让前端同学开发完之后直接放到 nginx 配置对应的目录，然后在通知群发提测通知带上对应的链接就可以了。

没有灰度吗？ 灰度完之后看下崩溃率，崩溃率高了就让研发排查修复，继续灰度，崩溃率达标在上线。

这就有点害怕了

我们上个月又刚裁了一波

感觉不是。自动化也是需要人来编写的；

每周 周三 15-18 点灰度， 每周 周四 15-18 点全网。

轮询查数据库比轮询调查询接口更快
查数据库是为了找出究竟是写入的慢 ；还是数据已经写入数据库了，接口获取还是获取不到
查 db 不是增加的工作量，是接口自动化的必须验证的。

新增之后轮询去查 db，db 里有了再直接调接口断言。db 里没有就说明还没创建成功。

《人民日报》：“别把劲儿都使在孩子身上，如果自己充实、快乐、有责任感，有情绪管理能力，孩子会模仿你的，教育从来不是要求，而是影响，教育之道无他，唯爱与榜样而已。真正觉醒的家庭：只改变自己，不改变孩子。”

阿里云 云企业网 主备切换

@ 爱摸鱼的小狐狸 到方法了，在 CSDN 找了一篇贴字，直达罗马了。 链接分享：https://blog.csdn.net/qq_37186947/article/details/116862628?ops_request_misc=%257B%2522request%255Fid%2522%253A%25224B3BE34D-179F-4CF4-A62B-57C3847A8F7F%2522%252C%2522scm%2522%253A%252220140713.130102334.pc%255Fall.%2522%257D&request_id=4B3BE34D-179F-4CF4-A62B-57C3847A8F7F&biz_id=0&utm_medium=distribute.pc_search_result.none-task-blog-2~all~first_rank_ecpm_v1~rank_v31_ecpm-2-116862628-null-null.142v100pc_search_result_base7&utm_term=app%E6%8A%93%E6%B3%A2%E7%BB%95%E8%BF%87%E8%AF%81%E4%B9%A6&spm=1018.2226.3001.4187^
重点是：VirtualXposed + JustTrustMe【修改了系统底层对 SSL 的校验】

@CmYzzZ 我这个头像是 qq 头像里自带的【大猫】头像，然后我保存截图的。qq 里自动的还有动效

@Vanessa 具体点呢

这个手机主板是如何联网的？
这样的实现和使用云手机有啥区别呢？

学历是铁门槛吗

好的

我不认同用旧不用新

想看看是否可以用它整性能测试

如果 locust 有 rust 加持是不是就很牛逼了

阿里