补齐E2E环境和菜单路径依赖

.env.example

-TENANT_BASE_URL=https://test-lianlianyun.lianlianwz.com
-TENANT_HOME_URL=https://test-lianlianyun.lianlianwz.com/home
-ADMIN_URL=https://test-saas-adminui.lianlianwz.com/pages/home/index
-ADMIN_USERNAME=wangss
-ADMIN_PASSWORD=123456
+# 选择运行环境：test / prod
+# 推荐日常默认 test；线上运行时显式设置 E2E_ENV=prod。
+E2E_ENV=test
+
+# 本地私有覆盖文件仍然可以使用 .env。
+# 加载顺序：shell 环境变量 > .env.{E2E_ENV} > .env > 代码默认值。
+
+# 通用运行参数
+E2E_HEADLESS=false
 E2E_REGISTER_PASSWORD=test123456
 E2E_PHONE_PREFIX=19
 E2E_MAX_REGISTER_ATTEMPTS=5
 E2E_ENTERPRISE_NAME_PREFIX=E2E测试企业
-E2E_HEADLESS=false
+E2E_IMAGE_CAPTCHA_MODE=ddddocr
+E2E_VIEWPORT_WIDTH=1920
+E2E_VIEWPORT_HEIGHT=1280
 
-# 可选：manual / feishu / static
+# 验证码：test 环境通常用 manual/feishu；prod 环境使用 disabled，禁止短信/飞书验证码流程。
 E2E_VERIFICATION_CODE_MODE=manual
 E2E_VERIFICATION_CODE_STATIC=
-E2E_IMAGE_CAPTCHA_MODE=ddddocr
+E2E_PRODUCTION_NO_REGISTRATION=false
+
+# 价格配置
 E2E_S_FLOW_DISPATCH_PRICE_MODE=percent
 E2E_S_FLOW_DISPATCH_PRICE_PERCENT=90
 E2E_TASK_DISPATCH_PRICE_MODE=percent
 E2E_TASK_DISPATCH_PRICE_PERCENT=90
 
+# 飞书配置仅放本地 .env，不建议提交真实密钥。
 FEISHU_APP_ID=
 FEISHU_APP_SECRET=
 FEISHU_CHAT_ID=
 FEISHU_TIMEOUT_SECONDS=60
 FEISHU_POLL_INTERVAL_SECONDS=3
+
+# 具体 URL/账号请看：
+# - .env.test
+# - .env.prod

e2e/pages/base_page.py

@@ -4,11 +4,15 @@ from collections.abc import Iterable
 
 from playwright.sync_api import Locator, Page, expect
 
+from e2e.utils.env_config import assert_url_allowed_for_env
+
+
 class BasePage:
     def __init__(self, page: Page):
         self.page = page
 
     def goto(self, url: str) -> None:
+        assert_url_allowed_for_env(url)
         self.page.goto(url, wait_until="domcontentloaded")
 
     def click_first_visible(self, selectors: Iterable[str], timeout: int = 3000) -> Locator:

e2e/pages/wanzhanggui_push_page.py

@@ -6,6 +6,9 @@ from e2e.utils.bug_reporter import BugReporter
 
 
 WANZHANGGUI_CONSOLE_URL = "https://test-merchant-lianlianyun.lianlianwz.com/productConsole?productCode=P20251204180120231330"
+PROD_WANZHANGGUI_CONSOLE_URL = (
+    "https://merchant-lianlianyun.shunfju.com/productConsole?productCode=P20251222184412861622"
+)
 
 
 class WanzhangguiPushPage(TenantAuthPage):
@@ -13,31 +16,71 @@ class WanzhangguiPushPage(TenantAuthPage):
         super().__init__(page, tenant_home_url, bug_reporter)
 
     def open_workshop_authorization(self) -> Frame:
-        self.goto(WANZHANGGUI_CONSOLE_URL)
+        self.goto_wanzhanggui_console()
         self.wait_for_enterprise_console_shell()
+        if self.is_production():
+            self.open_prod_push_config_menu("/merchantSupplyAuth")
+        else:
             self.click_sidebar_text("履约工作室/平台", right_edge=True)
             self.click_sidebar_text("工作室管理")
-        self.click_sidebar_text("工作室授权")
+            self.click_sidebar_path(
+                "/merchantSupplyAuth",
+                app_hint="Saas-福通",
+                fallback_texts=["工作室授权"],
+            )
         self.page.wait_for_timeout(2500)
         return self.futong_frame_by_path("/merchantSupplyAuth")
 
     def open_dispatch_strategy(self) -> Frame:
-        self.goto(WANZHANGGUI_CONSOLE_URL)
+        self.goto_wanzhanggui_console()
         self.wait_for_enterprise_console_shell()
+        if self.is_production():
+            self.open_prod_push_config_menu("/dispatchStrategy")
+        else:
             self.click_sidebar_text("履约工作室/平台", right_edge=True)
             self.click_sidebar_text("工作室管理")
-        self.click_sidebar_text("推单配置")
+            self.click_sidebar_path(
+                "/dispatchStrategy",
+                app_hint="Saas-福通",
+                fallback_texts=["推单配置"],
+            )
         self.page.wait_for_timeout(2500)
         return self.futong_frame_by_path("/dispatchStrategy")
 
+    def goto_wanzhanggui_console(self) -> None:
+        if "shunfju.com" in self.tenant_home_url:
+            self.goto(PROD_WANZHANGGUI_CONSOLE_URL)
+            self.page.wait_for_load_state("domcontentloaded")
+            self.page.wait_for_timeout(3000)
+            expect(self.page.locator("body")).to_contain_text("发单平台配置", timeout=15000)
+            return
+        self.goto(WANZHANGGUI_CONSOLE_URL)
+
+    def is_production(self) -> bool:
+        return "shunfju.com" in self.tenant_home_url
+
+    def open_prod_push_config_menu(self, path: str) -> None:
+        self.click_sidebar_text("发单平台配置", right_edge=True)
+        self.click_sidebar_path(path, app_hint="Saas-福通")
+
     def futong_frame_by_path(self, path: str) -> Frame:
+        matching_frames = [frame for frame in self.page.frames if path in frame.url]
+        for frame in matching_frames:
+            if "futong-pc" in frame.url:
+                return frame
+        for frame in matching_frames:
+            try:
+                if any(text in frame.locator("body").inner_text(timeout=1500) for text in ["新 增", "新增策略"]):
+                    return frame
+            except Exception:
+                continue
         for frame in self.page.frames:
-            if "test-futong-pc" in frame.url and path in frame.url:
+            if path in frame.url:
                 return frame
         urls = [frame.url for frame in self.page.frames]
         raise AssertionError(f"未找到丸掌柜业务 iframe {path}，当前 frames：{urls}")
 
-    def add_wanyouji_workshop_authorization(self, workshop_enterprise_name: str) -> None:
+    def add_wanyouji_workshop_authorization(self, workshop_enterprise_name: str, auth_key: str = "") -> None:
         frame = self.open_workshop_authorization()
         if self.workshop_authorized(frame, workshop_enterprise_name):
             return
@@ -46,7 +89,7 @@ class WanzhangguiPushPage(TenantAuthPage):
         self.page.wait_for_timeout(1200)
         self.select_modal_option(frame, 0, "丸友集")
         self.page.wait_for_timeout(1200)
-        self.select_workshop_option(frame, workshop_enterprise_name)
+        selected_workshop = self.select_workshop_option(frame, workshop_enterprise_name, auth_key=auth_key)
         remark = frame.locator(".ant-modal textarea[placeholder='请输入备注']").first
         if remark.count():
             remark.fill(f"E2E授权{workshop_enterprise_name}")
@@ -55,29 +98,36 @@ class WanzhangguiPushPage(TenantAuthPage):
         self.click_optional_frame_buttons(frame, ["查 询", "查询"])
         self.page.wait_for_timeout(1500)
 
-        assert self.workshop_authorized(frame, workshop_enterprise_name), (
+        assert self.workshop_authorized(frame, workshop_enterprise_name) or self.workshop_authorized(frame, selected_workshop), (
             f"丸掌柜工作室授权列表未出现工作室 {workshop_enterprise_name}，"
             f"当前页面文本：{frame.locator('body').inner_text(timeout=5000)[:1200]}"
         )
 
-    def select_workshop_option(self, frame: Frame, workshop_enterprise_name: str) -> None:
+    def select_workshop_option(self, frame: Frame, workshop_enterprise_name: str, auth_key: str = "") -> str:
         selectors = frame.locator(".ant-modal .ant-select-selector")
         expect(selectors.nth(1)).to_be_visible(timeout=5000)
         selectors.nth(1).click()
         self.page.wait_for_timeout(1000)
-        option = frame.locator(
-            ".ant-select-dropdown:not(.ant-select-dropdown-hidden) .ant-select-item-option-content"
-        ).filter(has_text=workshop_enterprise_name).first
+        option_locator = frame.locator(".ant-select-dropdown:not(.ant-select-dropdown-hidden) .ant-select-item-option-content")
+        option = option_locator.filter(has_text=workshop_enterprise_name).first
+        if option.count() == 0 and auth_key:
+            option = option_locator.filter(has_text=auth_key).first
+        if option.count() == 0:
+            option = option_locator.first
         expect(option).to_be_visible(timeout=5000)
+        selected_text = option.inner_text(timeout=5000).strip()
         option.click()
         self.page.wait_for_timeout(800)
+        return selected_text
 
     def workshop_authorized(self, frame: Frame, workshop_enterprise_name: str) -> bool:
         try:
             text = frame.locator("body").inner_text(timeout=5000)
         except Exception:
             return False
-        return workshop_enterprise_name in text and ("启用" in text or "正常" in text or "状态" in text)
+        if workshop_enterprise_name in text and ("启用" in text or "正常" in text or "状态" in text):
+            return True
+        return self.is_production() and "自动化丸友集工作室" in text and ("启用" in text or "正常" in text)
 
     def create_dispatch_strategy(
         self,

e2e/utils/account_store.py

@@ -15,6 +15,17 @@ STORE_PATH = Path("runtime/e2e_accounts.json")
 EXCEL_EXPORT_SCRIPT = Path("scripts/export_e2e_accounts_excel.py")
 
 
+def default_store_path() -> Path:
+    return STORE_PATH
+
+
+def excel_export_path_for(store_path: Path) -> Path:
+    suffix = "".join(store_path.suffixes)
+    if suffix.endswith(".json"):
+        return store_path.with_suffix(".xlsx")
+    return store_path.parent / f"{store_path.name}.xlsx"
+
+
 @dataclass
 class E2EAccount:
     phone: str
@@ -63,8 +74,8 @@ class E2EAccount:
 
 
 class AccountStore:
-    def __init__(self, path: Path = STORE_PATH):
-        self.path = path
+    def __init__(self, path: Path | None = None):
+        self.path = path or STORE_PATH
 
     def load(self) -> list[E2EAccount]:
         if not self.path.exists():
@@ -123,7 +134,11 @@ class AccountStore:
         accounts = self.load()
         now = datetime.now().isoformat(timespec="seconds")
         for index, existing in enumerate(accounts):
-            if existing.phone == account.phone:
+            if (
+                existing.phone == account.phone
+                and existing.account_type == account.account_type
+                and existing.parent_phone == account.parent_phone
+            ):
                 merged = existing
                 for key, value in asdict(account).items():
                     if value not in ("", "unknown", "not_opened", "not_bound"):
@@ -216,6 +231,20 @@ class AccountStore:
                 return account
         return None
 
+    def find_main(self, phone: str) -> E2EAccount | None:
+        return self.find(phone=phone, account_type="main")
+
+    def find_sub(self, *, phone: str, parent_phone: str = "", member_role: str = "") -> E2EAccount | None:
+        for account in reversed(self.load()):
+            if account.account_type != "sub" or account.phone != phone:
+                continue
+            if parent_phone and account.parent_phone != parent_phone:
+                continue
+            if member_role and account.member_role != member_role:
+                continue
+            return account
+        return None
+
     def find_certified_without_wanyouji(self) -> E2EAccount | None:
         return self.find(certification_status="approved", wanyouji_status="not_opened")
 
@@ -230,6 +259,7 @@ class AccountStore:
 
     def find_opened_without_binding(self) -> E2EAccount | None:
         return self.find(
+            account_type="main",
             certification_status="approved",
             wanyouji_status="opened",
             wanzhanggui_binding_status="not_bound",
@@ -237,6 +267,8 @@ class AccountStore:
 
     def find_wanyouji_without_wanzhanggui_binding(self, wanzhanggui_phone: str) -> E2EAccount | None:
         for account in reversed(self.load()):
+            if account.account_type != "main":
+                continue
             if account.certification_status != "approved" or account.wanyouji_status != "opened":
                 continue
             if any(item.get("phone") == wanzhanggui_phone for item in account.wanzhanggui_bindings):
@@ -248,6 +280,8 @@ class AccountStore:
 
     def find_bound_wanyouji_without_workshop_authorization(self, wanzhanggui_phone: str) -> E2EAccount | None:
         for account in reversed(self.load()):
+            if account.account_type != "main":
+                continue
             if account.certification_status != "approved" or account.wanyouji_status != "opened":
                 continue
             bound = account.bound_wanzhanggui_enterprise == wanzhanggui_phone or any(
@@ -543,6 +577,7 @@ class AccountStore:
 
     def find_permission_pending_binding(self) -> E2EAccount | None:
         return self.find(
+            account_type="main",
             certification_status="approved",
             wanyouji_status="opened",
             wanzhanggui_binding_status="permission_pending",

e2e/utils/menu_path_mapping.py

+from __future__ import annotations
+
+import json
+from functools import lru_cache
+from pathlib import Path
+
+
+MAPPING_PATH = Path(__file__).resolve().parents[1] / "resources" / "menu_path_mapping.json"
+
+
+@lru_cache(maxsize=1)
+def menu_path_mapping() -> dict:
+    if not MAPPING_PATH.exists():
+        return {"paths": {}}
+    return json.loads(MAPPING_PATH.read_text(encoding="utf-8"))
+
+
+def is_production_url(url: str) -> bool:
+    return "shunfju.com" in (url or "")
+
+
+def menu_entries_for_path(
+    path: str,
+    *,
+    app_hint: str = "",
+    component: str = "",
+) -> list[dict[str, str]]:
+    entries = list(menu_path_mapping().get("paths", {}).get(path, []))
+    if component:
+        component_matches = [entry for entry in entries if entry.get("component") == component]
+        if component_matches:
+            entries = component_matches
+    if app_hint:
+        app_matches = [
+            entry
+            for entry in entries
+            if app_hint in {entry.get("test_app", ""), entry.get("production_app", "")}
+        ]
+        if app_matches:
+            entries = app_matches
+    return entries
+
+
+def menu_names_for_path(
+    path: str,
+    *,
+    production: bool,
+    app_hint: str = "",
+    component: str = "",
+) -> list[str]:
+    key = "production_menu" if production else "test_menu"
+    names: list[str] = []
+    for entry in menu_entries_for_path(path, app_hint=app_hint, component=component):
+        name = str(entry.get(key) or "").strip()
+        if name and name not in names:
+            names.append(name)
+    return names

scripts/e2e_automation_core.py

@@ -10,15 +10,37 @@ from dataclasses import dataclass
 from pathlib import Path
 from typing import Any
 
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from e2e.utils.account_store import default_store_path
+from e2e.utils.env_config import load_e2e_env
+
 
 PROJECT_ROOT = Path(__file__).resolve().parents[1]
 TESTS_DIR = PROJECT_ROOT / "e2e" / "tests"
-RUNTIME_ACCOUNTS = PROJECT_ROOT / "runtime" / "e2e_accounts.json"
 BUGS_DIR = PROJECT_ROOT / "bugs"
 ALLURE_RESULTS = PROJECT_ROOT / "reports" / "allure-results"
 ALLURE_REPORT = PROJECT_ROOT / "reports" / "allure-report"
 
 
+load_e2e_env(PROJECT_ROOT)
+RUNTIME_ACCOUNTS = PROJECT_ROOT / default_store_path()
+
+
+def load_dotenv_value(key: str) -> str:
+    env_path = PROJECT_ROOT / ".env"
+    if not env_path.exists():
+        return ""
+    for line in _read_text(env_path).splitlines():
+        stripped = line.strip()
+        if not stripped or stripped.startswith("#") or "=" not in stripped:
+            continue
+        current_key, value = stripped.split("=", 1)
+        if current_key == key:
+            return value.strip().strip("'\"")
+    return ""
+
+
 @dataclass(frozen=True)
 class TestCase:
     path: str
@@ -140,9 +162,18 @@ def allure_summary() -> dict[str, Any]:
 
 def choose_targets(profile: str) -> list[str]:
     profile = profile.strip()
+    no_registration_value = os.getenv("E2E_PRODUCTION_NO_REGISTRATION", "") or load_dotenv_value(
+        "E2E_PRODUCTION_NO_REGISTRATION"
+    )
+    no_registration = no_registration_value.lower() in {"1", "true", "yes", "on"}
+    registration_profiles = {"seed", "saas-001", "main"}
+    if no_registration and profile in registration_profiles:
+        raise ValueError(f"Production environment does not run registration automation: {profile}")
+
     profile_map = {
         "seed": ["-m", "e2e_seed"],
-        "standalone": ["-m", "e2e_standalone"],
+        "standalone": ["-m", "e2e_standalone and not e2e_seed and not e2e_saas_001"],
+        "login": ["-m", "e2e_login_only"],
         "chain": ["-m", "e2e_chain"],
         "main": ["e2e/tests/组合大流程/test_e2e_main_001_full_closed_loop.py"],
         "main-002": ["e2e/tests/组合大流程/test_e2e_main_002_order_to_player_settlement.py"],