refactor: migrate framework modules to dataloop automation and update business logic components

framework/business/data_management.py

@@ -121,9 +121,9 @@ class DataManagement:
         if scope == "all":
             active_modules = ["file", "dev", "cloud", "mirror", "quant" , "3d"]
         elif scope == "smoke":
-            active_modules = ["file"] # 核心资源生命周期
+            active_modules = ["monkey"] # 核心资源生命周期
         elif scope == "core":
-            active_modules = ["cloud","dev", "mirror", "quant"] # 业务闭环
+            active_modules = ["cloud","dev", "mirror", "quant" ,"file"] # 业务闭环
         else:
             active_modules = ["cloud"] # 默认兜底只跑云桌面
             
@@ -244,6 +244,34 @@ class DataManagement:
         
         logger.info(f"🎉 Robogo {env_name} 环境 {scope} 巡检执行圆满完成！")
 
+    def ensure_prerequisites(self):
+        """
+        前置环境自检：确保两个关键平台目录存在，避免人工干预
+          - 数据管理 / 3d生成(勿删)   ← three_d_generation_scenario 的保存路径
+          - 3D资产  / ui_test         ← three_d_assets_scenario 的归档资产包
+        """
+        logger.info("🏗️  开始前置环境自检 (ensure_prerequisites)...")
+
+        # ── Step 1: 数据管理 / 3d生成(勿删) ─────────────────────────────────────
+        logger.info("🔎 [Step 1/2] 检查 '3d生成(勿删)' 文件夹...")
+        try:
+            self.fm.navigate_to()
+            self.fm.ensure_folder("3d生成(勿删)")
+        except BaseException as e:
+            # 用 BaseException 捕获 Playwright 可能抛出的 SystemExit/KeyboardInterrupt 等
+            logger.warning(f"⚠️ '3d生成(勿删)' 文件夹自检失败，请手动确认: {type(e).__name__}: {e}")
+
+        # ── Step 2: 3D资产 / ui_test ─────────────────────────────────────────────
+        logger.info("🔎 [Step 2/2] 检查 'ui_test' 资产包...")
+        try:
+            self.ta.navigate_to()
+            self.ta.ensure_asset_package("ui_test")
+        except BaseException as e:
+            logger.warning(f"⚠️ 'ui_test' 资产包自检失败，请手动确认: {type(e).__name__}: {e}")
+
+        logger.info("✅ 前置环境自检完成")
+
+
     def run(self, user, pwd):
         """主入口"""
         try:
@@ -251,6 +279,9 @@ class DataManagement:
             if not self.login(user, pwd):
                 return
 
+            # 前置环境自检（自动创建依赖文件夹）
+            self.ensure_prerequisites()
+
             # 开始执行指挥任务
             self.run_all_scenarios()
             

framework/business/quantization_page.py

@@ -79,16 +79,18 @@ class QuantizationPage(BasePage):
         option.click()
         time.sleep(1)
 
-    def select_image(self, image_name="dcloud_ai_toolchain_ubuntu_22_s100_cpu-2"):
+    def select_image(self, image_name="dcloud_ai_toolchain_ubuntu_22_s100_s600_cpu"):
         """选择镜像流程"""
         logger.info(f"点击选择镜像按钮，拉起弹窗...")
         self.page.locator("button.p-button-link:has-text('选择镜像')").click()
         
-        # 在弹窗中定位镜像项
+        # 在弹窗中定位镜像项 (增强鲁棒性)
         logger.info(f"📍 在弹窗中寻找镜像: {image_name}")
-        image_row = self.page.locator(f"div:has-text('{image_name}')").last
+        # 直接使用文本搜索，并等待其可见
-        image_row.scroll_into_view_if_needed()
+        target = self.page.get_by_text(image_name, exact=True).first
-        image_row.click()
+        target.wait_for(state="visible", timeout=15000)
+        target.scroll_into_view_if_needed()
+        target.click()
         
         # 点击弹窗内部的确定 (限制定位器在 p-dialog 容器内)
         logger.info("⏳ 点击[镜像弹窗]确认按钮...")

framework/business/three_d_assets_page.py

@@ -18,6 +18,44 @@ class ThreeDAssetsPage(BasePage):
         self.smart_click(self.MENU_TEXT)
         time.sleep(2)
 
+    def ensure_asset_package(self, package_name="ui_test"):
+        """前置检查：若资产包不存在则自动创建"""
+        logger.info(f"🔍 检查资产包 [{package_name}] 是否已存在...")
+        selector = f"div.p-card:has-text('{package_name}')"
+        try:
+            self.page.wait_for_selector(selector, timeout=5000)
+            logger.info(f"✅ 资产包 [{package_name}] 已存在，无需创建")
+            return True
+        except:
+            logger.info(f"📦 资产包 [{package_name}] 不存在，准备自动创建...")
+
+        try:
+            # 点击"新建资产包"按钮
+            # Playwright 不支持 CSS 逗号复合选择器，用 .or_() 代替
+            create_btn = (
+                self.page.locator("button:has-text('新建资产包')")
+                .or_(self.page.locator("button:has-text('新建')"))
+            )
+            create_btn.first.click()
+            time.sleep(1)
+
+            # 填写资产包名称
+            name_input = self.page.locator("input[placeholder*='名称'], input#packageName, .p-dialog input").first
+            name_input.fill(package_name)
+            time.sleep(0.5)
+
+            # 点击确认按钮
+            self.page.locator(".p-dialog button:has-text('确定'), .p-dialog button:has-text('确认')").last.click()
+            time.sleep(2)
+
+            # 验证创建成功
+            self.page.wait_for_selector(selector, timeout=8000)
+            logger.info(f"✅ 资产包 [{package_name}] 创建成功")
+            return True
+        except Exception as e:
+            logger.warning(f"⚠️ 资产包 [{package_name}] 自动创建失败，请手动确认: {e}")
+            return False
+
     def enter_asset_package(self, package_name="ui_test"):
         """进入指定的资产包"""
         logger.info(f"📂 正在寻找并进入资产包: {package_name}")

framework/business/three_d_generation_page.py

@@ -84,22 +84,44 @@ class ThreeDGenerationPage(BasePage):
         raise Exception(error_msg)
 
 
-    def wait_for_result_and_save(self, timeout=600):
+    def wait_for_result_and_save(self, timeout=900):
-        """等待右上角结果出现并点击保存"""
+        """等待右上角[保存]按钮变为可点击状态后，立刻点击"""
-        logger.info("⏳ 等待生成结果...")
+        logger.info(f"⏳ 等待 3D 生成结果 (最长 {timeout}s)...")
-        # 右上角模型参数卡片里出现内容或“保存”按钮变得可用
         start_time = time.time()
+        last_log = start_time
+
         while time.time() - start_time < timeout:
-            content = self.page.content()
+            elapsed = int(time.time() - start_time)
-            # 检查是否有反映生成完成的标志，或者截图中的“保存”按钮
+
-            if "保存" in content:
+            # 每 30s 打印一次进度，方便追踪
-                save_btn = self.page.locator("button:has-text('保存')").nth(0)
+            if time.time() - last_log >= 30:
-                if save_btn.is_visible() and save_btn.is_enabled():
+                logger.info(f"   ⏱️ 已等待 {elapsed}s，继续监听保存按鈕...")
-                    logger.info("✨ 生成结果已出现，点击[保存]按钮")
+                last_log = time.time()
-                    save_btn.click()
+
-                    return True
+            try:
-            time.sleep(5)
+                save_btns = self.page.locator("button:has-text('\u4fdd\u5b58')")
-        logger.error("❌ 等待结果超时")
+                count = save_btns.count()
+
+                for i in range(count):
+                    btn = save_btns.nth(i)
+                    try:
+                        # 跳过弹窗内的按鈕（弹窗内的“确定”不是目标）
+                        in_dialog = btn.evaluate("el => !!el.closest('.p-dialog')")
+                        if in_dialog:
+                            continue
+                        if btn.is_visible() and btn.is_enabled():
+                            logger.info(f"✨ 第 {elapsed}s 时检测到[保存]按鈕就绪，立刻点击")
+                            btn.click()
+                            return True
+                    except Exception:
+                        continue
+
+            except Exception:
+                pass  # 按鈕还未出现，继续等待
+
+            time.sleep(3)
+
+        logger.error(f"❌ 等待结果超时 ({timeout}s)，保存按鈕始终未就绪")
         return False
 
     def handle_save_dialog(self):

framework/scripts/quantization_scenario.py

@@ -19,7 +19,7 @@ def _create_and_submit_task(quant_page: QuantizationPage, task_name: str):
     quant_page.select_mode_fast_eval()
     
     # 4. 选择镜像
-    quant_page.select_image("dcloud_ai_toolchain_ubuntu_22_s100_cpu-2")
+    quant_page.select_image("dcloud_ai_toolchain_ubuntu_22_s100_s600_cpu")
     
     # 5. 选取模型文件
     quant_page.upload_model_file("model", "result.onnx")

platform_app.py

@@ -40,7 +40,7 @@ PRODUCTS = {
         "name": "数据闭环",
         "desc": "数据闭环平台端到端业务流水线验证",
         "icon": "🔄",
-        "entry": None # 待接入
+        "entry": "run_data_loop.py"
     }
 }
 
@@ -182,7 +182,12 @@ def run_task_process(task):
 
     total_pass, total_fail = 0, 0
     current_run = 0
-    max_runs = run_limit + retry_count # 潜在的最大运行次数
+    # 有效运行轮次（不含重试补充轮），用于 pass/total 统计
+    planned_runs = run_limit
+    # 记录上一次运行的成败，用于最终结果判断
+    last_run_success = None
+    # 标记本轮是否为重试补充轮（重试不计入 total_runs）
+    is_retry_run = False
 
     push(f"🎬 任务开始 — 环境: {env['ROBOGO_ENV']} | 范围: {env['ROBOGO_SCOPE']}", "INFO")
 
@@ -191,7 +196,7 @@ def run_task_process(task):
 
     while current_run < run_limit:
         current_run += 1
-        push(f"🚀 第 {current_run}/{run_limit} 次运行中...", "INFO")
+        push(f"🚀 第 {current_run}/{planned_runs} 次运行中...", "INFO")
         
         try:
             # 开启进程组 (Process Group)，以便停止时能连带子进程一起干掉
@@ -208,9 +213,18 @@ def run_task_process(task):
             process_pids.pop(task_id, None)
 
             if proc.returncode == 0:
-                total_pass += 1
+                last_run_success = True
-                push(f"✅ 第 {current_run} 次成功", "SUCCESS")
+                if is_retry_run:
+                    # 重试成功：撤销上一次的失败计数，补记为成功
+                    total_fail = max(0, total_fail - 1)
+                    total_pass += 1
+                    push(f"✅ 重试第 {current_run} 次成功（已覆盖上次失败）", "SUCCESS")
+                else:
+                    total_pass += 1
+                    push(f"✅ 第 {current_run} 次成功", "SUCCESS")
+                is_retry_run = False
             else:
+                last_run_success = False
                 total_fail += 1
                 push(f"❌ 第 {current_run} 次失败", "ERROR")
                 # 失败重跑
@@ -218,18 +232,25 @@ def run_task_process(task):
                     push(f"🔁 触发重跑 (剩余 {retry_count} 次)，等待 {retry_delay}s...", "WARN")
                     time.sleep(retry_delay)
                     retry_count -= 1
-                    run_limit += 1 # 延长循环
+                    run_limit += 1  # 延长循环
+                    is_retry_run = True  # 下一轮为重试补充轮
+                else:
+                    is_retry_run = False
         except Exception as e:
             push(f"💥 系统爆破: {e}", "ERROR")
+            last_run_success = False
             total_fail += 1
+            is_retry_run = False
 
     # 收尾
     finished_at = datetime.now().isoformat()
+    # 最终结果：以最后一次运行成败为准（重试成功视为整体成功）
+    final_result = "PASS" if last_run_success else "FAIL"
     report = {
         "task_id": task_id, "task_name": task["name"], "product": task["product"],
-        "total_runs": current_run, "pass": total_pass, "fail": total_fail,
+        "total_runs": planned_runs, "pass": total_pass, "fail": total_fail,
         "started_at": task["started_at"], "finished_at": finished_at,
-        "result": "PASS" if total_fail == 0 else "FAIL"
+        "result": final_result
     }
     
     # 保存物理日志
@@ -240,7 +261,7 @@ def run_task_process(task):
     except: pass
 
     reports_db[task_id] = report
-    task["status"] = "pass" if total_fail == 0 else "fail"
+    task["status"] = "pass" if final_result == "PASS" else "fail"
     task["finished_at"] = finished_at
     task["report_id"] = task_id
     
@@ -285,7 +306,7 @@ def create_task():
         "scheduled_at": body.get("scheduled_at"),
         "schedule_type": body.get("schedule_type", "once"),
         "schedule_window": body.get("schedule_window", "00:00-23:59"),
-        "alert_channels": body.get("alert_channels", []),
+        "alert_channels": body.get("alert_channels", ["lark"]),
         "alert_rule": body.get("alert_rule", "always"),
         "entry": p.get("entry")
     }
@@ -443,14 +464,23 @@ def get_stats():
         f_tasks = []
         sorted_fails = sorted(failed_reports, key=lambda x: x.get("finished_at", ""), reverse=True)[:10]
         for r in sorted_fails:
+            tid = r.get("task_id")
             f_at = r.get("finished_at", "T00:00")
             time_str = f_at.split("T")[1][:5] if "T" in f_at else "00:00"
+            # 尝试获取一个缩略图
+            thumbnail = None
+            if os.path.exists(SCREENSHOTS_DIR):
+                ss = [s for s in os.listdir(SCREENSHOTS_DIR) if s.startswith(tid)]
+                if ss:
+                    thumbnail = ss[0]
+
             f_tasks.append({
-                "id": r.get("task_id"),
+                "id": tid,
                 "name": r.get("task_name", "未知任务"),
                 "product": PRODUCTS.get(r.get("product"), {}).get("name", r.get("product")),
                 "finished_at": time_str,
-                "reason": "执行异常 (请查看报告)"
+                "reason": "执行异常 (请查看报告)",
+                "thumbnail": thumbnail
             })
 
         return jsonify({
@@ -474,8 +504,21 @@ def get_stats():
 
 @app.route("/api/reports")
 def list_reports():
-    # 过滤掉软删除的任务报告，并且过滤掉“孤儿报告”（即 task 彻底不存在的残留垃圾）
+    # 过滤掉软删除的任务报告，并且过滤掉“孤儿报告”
-    r_list = [r for tid, r in reports_db.items() if tid in tasks_db and not tasks_db[tid].get("is_deleted")]
+    r_list = []
+    for tid, r in reports_db.items():
+        if tid in tasks_db and not tasks_db[tid].get("is_deleted"):
+            item = r.copy()
+            # 动态寻找该任务的一个缩略图
+            item["thumbnail"] = None
+            if os.path.exists(SCREENSHOTS_DIR):
+                try:
+                    for f in os.listdir(SCREENSHOTS_DIR):
+                        if f.startswith(tid):
+                            item["thumbnail"] = f
+                            break
+                except: pass
+            r_list.append(item)
     return jsonify(r_list)
 
 @app.route("/api/reports/<tid>")

run_data_loop.py

@@ -0,0 +1,106 @@
+"""
+run_data_loop.py
+────────────────
+数据闭环巡检入口 — 指挥官模式
+
+不再通过 pytest 调度 410 个分散用例,
+而是直接实例化 DataLoopCommander:
+  1 个浏览器 → 1 次登录 → 9 个场景串行跑完 → 关闭
+
+由 AutoFlow 平台 (platform_app.py) 调用:
+  python run_data_loop.py
+"""
+import os
+import sys
+import json
+from pathlib import Path
+
+
+def main():
+    print("🎬 数据闭环巡检 — 指挥官模式启动")
+
+    # ── 环境变量 (由 platform_app.py 注入) ────────────────────────────────────
+    task_id = os.environ.get("ROBOGO_TASK_ID", "default_task")
+    scope = os.environ.get("ROBOGO_SCOPE", "all")
+
+    # 凭证传递
+    if os.environ.get("AUTH_ACCOUNT"):
+        os.environ["TEST_USERNAME"] = os.environ["AUTH_ACCOUNT"]
+    if os.environ.get("AUTH_PASSWORD"):
+        os.environ["TEST_PASSWORD"] = os.environ["AUTH_PASSWORD"]
+
+    # 环境 URL 传递
+    robo_env = os.environ.get("ROBOGO_ENV", "FAT").upper()
+    if robo_env == "PROD":
+        os.environ.setdefault("BASE_URL", "https://cloud.d-robotics.cc/d-cloud/welcome")
+        os.environ.setdefault("LOGIN_URL", "https://sso.d-robotics.cc/")
+    else:
+        os.environ.setdefault("BASE_URL", "https://cloud-fat.d-robotics.cc/d-cloud/welcome")
+        os.environ.setdefault("LOGIN_URL", "https://sso-fat.d-robotics.cc/")
+
+    # ── 关键：在 chdir 之前，把所有路径统一转成绝对路径写回环境变量 ────────────
+    #    平台注入的已经是绝对路径, 但本地调试 fallback 是相对路径, 必须在此处锁定
+    platform_root = Path.cwd().absolute()
+    reports_dir = os.environ.get("ROBOGO_REPORTS_DIR", str(platform_root / "platform_reports"))
+    screenshots_dir = os.environ.get("ROBOGO_SCREENSHOTS_DIR", str(platform_root / "platform_artifacts" / "screenshots"))
+
+    # 确保路径是绝对路径 (如果平台传的是相对路径也兜住)
+    reports_dir = str(Path(reports_dir).absolute())
+    screenshots_dir = str(Path(screenshots_dir).absolute())
+
+    # 写回环境变量 — Commander 内部直接读这些
+    os.environ["ROBOGO_REPORTS_DIR"] = reports_dir
+    os.environ["ROBOGO_SCREENSHOTS_DIR"] = screenshots_dir
+
+    # ── 切换工作目录到 dataloop 项目根 ────────────────────────────────────────
+    dataloop_root = Path("dataloop/data-loop-automation").absolute()
+    os.chdir(dataloop_root)
+    sys.path.insert(0, str(dataloop_root))
+
+    # ── 导入并执行指挥官 ─────────────────────────────────────────────────────
+    from business.data_loop_commander import DataLoopCommander
+
+    headless_str = os.environ.get("HEADLESS", "false").lower()
+    headless = headless_str == "true"
+
+    print(f"🚀 执行参数: 环境={robo_env} | 范围={scope} | 无头={headless}")
+    print(f"   任务ID={task_id}")
+    print(f"   报告目录={reports_dir}")
+    print(f"   截图目录={screenshots_dir}")
+
+    commander = DataLoopCommander(headless=headless)
+
+    try:
+        commander.start()
+        commander.login()
+        commander.run_all_scenarios(scope=scope)
+        print("✅ 巡检任务执行成功")
+        exit_code = 0
+    except Exception as e:
+        print(f"❌ 巡检任务执行失败: {e}")
+        exit_code = 1
+    finally:
+        # ── 保存结构化报告 ─────────────────────────────────────────────────────
+        commander.save_results()
+
+        # 确认报告文件存在
+        result_file = Path(reports_dir) / f"{task_id}_results.json"
+        if result_file.exists():
+            print(f"📊 结构化报告已生成: {result_file}")
+        else:
+            # 兜底: 直接写入
+            try:
+                Path(reports_dir).mkdir(exist_ok=True, parents=True)
+                with open(result_file, "w", encoding="utf-8") as f:
+                    json.dump(commander.results, f, ensure_ascii=False, indent=2)
+                print(f"📊 结构化报告已生成 (兜底): {result_file}")
+            except Exception as e:
+                print(f"⚠️ 报告写入失败: {e}")
+
+        commander.stop()
+
+    sys.exit(exit_code)
+
+
+if __name__ == "__main__":
+    main()