first commit

framework/README.md

@@ -0,0 +1,45 @@
+# Robogo 自动化测试框架
+
+该框架基于 Python 实现，旨在将 API 通用逻辑、业务流程和配置信息解耦，方便不同基础的人员快速编写业务测试脚本。
+
+## 目录结构说明
+
+- `framework/core/`: **核心基类层**。包含 `BaseAPI` (封装了请求、鉴权、响应检查)、`logger` (统一日志)、`exceptions` (异常定义)。
+- `framework/business/`: **业务服务层**。按业务领域(云桌面、开发机、账单等)封装原子操作。
+- `framework/config/`: **配置层**。统一管理 URL、Token、资源 ID 等配置。
+- `framework/models/`: **模型层**。定义 `StepResult` 等通用数据结构。
+- `framework/scripts/`: **业务逻辑层 (脚本区)**。在此处调用业务服务类编写具体的测试流。
+
+## 快速上手
+
+### 1. 编写一个新脚本
+在 `framework/scripts/` 下创建一个 `.py` 文件，例如 `my_test.py`:
+
+```python
+from framework.business.cloud_desktop import CloudDesktopService
+from framework.config.settings import Config
+
+def test_flow():
+    # 初始化服务
+    service = CloudDesktopService()
+    
+    # 直接调用业务方法
+    desktop_id = service.create_desktop("我的测试机")
+    print(f"创建成功: {desktop_id}")
+    
+    # 更多操作...
+    service.stop_desktop(desktop_id)
+
+if __name__ == "__main__":
+    test_flow()
+```
+
+### 2. 更新配置
+直接修改 `framework/config/settings.py` 中的 `AUTH_TOKEN` 或相关 `ID` 即可，全局生效。
+
+---
+
+## 优势
+- **低门槛**: 业务人员只需关注 `scripts/` 下的逻辑，无需关心复杂的 Header 构造和错误处理。
+- **易维护**: 接口变更只需在 `business/` 层修改一处。
+- **可扩展**: 支持后续加入 `BaseUI` (如 Playwright) 扩展 UI 自动化能力。

framework/bot.py

@@ -0,0 +1,156 @@
+import re
+import json
+import threading
+import concurrent.futures
+from datetime import datetime
+import lark_oapi as lark
+from lark_oapi.api.im.v1 import (
+    CreateMessageRequest,
+    CreateMessageRequestBody,
+)
+from lark_oapi.event.callback.model.p2_card_action_trigger import P2CardActionTrigger
+
+from framework.config.settings import Config as config
+from framework.core.logger import get_logger
+from framework.business.cloud_desktop import CloudDesktopService
+from framework.business.dev_machine import DevMachineService
+from framework.business.billing import BillingService
+from framework.business.meal_reminder import MealReminderService
+from framework.business.boilerplate import BoilerplateService
+from framework.core.exceptions import TokenExpiredError
+
+logger = get_logger("FeishuBot")
+
+# ---------- 飞书卡片常量 ----------
+TOKEN_UPDATE_CARD_ID = "AAq2uZZvyWY01"
+RESULT_CARD_ID = "AAqKcs3OrZBnJ"
+INTRO_CARD_ID = "AAq2uZZvyWY06"
+
+# 线程池
+executor = concurrent.futures.ThreadPoolExecutor(max_workers=10)
+
+# ---------- 业务服务实例化 ----------
+cd_service = CloudDesktopService()
+dm_service = DevMachineService()
+billing_service = BillingService()
+meal_service = MealReminderService()
+bp_service = BoilerplateService()
+
+# ---------- 全局计数器 ----------
+_counter_lock = threading.Lock()
+_global_counter = 0
+
+def _next_default_name() -> str:
+    global _global_counter
+    with _counter_lock:
+        _global_counter += 1
+        return f"验证{_global_counter:02d}"
+
+# ---------- 飞书 API Client ----------
+_lark_client = lark.Client.builder() \
+    .app_id(config.FEISHU_APP_ID) \
+    .app_secret(config.FEISHU_APP_SECRET) \
+    .build()
+
+def send_feishu_message(chat_id: str, text: str):
+    body = CreateMessageRequestBody.builder().receive_id(chat_id).msg_type("text").content(json.dumps({"text": text})).build()
+    request = CreateMessageRequest.builder().receive_id_type("chat_id").request_body(body).build()
+    _lark_client.im.v1.message.create(request)
+
+def send_card_json(chat_id: str, card_json: dict):
+    body = CreateMessageRequestBody.builder().receive_id(chat_id).msg_type("interactive").content(json.dumps(card_json)).build()
+    request = CreateMessageRequest.builder().receive_id_type("chat_id").request_body(body).build()
+    _lark_client.im.v1.message.create(request)
+
+def _handle_single_task(chat_id, name, task_type):
+    try:
+        if task_type == "cd_post":
+            results, instance_id = cd_service.run_lifecycle_test(name, flow_type="postpaid_to_prepaid")
+            label = "云桌面(按转包)"
+        elif task_type == "cd_pre":
+            results, instance_id = cd_service.run_lifecycle_test(name, flow_type="prepaid_to_postpaid")
+            label = "云桌面(包转按)"
+        elif task_type == "bp":
+            results, instance_id = bp_service.run_lifecycle_test(name)
+            label = "样本工程"
+        else:
+            results, instance_id = dm_service.run_lifecycle_test(name)
+            label = "开发机"
+        
+        send_feishu_message(chat_id, f"✅ {label} [{name}] 验证完成！")
+    except TokenExpiredError:
+        send_feishu_message(chat_id, "⚠️ 认证过期，请更新 Token")
+    except Exception as e:
+        logger.error(f"Task error: {e}")
+        send_feishu_message(chat_id, f"❌ {label} 验证异常: {e}")
+
+def _parse_command(text: str):
+    patterns = {
+        "cd_post": r"robogo-云桌面[ \-:=]+(.+)$",
+        "cd_pre": r"robogo-云桌面包月[ \-:=]+(.+)$",
+        "dm": r"robogo-开发机[ \-:=]+(.+)$",
+        "bp": r"robogo-样本工程[ \-:=]+(.+)$",
+        "billing": r"账单[ \-:=]*(.*)$",
+    }
+    # 辅助匹配模式（不带名称的情况）
+    fallback_patterns = {
+        "cd_post": r"robogo-云桌面\s*$",
+        "cd_pre": r"robogo-云桌面包月\s*$",
+        "dm": r"robogo-开发机\s*$",
+        "bp": r"robogo-样本工程\s*$",
+    }
+
+    text_clean = text.strip()
+    logger.info(f"Parsing command from text: '{text_clean}'")
+    
+    # 先试带有名称的模式
+    for cmd, p in patterns.items():
+        m = re.search(p, text_clean)
+        if m: 
+            logger.info(f"Matched named pattern: {cmd}, name: {m.group(1).strip()}")
+            return cmd, m.group(1).strip()
+    
+    # 再试默认模式
+    for cmd, p in fallback_patterns.items():
+        m = re.search(p, text_clean)
+        if m: 
+            logger.info(f"Matched fallback pattern: {cmd}")
+            return cmd, None
+
+    logger.warning(f"No pattern matched for text: '{text_clean}'")
+    return None, None
+
+def _on_message(data):
+    msg = data.event.message
+    chat_id = msg.chat_id
+    content = json.loads(msg.content)
+    text = content.get("text", "")
+    
+    clean_text = re.sub(r'@[^ ]+', '', text).strip()
+    
+    if "饭否" in clean_text:
+        send_card_json(chat_id, meal_service.build_card())
+        return
+
+    cmd, name = _parse_command(clean_text)
+    if cmd == "billing":
+        items = billing_service.fetch_billing_data(name)
+        send_feishu_message(chat_id, f"共找到 {len(items)} 条账单记录")
+    elif cmd:
+        name = name or _next_default_name()
+        executor.submit(_handle_single_task, chat_id, name, cmd)
+    else:
+        send_feishu_message(chat_id, "未能识别指令，请尝试 [robogo-云桌面] 或 [账单]")
+
+def main():
+    # 使用 WebSocket (Long Polling) 模式，无需公网 IP
+    event_handler = lark.EventDispatcherHandler.builder(config.FEISHU_APP_ID, config.FEISHU_APP_SECRET) \
+        .register_p2_im_message_receive_v1(_on_message) \
+        .build()
+    
+    ws_client = lark.ws.Client(config.FEISHU_APP_ID, config.FEISHU_APP_SECRET, event_handler=event_handler)
+    logger.info("Bot started in WebSocket mode...")
+    ws_client.start()
+
+if __name__ == "__main__":
+    main()

framework/business/billing.py

@@ -0,0 +1,32 @@
+from framework.core.base_api import BaseAPI
+from framework.config.settings import Config
+
+class BillingService(BaseAPI):
+    def __init__(self):
+        super().__init__(Config.CLOUD_BASE_URL, Config.CLOUD_AUTH_TOKEN)
+
+    def fetch_billing_data(self, name: str = None, page_size: int = 500):
+        endpoint = "/api/dcloudResourceApi/consumptions"
+        params = {
+            "page": 1,
+            "pageSize": page_size,
+            "exactMatch": "true"
+        }
+        if name:
+            params["instanceName"] = name
+            
+        headers = self.get_common_headers("/admin/consumptions")
+        data = self.request("GET", endpoint, params=params, headers=headers)
+        
+        # 提取数据项
+        data_body = data.get("data", {})
+        if isinstance(data_body, list):
+            items = data_body
+        else:
+            items = data_body.get("list") or data_body.get("items") or []
+            
+        # 精准匹配校准 (如果接口过滤不准)
+        if name:
+            items = [item for item in items if str(item.get("instanceName", "")) == name]
+            
+        return items

framework/business/boilerplate.py

@@ -0,0 +1,57 @@
+import time
+from framework.business.cloud_desktop import CloudDesktopService
+from framework.config.settings import Config
+from framework.core.logger import get_logger
+from framework.models.result import StepResult
+
+logger = get_logger("BoilerplateService")
+
+class BoilerplateService:
+    def __init__(self):
+        self.service = CloudDesktopService()
+
+    def run_lifecycle_test(self, name, progress_callback=None):
+        """样本工程快捷启动生命周期验证"""
+        results = []
+        desktop_id = None
+        
+        def _log(msg):
+            logger.info(msg)
+            if progress_callback: progress_callback(msg)
+
+        try:
+            _log(f"Step 1: 快速启动样本工程 {name}")
+            t0 = time.time()
+            desktop_id = self.service.create_boilerplate_desktop(name)
+            results.append(StepResult("创建云桌面", True, "成功", time.time()-t0))
+            time.sleep(Config.WAIT['create'])
+            
+            # 复用基础流程
+            _log("Step 2: 关机")
+            t0 = time.time()
+            self.service.stop_desktop(desktop_id)
+            results.append(StepResult("关机云桌面", True, "成功", time.time()-t0))
+            time.sleep(Config.WAIT['stop'])
+            
+            _log("Step 3: 开机")
+            t0 = time.time()
+            self.service.start_desktop(desktop_id)
+            results.append(StepResult("开机云桌面", True, "成功", time.time()-t0))
+            time.sleep(Config.WAIT['start'])
+            
+            _log("Step 4: 按量转包月")
+            t0 = time.time()
+            self.service.switch_to_prepaid(desktop_id)
+            results.append(StepResult("按量转包月", True, "成功", time.time()-t0))
+            time.sleep(Config.WAIT['switch'])
+            
+            _log("Step 5: 删除")
+            t0 = time.time()
+            self.service.delete_desktop(desktop_id)
+            results.append(StepResult("删除云桌面", True, "成功", time.time()-t0))
+                
+        except Exception as e:
+            logger.error(f"Flow failed: {e}")
+            results.append(StepResult("执行中止", False, str(e)))
+            
+        return results, desktop_id or "N/A"

framework/business/cloud_desktop.py

@@ -0,0 +1,129 @@
+import time
+from framework.core.base_api import BaseAPI
+from framework.config.settings import Config
+from framework.core.logger import get_logger
+from framework.models.result import StepResult
+
+logger = get_logger("CloudDesktopService")
+
+class CloudDesktopService(BaseAPI):
+    def __init__(self):
+        super().__init__(Config.BASE_URL, Config.AUTH_TOKEN, Config.COOKIE)
+
+    def create_desktop(self, name, charge_type="PostPaid", image_id=None):
+        """创建云桌面"""
+        image_id = image_id or Config.CloudDesktop.POST_IMAGE_ID
+        body = {
+            "name": name,
+            "sku_id": Config.CloudDesktop.SKU_ID,
+            "system_disk_size": Config.CloudDesktop.DISK_SIZE,
+            "image_id": image_id,
+            "specification_id": Config.CloudDesktop.SPEC_ID,
+            "sub_path": "",
+            "timer_id": Config.CloudDesktop.TIMER_ID,
+            "charge_type": charge_type,
+        }
+        if charge_type == "PrePaid":
+            body["price_policy_id"] = Config.CloudDesktop.PRICE_POLICY_ID
+
+        headers = self.get_common_headers("/cloud-desktop")
+        headers["content-type"] = "application/json"
+        
+        data = self.request("POST", "/api/artificerApi/desktops", headers=headers, json=body)
+        desktop_id = data.get("data", {}).get("desktop_id") or data.get("desktop_id")
+        return desktop_id
+
+    def create_boilerplate_desktop(self, name):
+        """创建样本工程专用云桌面"""
+        return self.create_desktop(name, charge_type="PostPaid", image_id=Config.CloudDesktop.BP_IMAGE_ID)
+
+    def stop_desktop(self, desktop_id):
+        """关机"""
+        return self.request("GET", f"/api/artificerApi/desktops/{desktop_id}/stop")
+
+    def start_desktop(self, desktop_id):
+        """开机"""
+        params = {"desktop_id": desktop_id}
+        headers = self.get_common_headers("/cloud-desktop")
+        headers["content-length"] = "0"
+        return self.request("POST", "/api/artificerApi/desktops/start", params=params, headers=headers)
+
+    def delete_desktop(self, desktop_id):
+        """删除"""
+        params = {"desktop_id": desktop_id}
+        return self.request("DELETE", "/api/desktopManagerApi/api/v1/desktops/delete", params=params)
+
+    def switch_to_prepaid(self, desktop_id):
+        """按量转包月"""
+        body = {"desktop_id": desktop_id, "price_policy_id": Config.CloudDesktop.PRICE_POLICY_ID}
+        return self.request("POST", "/api/artificerApi/desktops/switch_to_prepaid", json=body)
+
+    def cancel_auto_renew(self, desktop_id):
+        """包月转按量 (取消续费)"""
+        params = {"desktop_id": desktop_id}
+        headers = self.get_common_headers("/cloud-desktop")
+        headers["content-length"] = "0"
+        return self.request("POST", "/api/artificerApi/desktops/cancel_auto_renew", params=params, headers=headers)
+
+    def run_lifecycle_test(self, desktop_name, flow_type="postpaid_to_prepaid", progress_callback=None):
+        """运行完整生命周期验证流"""
+        results = []
+        desktop_id = None
+        
+        def _log(msg):
+            logger.info(msg)
+            if progress_callback: progress_callback(msg)
+
+        try:
+            # 1. 创建
+            charge_type = "PrePaid" if flow_type == "prepaid_to_postpaid" else "PostPaid"
+            image_id = Config.CloudDesktop.PRE_IMAGE_ID if flow_type == "prepaid_to_postpaid" else Config.CloudDesktop.POST_IMAGE_ID
+            
+            _log(f"Step 1: 创建 ({charge_type})")
+            t0 = time.time()
+            desktop_id = self.create_desktop(desktop_name, charge_type=charge_type, image_id=image_id)
+            results.append(StepResult("创建云桌面", True, "成功", time.time()-t0))
+            
+            time.sleep(Config.WAIT['create'])
+            
+            # 2. 关机
+            _log("Step 2: 关机")
+            t0 = time.time()
+            self.stop_desktop(desktop_id)
+            results.append(StepResult("关机云桌面", True, "成功", time.time()-t0))
+            time.sleep(Config.WAIT['stop'])
+            
+            # 3. 开机
+            _log("Step 3: 开机")
+            t0 = time.time()
+            self.start_desktop(desktop_id)
+            results.append(StepResult("开机云桌面", True, "成功", time.time()-t0))
+            time.sleep(Config.WAIT['start'])
+            
+            # 4. 转换
+            if flow_type == "prepaid_to_postpaid":
+                _log("Step 4: 包月转按量")
+                t0 = time.time()
+                self.cancel_auto_renew(desktop_id)
+                results.append(StepResult("包月转按量", True, "成功", time.time()-t0))
+                time.sleep(Config.WAIT['switch'])
+                _log("Step 5: 最终关机")
+                t0 = time.time()
+                self.stop_desktop(desktop_id)
+                results.append(StepResult("关机云桌面", True, "成功", time.time()-t0))
+            else:
+                _log("Step 4: 按量转包月")
+                t0 = time.time()
+                self.switch_to_prepaid(desktop_id)
+                results.append(StepResult("按量转包月", True, "成功", time.time()-t0))
+                time.sleep(Config.WAIT['switch'])
+                _log("Step 5: 删除")
+                t0 = time.time()
+                self.delete_desktop(desktop_id)
+                results.append(StepResult("删除云桌面", True, "成功", time.time()-t0))
+                
+        except Exception as e:
+            logger.error(f"Flow failed: {e}")
+            results.append(StepResult("执行中止", False, str(e)))
+            
+        return results, desktop_id or "N/A"

framework/business/dev_machine.py

@@ -0,0 +1,81 @@
+import time
+from framework.core.base_api import BaseAPI
+from framework.config.settings import Config
+from framework.core.logger import get_logger
+from framework.models.result import StepResult
+
+logger = get_logger("DevMachineService")
+
+class DevMachineService(BaseAPI):
+    def __init__(self):
+        super().__init__(Config.BASE_URL, Config.AUTH_TOKEN, Config.COOKIE)
+
+    def create_machine(self, name):
+        """创建开发机"""
+        body = {
+            "displayName": name,
+            "imageID": Config.DevMachine.IMAGE_ID,
+            "skuID": Config.DevMachine.SKU_ID,
+            "sshPublicKey": Config.DevMachine.SSH_KEY,
+            "systemDiskSize": Config.DevMachine.DISK_SIZE,
+            "sourceDir": "",
+        }
+        headers = self.get_common_headers("/dev-machine")
+        headers["content-type"] = "application/json"
+        
+        data = self.request("POST", "/api/artificerApi/devMachine/createInstance", headers=headers, json=body)
+        instance_id = data.get("data", {}).get("instanceId") or data.get("instanceId")
+        return instance_id
+
+    def stop_machine(self, instance_id):
+        """关机"""
+        body = {"instanceId": instance_id}
+        return self.request("POST", "/api/artificerApi/devMachine/stopInstance", json=body)
+
+    def start_machine(self, instance_id):
+        """开机"""
+        body = {"instanceId": instance_id}
+        return self.request("POST", "/api/artificerApi/devMachine/startInstance", json=body)
+
+    def delete_machine(self, instance_id):
+        """删除"""
+        body = {"instanceId": instance_id}
+        return self.request("POST", "/api/artificerApi/devMachine/deleteInstance", json=body)
+
+    def run_lifecycle_test(self, name, progress_callback=None):
+        results = []
+        instance_id = None
+        
+        def _log(msg):
+            logger.info(msg)
+            if progress_callback: progress_callback(msg)
+
+        try:
+            _log(f"Step 1: 创建 {name}")
+            t0 = time.time()
+            instance_id = self.create_machine(name)
+            results.append(StepResult("创建开发机", True, "成功", time.time()-t0))
+            time.sleep(Config.WAIT['create'])
+            
+            _log("Step 2: 关机")
+            t0 = time.time()
+            self.stop_machine(instance_id)
+            results.append(StepResult("关机开发机", True, "成功", time.time()-t0))
+            time.sleep(Config.WAIT['stop'])
+            
+            _log("Step 3: 开机")
+            t0 = time.time()
+            self.start_machine(instance_id)
+            results.append(StepResult("开机开发机", True, "成功", time.time()-t0))
+            time.sleep(Config.WAIT['start'])
+            
+            _log("Step 4: 删除")
+            t0 = time.time()
+            self.delete_machine(instance_id)
+            results.append(StepResult("删除开发机", True, "成功", time.time()-t0))
+            
+        except Exception as e:
+            logger.error(f"Flow failed: {e}")
+            results.append(StepResult("执行中止", False, str(e)))
+            
+        return results, instance_id or "N/A"

framework/business/meal_reminder.py

@@ -0,0 +1,95 @@
+import random
+from datetime import datetime
+from framework.core.base_api import BaseAPI
+from framework.config.settings import Config
+from framework.core.logger import get_logger
+
+logger = get_logger("MealReminderService")
+
+class MealReminderService(BaseAPI):
+    MEAL_PERIODS = {
+        "breakfast": {
+            "range": (6, 9), "emoji": "🌅", "label": "早餐", "keywords": "早餐|早点|咖啡",
+            "greetings": ["早安！中关村的早晨从热腾腾的早点开始 ☀️", "美好的早晨，看看附近有什么好吃的早餐 ☕"]
+        },
+        "lunch": {
+            "range": (10, 13), "emoji": "☀️", "label": "午餐", "keywords": "餐厅|快餐|中餐",
+            "greetings": ["午饭时间到！放下代码，看看公司附近吃什么 🍚", "到点干饭了，为您搜罗了附近的优质午餐 🤔"]
+        },
+        "afternoon_tea": {
+            "range": (14, 16), "emoji": "🍰", "label": "下午茶", "keywords": "甜品|饮品|咖啡",
+            "greetings": ["下午茶时间～来点甜的补充能量 🧁", "三点几嚟，饮茶先啦！🫖"]
+        },
+        "dinner": {
+            "range": (17, 20), "emoji": "🌆", "label": "晚餐", "keywords": "特色菜|火锅|烧烤|餐厅",
+            "greetings": ["下班啦！中关村的夜晚，吃顿好的犒劳自己 🎉", "晚餐时间到，看看壹号周边有哪些人气餐厅 🍽️"]
+        },
+        "late_night": {
+            "range": (21, 5), "emoji": "🌙", "label": "夜宵", "keywords": "烧烤|宵夜|快餐",
+            "greetings": ["深夜食堂正式营业！🏮", "加班辛苦了，看看附近有什么深夜美食 🌙"]
+        },
+    }
+
+    def __init__(self):
+        super().__init__("https://restapi.amap.com") # 高德 API 基地址
+
+    def get_current_period(self) -> dict:
+        hour = datetime.now().hour
+        for period_info in self.MEAL_PERIODS.values():
+            start, end = period_info["range"]
+            if start <= end:
+                if start <= hour <= end: return period_info
+            else:
+                if hour >= start or hour <= end: return period_info
+        return self.MEAL_PERIODS["lunch"]
+
+    def fetch_nearby_restaurants(self, keywords: str) -> list:
+        endpoint = "/v3/place/around"
+        params = {
+            "key": Config.AMAP_KEY,
+            "location": Config.AMAP_COORDINATES,
+            "keywords": keywords,
+            "types": "050000", # 餐饮服务
+            "radius": 1000,
+            "sortrule": "weight",
+            "offset": 10,
+            "page": 1,
+            "extensions": "all"
+        }
+        try:
+            # BaseAPI.request 默认会拼 BASE_URL，这里高德是不同的地址，手动处理或实例化时指定
+            import requests # 临时直接用 requests，或者修改 BaseAPI 支持绝对路径
+            resp = requests.get(f"{self.base_url}{endpoint}", params=params, timeout=5)
+            data = resp.json()
+            if data.get("status") == "1":
+                return data.get("pois", [])
+        except Exception as e:
+            logger.error(f"高德 API 调用失败: {e}")
+        return []
+
+    def build_card(self) -> dict:
+        period = self.get_current_period()
+        greeting = random.choice(period["greetings"])
+        pois = self.fetch_nearby_restaurants(period["keywords"])
+        
+        elements = [{"tag": "div", "text": {"tag": "lark_md", "content": f"**{greeting}**"}}, {"tag": "hr"}]
+        
+        if pois:
+            for poi in pois[:5]:
+                name = poi.get("name", "未知餐厅")
+                biz_ext = poi.get("biz_ext", {})
+                rating = biz_ext.get("rating") if isinstance(biz_ext, dict) else "暂无"
+                distance = poi.get("distance", "未知")
+                elements.append({
+                    "tag": "div",
+                    "text": {"tag": "lark_md", "content": f"🏠 **{name}**\n⭐️ 评分: {rating} | 📍 距离: {distance}m"}
+                })
+        else:
+            elements.append({"tag": "div", "text": {"tag": "lark_md", "content": "⚠️ 暂时没搜到附近的店"}})
+
+        elements.extend([{"tag": "hr"}, {"tag": "note", "elements": [{"tag": "plain_text", "content": f"🕒 {datetime.now().strftime('%H:%M')}"}]}])
+        
+        return {
+            "header": {"title": {"tag": "plain_text", "content": f"{period['emoji']} 饭否 · {period['label']}实时推荐"}, "template": "orange"},
+            "elements": elements
+        }

framework/config/settings.py

@@ -0,0 +1,49 @@
+import os
+
+class Config:
+    # --- 基础配置 ---
+    BASE_URL = "https://robogo-fat.d-robotics.cc"
+    CLOUD_BASE_URL = "https://cloud-fat.d-robotics.cc"
+    VOLCANO_BASE_URL = "http://120.48.128.161:16001/api"
+    
+    # --- 认证信息 ---
+    AUTH_TOKEN = "Bearer eyJhbGciOiJSUzI1NiIsImtpZCI6Inl4LXl5ZHMta2V5IiwidHlwIjoiSldUIn0.eyJ1c2VyX2lkIjoiY2M5MDE5NjEtZTIxNS00NDRjLWIwMTMtMmQyMDZkZjU2ODc1IiwidXNlcm5hbWUiOiJ0YW9oYW5neXUxIiwiaXNzIjoidXNlci1jZW50ZXIiLCJzdWIiOiJ0YW9oYW5neXUxIiwiZXhwIjoxNzczMzAwMTUzLCJuYmYiOjE3NzI2OTUzNTMsImlhdCI6MTc3MjY5NTM1M30.HSjbyXUJOR2QqezPxfCAiUgunquaNjOceSeOFS-kVg6c6Y74hw2vnfWzoBaXprKIBJMqUJaB5S4mek7icDbX-MKTOOz0ClDrHNxJGDzkMV5WNsuJSPnSjqzjm0gUQ73A3Dh3FeagPvkG6fi3gzdvXCVEz4MbHI3BeogCr-Thode0DqGNm2yG-Dn3cqctMak-PUiT9V1konENRUj-jYRq28uFVM8OWh7e70P-ToyRBASPIJdTUN408eZ_4m_6mFtYgvPV6WI7-B6VWea31oX9bWX5MHvARseZiTrOuoonG9Y7QB_TUX90WwgTUbqiTRU89kOCw4UhrwkTKXURTb5xhw"
+    COOKIE = "_bl_uid=0amOhkt1m77etFp8gi37oyk7wjg8; acw_tc=276a376517727797494882119e27aa25f45d6b05e3eb7501724f9c9d13faaa"
+    CLOUD_AUTH_TOKEN = "Bearer eyJhbGciOiJSUzI1NiIsImtpZCI6Inl4LXl5ZHMta2V5IiwidHlwIjoiSldUIn0.eyJ1c2VyX2lkIjoiOWJiNmZjZTYtZDk3NC0xMWVmLWJiMjUtZmEyMDIwMzQxMWNlIiwidXNlcm5hbWUiOiJBZG1pbmlzdHJhdG9yIiwiaXNzIjoidXNlci1jZW50ZXIiLCJzdWIiOiJBZG1pbmlzdHJhdG9yIiwiZXhwIjoxNzczNjUxNzE5LCJuYmYiOjE3NzMwNDY5MTksImlhdCI6MTc3MzA0NjkxOX0.DzjEc48rVPr6zOqRDqURxR-Fsu3DfZxa4gbapdIS3wSUEusC-hGSw9TDAPysfsFSigfZP1GDP9J8OJhZwPPbaHSum9IvUhIagxqaD8xKuzDlDc_cAGij7nHVed6h4SG-lFJNq-jaXhx08GQ6E1d79jRdF-2IFl23trCjoMVa_70qIzS6ZWhvUTKcJ96hzOQhYJhUYVJjdphChPABPkpMH8ljU5gD-VHsx2Lr_xTtOI1HiEmkukiS0frwbqLkGl2Lja6-tvZhc3hQifJMIDNLRNvtzvbuC412ljUQxhA7euE4JIfDuU3S3l7BQ7bQkSIwd0HzaVqg_Agw"
+
+    # --- 飞书机器人配置 ---
+    FEISHU_APP_ID = "cli_a9aeb4fb2c78dcb6"
+    FEISHU_APP_SECRET = "7nYs724srjEn4jgNPJW9cfuqL4e2OVT6"
+    FEISHU_ALERT_CHAT_ID = "oc_866c9db02166986e0607f0f62d1c698e"
+    
+    # --- 资源参数 ---
+    class CloudDesktop:
+        SKU_ID = "4c7d50d9-b6c5-4b90-b829-3b6e8a08a13f"
+        DISK_SIZE = 500
+        SPEC_ID = "eds.graphics_flagship_pro.16c32g.12g"
+        TIMER_ID = "NO_SHUTDOWN"
+        PRICE_POLICY_ID = "57bf71a4-f80c-ff87-6a0b-d74e4ed981a7"
+        
+        POST_IMAGE_ID = "34ef57b5-29e5-45e4-9cd4-202fab3ea9a8"
+        PRE_IMAGE_ID = "4b2a81d4-7f8b-4515-b4b5-bb212557aec1"
+        BP_IMAGE_ID = "2af0347a-b825-4272-a313-dde1d350d73d"
+
+    class DevMachine:
+        SKU_ID = "e154fd3a-4719-db77-3911-21c6355349ec"
+        IMAGE_ID = "ed2ae2e7-0373-4d15-9fdb-b44c83f81231"
+        DISK_SIZE = 512
+        SSH_KEY = "ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIBGp6Be0pt0Xy3Ipxm+AQTz6JQq8DAzIU6XHqD+/gzH6"
+
+    # --- 高德地图 API 配置 ---
+    AMAP_KEY = "71ef48b583ad060c20416868c30f6b99"
+    AMAP_LOCATION_NAME = "北京市海淀区中关村壹号"
+    AMAP_COORDINATES = "116.239322,40.074312"
+
+    # --- 超时等待 (秒) ---
+    WAIT = {
+        "create": 120,
+        "stop": 60,
+        "start": 60,
+        "switch": 30
+    }
+    INSPECTION_INTERVAL = 300

framework/core/base_api.py

@@ -0,0 +1,63 @@
+import requests
+import uuid
+import time
+from framework.core.exceptions import TokenExpiredError, APIError
+from framework.core.logger import get_logger
+
+logger = get_logger("BaseAPI")
+
+class BaseAPI:
+    def __init__(self, base_url, token=None, cookie=None):
+        self.base_url = base_url
+        self.token = token
+        self.cookie = cookie
+        self.session = requests.Session()
+
+    def get_common_headers(self, referer_path=""):
+        headers = {
+            "accept": "application/json, text/plain, */*",
+            "accept-language": "zh-CN,zh;q=0.9",
+            "authorization": self.token,
+            "cookie": self.cookie,
+            "origin": self.base_url,
+            "referer": f"{self.base_url}{referer_path}",
+            "sourceapp": "artificer",
+            "user-agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/144.0.0.0 Safari/537.36",
+            "x-request-id": uuid.uuid4().hex[:21],
+        }
+        return headers
+
+    def request(self, method, endpoint, headers=None, **kwargs):
+        url = f"{self.base_url}{endpoint}"
+        
+        # Merge default headers if none provided
+        if headers is None:
+            headers = self.get_common_headers()
+        
+        try:
+            resp = self.session.request(method, url, headers=headers, **kwargs)
+            return self._check_response(resp, endpoint)
+        except Exception as e:
+            logger.error(f"Request failed: {method} {url} - {str(e)}")
+            raise e
+
+    def _check_response(self, resp, step_name):
+        if resp.status_code in (401, 403):
+            raise TokenExpiredError(f"[{step_name}] HTTP {resp.status_code}: 认证失败")
+        
+        if resp.status_code < 200 or resp.status_code >= 300:
+            raise APIError(f"[{step_name}] HTTP {resp.status_code}: {resp.text}")
+            
+        try:
+            data = resp.json()
+        except Exception:
+            raise APIError(f"[{step_name}] 非 JSON 响应: {resp.text}")
+        
+        body_status = data.get("status", 0)
+        if body_status != 0:
+            msg = data.get("message", "")
+            if body_status in (401, 403) or "token" in msg.lower():
+                raise TokenExpiredError(f"[{step_name}] 认证失效: {msg}")
+            raise APIError(f"[{step_name}] 业务错误: {msg}")
+            
+        return data

framework/core/base_ui.py

@@ -0,0 +1,22 @@
+"""
+UI 自动化基类占位 (推荐使用 Playwright)
+"""
+from framework.core.logger import get_logger
+
+logger = get_logger("BaseUI")
+
+class BaseUI:
+    def __init__(self, browser_type="chromium"):
+        self.browser_type = browser_type
+        # TODO: 集成 playwright/selenium 初始化
+        pass
+
+    def navigate(self, url):
+        logger.info(f"Navigate to {url}")
+        # self.page.goto(url)
+        pass
+
+    def click(self, selector):
+        logger.info(f"Clicking {selector}")
+        # self.page.click(selector)
+        pass

framework/core/exceptions.py

@@ -0,0 +1,11 @@
+class FrameworkError(Exception):
+    """框架基础异常"""
+    pass
+
+class TokenExpiredError(FrameworkError):
+    """认证过期异常"""
+    pass
+
+class APIError(FrameworkError):
+    """API 调用业务异常"""
+    pass

framework/core/logger.py

@@ -0,0 +1,21 @@
+import logging
+import os
+
+def get_logger(name: str):
+    logger = logging.getLogger(name)
+    if not logger.handlers:
+        logger.setLevel(logging.INFO)
+        formatter = logging.Formatter('%(asctime)s [%(name)s] [%(levelname)s] %(message)s', datefmt='%Y-%m-%d %H:%M:%S')
+        
+        # 控制台输出
+        ch = logging.StreamHandler()
+        ch.setFormatter(formatter)
+        logger.addHandler(ch)
+        
+        # 文件输出
+        log_file = os.path.join(os.getcwd(), 'automation.log')
+        fh = logging.FileHandler(log_file)
+        fh.setFormatter(formatter)
+        logger.addHandler(fh)
+        
+    return logger

framework/models/result.py

@@ -0,0 +1,14 @@
+from datetime import datetime
+
+class StepResult:
+    """自动化步骤执行结果模型"""
+    def __init__(self, step_name: str, success: bool, message: str, elapsed: float = 0):
+        self.step_name = step_name
+        self.success = success
+        self.message = message
+        self.elapsed = elapsed
+        self.timestamp = datetime.now()
+
+    def __str__(self):
+        icon = "✅" if self.success else "❌"
+        return f"{icon} {self.step_name}: {self.message} ({self.elapsed:.2f}s)"

framework/requirements.txt

@@ -0,0 +1,2 @@
+lark-oapi>=1.3.0
+requests>=2.28.0

framework/scripts/desktop_lifecycle.py

@@ -0,0 +1,52 @@
+import time
+from framework.business.cloud_desktop import CloudDesktopService
+from framework.models.result import StepResult
+from framework.config.settings import Config
+from framework.core.logger import get_logger
+
+logger = get_logger("LifecycleScript")
+
+def run_postpaid_lifecycle(name):
+    """
+    业务逻辑：按量转包月完整生命周期示例
+    """
+    service = CloudDesktopService()
+    results = []
+    
+    # 1. 创建
+    logger.info(f"Step 1: 创建云桌面 {name}")
+    t0 = time.time()
+    try:
+        desktop_id = service.create_desktop(name, charge_type="PostPaid")
+        results.append(StepResult("创建云桌面", True, f"成功 ID: {desktop_id}", time.time()-t0))
+    except Exception as e:
+        results.append(StepResult("创建云桌面", False, str(e), time.time()-t0))
+        return results, None
+
+    # 等待部署
+    time.sleep(Config.WAIT['create'])
+
+    # 2. 关机
+    logger.info("Step 2: 关机")
+    t0 = time.time()
+    try:
+        service.stop_desktop(desktop_id)
+        results.append(StepResult("关机", True, "成功", time.time()-t0))
+    except Exception as e:
+        results.append(StepResult("关机", False, str(e), time.time()-t0))
+
+    # 3. 删除 (示例简化，不写全部逻辑)
+    logger.info("Step 3: 删除")
+    t0 = time.time()
+    try:
+        service.delete_desktop(desktop_id)
+        results.append(StepResult("删除", True, "成功", time.time()-t0))
+    except Exception as e:
+        results.append(StepResult("删除", False, str(e), time.time()-t0))
+
+    return results, desktop_id
+
+if __name__ == "__main__":
+    res, d_id = run_postpaid_lifecycle("RobotFramework-Test")
+    for r in res:
+        print(r)

framework/scripts/test_billing.py

@@ -0,0 +1,30 @@
+
+import requests
+import json
+import config
+
+url = f"https://cloud-fat.d-robotics.cc/api/dcloudResourceApi/consumptions"
+params = {
+    "page": 1,
+    "pageSize": 50,
+    "instanceName": "验证01"
+}
+headers = {
+    "Accept": "application/json, text/plain, */*",
+    "Authorization": config.CLOUD_AUTH_TOKEN,
+    "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/144.0.0.0 Safari/537.36",
+}
+
+print(f"Requesting URL: {url}")
+resp = requests.get(url, params=params, headers=headers)
+print(f"Status Code: {resp.status_code}")
+try:
+    data = resp.json()
+    print(f"Status in JSON: {data.get('status')}")
+    print(f"Data content: {data.get('data')}")
+    items = data.get("data", {}).get("items", [])
+    for item in items:
+        print(f"- {item.get('instanceName')}")
+except Exception as e:
+    print(f"Error parsing JSON: {e}")
+    print(f"Raw Response: {resp.text[:500]}")

run_bot.py

@@ -0,0 +1,4 @@
+from framework.bot import main
+
+if __name__ == "__main__":
+    main()