Compare commits
9 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| d3fee3f46c | |||
| ebea77b5f2 | |||
| 430274dbf9 | |||
| 0157fde0f1 | |||
| aa9bf262fe | |||
| 52831b80e0 | |||
| 17a7a54349 | |||
| c21339ae44 | |||
| 9878f570d7 |
74
CHANGELOG.md
74
CHANGELOG.md
@@ -2,6 +2,80 @@
|
||||
|
||||
本文件对应技能市场 **「更新日志」** Tab。
|
||||
|
||||
## 1.1.28
|
||||
|
||||
- **彻底修复「SPA 二级登录页卡顿」**:
|
||||
- v1.1.27 失败根因:URL 已稳定在 `/login` 时,React SPA input 可能还没渲染(hydrate 慢),`_fill_demo_on_spa_login` 找不到 input → False → 卡 180 秒等用户输入
|
||||
- 修复:步骤 4.5 先 `wait_for_selector(input, timeout=10s, state="attached")` 等 input 真正 hydrate + 0.5s React 稳定,再 fill demo/123456 + 点登录
|
||||
- demo fill 后如果 URL 仍卡在 login 页(demo 失效)→ headed 模式 fallback 等用户手工登录,headless 模式抛 `LoginTimeoutError`
|
||||
- 不再依赖「input 是否预填」—— sandbox SPA 二级登录页 input 默认不预填,必须强制覆盖 fill
|
||||
- 合计 66/66
|
||||
|
||||
## 1.1.27
|
||||
|
||||
- **修复「到 SPA 二级登录页卡 30 秒」问题**:
|
||||
- 之前步骤 4 等 URL 稳定到 `/app/` 或 `/select` 30 秒,但 sandbox 登录后会把人踢回 `/industries/campus/smartcampus/login`(URL 含 `/login`),等满 30 秒才发现要进 SPA 登录兜底
|
||||
- 步骤 4 改成等 URL 跳到 `/app/` 或 `/select` 或 `/login` 或 `/signin`(任一即可,**最多 30 秒**),URL 含 `/login` 立即退出等待
|
||||
- **新增 `_fill_demo_on_spa_login(page)`**:SPA 二级登录页 input 默认不预填(与根域不同),主动 `fill('demo')` + `fill('123456')` + 点登录按钮(Sign in / 登录 / Login / submit 任一)
|
||||
- 步骤 4.5 改为:URL 落在 SPA 登录页 → 自动 `_fill_demo_on_spa_login` → 不再卡 180 秒等用户输入
|
||||
- 新增 3 个测试覆盖 `_fill_demo_on_spa_login`(basic / 中文按钮 / 无 input),合计 66/66
|
||||
|
||||
## 1.1.26
|
||||
|
||||
- **按用户最新要求改回「自动点登录」行为**:「到了 /industries/campus/smartcampus/login 整个页面,有账号和密码可用直接点登录跳转」
|
||||
- `_wait_for_login_inputs_filled` 改回 v1.1.21 行为:input 都有值 → 自动点 Sign in / 登录按钮 → return True
|
||||
- 与 `_try_auto_sign_in` 行为统一:登录页 = 「有数据自动点」/ 「无数据等用户输入」
|
||||
- 测试替换:`wait_login_inputs_filled_no_auto_click_v1_24` → `wait_login_inputs_filled_auto_click_v1_26`,合计 63/63
|
||||
|
||||
## 1.1.25
|
||||
|
||||
- **按用户要求修正登录页等待逻辑**:
|
||||
- 有数据(账号密码已存在)→ 自动点登录按钮(`_try_auto_sign_in`)
|
||||
- **无数据 → 等待用户输入完成(最多 180 秒)→ 等 URL 跳离登录页**(按钮交给用户自己点,脚本不抢按钮)
|
||||
- `_wait_for_login_inputs_filled` 重写为「**等待模式**」:不再自动点 Sign in / 登录按钮;只轮询 input 值 + URL,每 15 秒提示「请点登录按钮」或「等待输入」
|
||||
- 根域登录 + SPA 二级登录页统一使用同一套行为
|
||||
- 新增测试 `wait_login_inputs_filled_no_auto_click_v1_24` 验证不抢按钮,合计 63/63
|
||||
|
||||
## 1.1.24
|
||||
|
||||
- **修复 `/industries/campus/smartcampus/login` 等待过久问题**:sandbox 在用户首次登录后可能把人踢回 SPA 二级登录页(按钮中文「登录」),之前根域自动登录代码不识别这个 URL + 中文按钮,导致脚本卡死等用户输入
|
||||
- `_try_auto_sign_in` 扩展为通用版本:支持 `#jc-user` / `input[type='tel']` / `input[autocomplete='username']` 多种 user input selector;按钮文本按优先级匹配 `Sign in` > `登录` > `Login` > `signin` > `submit`
|
||||
- `collect_all` 登录后新增步骤 4.5:等 URL 稳定到 `/app/` 或 `/select` 后,若发现 URL 仍含 `/login` / `/signin`(落在 SPA 二级登录页),自动重走一次 `_try_auto_sign_in` → 再等 URL 跳到 `/app/` 或 `/select`
|
||||
- 新增 4 个测试覆盖 `_try_auto_sign_in`(SPA 中文按钮 / 根域英文按钮 / input 空 / selector 超时),合计 58/58
|
||||
|
||||
## 1.1.23
|
||||
|
||||
- **修复登录后跳 /app/home 的流程**:
|
||||
- `_wait_for_url_stable` 支持 tuple targets(任一匹配即稳定)
|
||||
- 登录后等 URL 稳定到 `/app/` 或 `/select`(sandbox 中间页)
|
||||
- 显式 `goto SPA_HOME_URL` (/app/home) → 等 sidebar link 出现 → click `<a href="/industries/campus/smartcampus/app/hr/contracts">合同到期续聘</a>`
|
||||
- 删掉重复的 # 4 块(之前手抖复制粘贴的 bug)
|
||||
|
||||
## 1.1.22
|
||||
|
||||
- **登录流程接入 sandbox 自动登录**:root 登录页(https://sandbox.jc2009.com/)的 #jc-user / #jc-pass 默认有 demo/123456 预填
|
||||
- 新增 `_try_auto_sign_in(page, wait_input_sec=2)`:等 input 出现 2 秒 → 检查是否有值 → 有则自动点 Sign in
|
||||
- 新增 `_wait_for_login_inputs_filled(page, wait_sec)`:input 为空时等用户输入 → 两个 input 都有值后自动点 Sign in
|
||||
- 已登录态(profile cookie 复用)直接跳过
|
||||
- 输入超时 fallback 到 `_wait_for_manual_login`(多信号判定)
|
||||
|
||||
## 1.1.21
|
||||
|
||||
- **修复登录完成检测盲区**:`_wait_for_manual_login` 现在监听 3 个信号判断「登录已完成」,任一触发即继续
|
||||
1. URL 含 `/app/` 且稳定 2 秒(v1.1.20 旧逻辑)
|
||||
2. URL 已离开 `/login` / `/signin` / `/auth/` 等登录路径
|
||||
3. 页面出现 sidebar 元素 `a[href$='/hr/contracts']`
|
||||
- 解决「用户已经登录但脚本还在傻等」的回归:之前只判 URL 变化,cookie 已生效复用 profile 场景或 SPA 不跳 URL 的场景会卡 180 秒
|
||||
|
||||
## 1.1.20
|
||||
|
||||
- **登录流程重写**:第一跳改为 `https://sandbox.jc2009.com/`(根域,用户最初看到的页面),**不做任何操作**等 URL 稳定
|
||||
- 已登录态:根域自动跳到 `/app/home` → 直接走侧边栏到 `hr/contracts` 采集
|
||||
- 未登录态:**纯等用户在浏览器手工输入账号密码**,不再自动 fill demo/123456,不再 retry 强制 reload 打断用户输入
|
||||
- 新增 `_wait_for_url_stable(page, target_substr, max_wait_sec)`:等 URL 真正稳定 2 秒后才返回,避免 sandbox 中间页误导
|
||||
- `_wait_for_manual_login` 重写:每 5 秒轮询 URL(之前 2 秒太密),每 15 秒打印一次剩余时间提示,**等待期间不触发任何 goto / reload / click**
|
||||
- 新增 4 个 login flow 测试(`wait_url_stable_immediate` / `wait_url_stable_transition` / `wait_url_stable_timeout` / `wait_manual_login_no_goto`),合计 58/58
|
||||
|
||||
## 1.1.19
|
||||
|
||||
- **修复 1.1.18 引入的采集回归**:把 `_persist_contracts` 里的入队调用移到独立事务 `_enqueue_download_tasks_safe`,FK 失败不再回滚 `contracts` / `materials_content` 落库;返回 JSON 新增 `download_queue` 字段方便 UI 直接看到入队结果
|
||||
|
||||
2
SKILL.md
2
SKILL.md
@@ -1,7 +1,7 @@
|
||||
---
|
||||
name: 合同到期智能预警与续聘辅助
|
||||
description: "当用户说要扫描即将到期的合同、查 90/60/30 天预警名单、给某位教师生成续聘任务书、下发单位版续聘清单、从 SmartCampus HR 平台抓合同数据、解析续聘材料 PDF/Word、查看本月合同到期人员清单时,加载本技能。支持定时扫描(每日凌晨)+ 数据管理 + Agent 三入口,覆盖合同到期智能预警与续聘辅助全流程。"
|
||||
version: 1.1.19
|
||||
version: 1.1.28
|
||||
author: 深圳匠厂科技有限公司
|
||||
metadata:
|
||||
openclaw:
|
||||
|
||||
@@ -36,6 +36,7 @@ from util.constants import (
|
||||
FULL_RENDER_WAIT_MS,
|
||||
GOTO_TIMEOUT_MS,
|
||||
HUMAN_POLL_INTERVAL,
|
||||
HUMAN_PROMPT_INTERVAL,
|
||||
HUMAN_WAIT_TIMEOUT,
|
||||
POST_LOGIN_SETTLE_MS,
|
||||
POST_RELOAD_LOGIN_HYDRATION_MS,
|
||||
@@ -55,6 +56,8 @@ from util.constants import (
|
||||
TABLE_POLL_INTERVAL_MS,
|
||||
TABLE_POLL_MAX_ATTEMPTS,
|
||||
TABLE_TR_TIMEOUT_MS,
|
||||
URL_STABLE_POLL_S,
|
||||
URL_STABLE_REQUIRED_S,
|
||||
URL_AFTER_LOGIN_TIMEOUT_MS,
|
||||
)
|
||||
from util.logging_setup import get_skill_logger
|
||||
@@ -164,33 +167,348 @@ def _is_logged_in(page) -> bool:
|
||||
)
|
||||
|
||||
|
||||
def _is_spa_routed(page) -> bool:
|
||||
"""v1.1.22:检测 sandbox SPA 是否已路由到 SPA 页面(含 /select /app/)。"""
|
||||
url_lower = page.url.lower()
|
||||
return (
|
||||
("/app/" in url_lower or "/select" in url_lower)
|
||||
and "/login" not in url_lower
|
||||
and "/signin" not in url_lower
|
||||
)
|
||||
|
||||
|
||||
def _try_auto_sign_in(page, wait_input_sec: int = 2) -> bool:
|
||||
"""v1.1.21:检测登录页 demo/123456 是否已预填,有则自动点 Sign in。
|
||||
|
||||
v1.1.24:扩展到支持 SPA 二级登录页(/industries/campus/smartcampus/login),
|
||||
按钮文本可能是中文「登录」而非英文「Sign in」。
|
||||
|
||||
适用场景:
|
||||
- 根域登录页(https://sandbox.jc2009.com/):按钮「Sign in」
|
||||
- SPA 二级登录页(/industries/campus/smartcampus/login):按钮「登录」
|
||||
默认有 demo/123456(或 18677308191 等真实账号)预填在 input。
|
||||
|
||||
行为:
|
||||
1. 等 input 出现,最长 wait_input_sec 秒(默认 2s)
|
||||
2. 检查 input 是否都有值
|
||||
3. 有值 → 点 Sign in / 登录 → return True
|
||||
4. 无值 → return False(交给 _wait_for_login_inputs_filled 接管)
|
||||
|
||||
不会自动 fill(如果没值就放弃,让用户自己输)。
|
||||
"""
|
||||
try:
|
||||
page.wait_for_selector(
|
||||
"#jc-user, #jc-pass, input[type='tel'], input[type='password']",
|
||||
timeout=wait_input_sec * 1000,
|
||||
)
|
||||
except Exception:
|
||||
logger.info(
|
||||
"platform_collector: %ds 内未出现登录 input,跳过自动登录",
|
||||
wait_input_sec,
|
||||
)
|
||||
return False
|
||||
|
||||
# 多种 input selector 优先级
|
||||
user_input = None
|
||||
for sel in ("#jc-user", "input[type='tel']", "input[autocomplete='username']"):
|
||||
loc = page.locator(sel).first
|
||||
if loc.count() > 0:
|
||||
user_input = loc
|
||||
break
|
||||
if user_input is None:
|
||||
return False
|
||||
|
||||
pass_input = None
|
||||
for sel in ("#jc-pass", "input[type='password']", "input[autocomplete='current-password']"):
|
||||
loc = page.locator(sel).first
|
||||
if loc.count() > 0:
|
||||
pass_input = loc
|
||||
break
|
||||
if pass_input is None:
|
||||
return False
|
||||
|
||||
user_val = (user_input.input_value() or "").strip()
|
||||
pass_val = (pass_input.input_value() or "").strip()
|
||||
|
||||
if not user_val or not pass_val:
|
||||
logger.info(
|
||||
"platform_collector: 登录页 input 为空(user=%r, pass=%r),"
|
||||
"等用户输入",
|
||||
bool(user_val), bool(pass_val),
|
||||
)
|
||||
return False
|
||||
|
||||
# 多种按钮文本优先级:英文 Sign in > 中文登录 > submit
|
||||
btn = None
|
||||
btn_text_used = ""
|
||||
for txt in ("Sign in", "登录", "Login", "signin"):
|
||||
loc = page.locator(f"button:has-text('{txt}')").first
|
||||
if loc.count() > 0:
|
||||
btn = loc
|
||||
btn_text_used = txt
|
||||
break
|
||||
if btn is None:
|
||||
btn = page.locator("button[type='submit']").first
|
||||
btn_text_used = "submit"
|
||||
if btn is None or btn.count() == 0:
|
||||
logger.warning("platform_collector: 未找到登录按钮")
|
||||
return False
|
||||
|
||||
logger.info(
|
||||
"platform_collector: 登录页已预填账号(user=%r),自动点 %s 按钮",
|
||||
user_val, btn_text_used,
|
||||
)
|
||||
btn.click()
|
||||
return True
|
||||
|
||||
|
||||
def _wait_for_login_inputs_filled(page, wait_sec: int = HUMAN_WAIT_TIMEOUT) -> bool:
|
||||
"""v1.1.21:等用户输入完账号密码,然后自动点登录按钮。
|
||||
|
||||
v1.1.26:按用户最新要求改回**自动点登录**行为——
|
||||
「到了 /industries/campus/smartcampus/login 整个页面,有账号和密码可用直接点登录跳转」,
|
||||
不再等用户自己点。
|
||||
|
||||
优先级(任一 selector 命中即视为有 input):
|
||||
- user: `#jc-user` / `input[type='tel']` / `input[autocomplete='username']`
|
||||
- pass: `#jc-pass` / `input[type='password']` / `input[autocomplete='current-password']`
|
||||
|
||||
按钮文本优先级:`Sign in` / `登录` / `Login` / `signin` / `submit`。
|
||||
"""
|
||||
deadline = time.monotonic() + wait_sec
|
||||
last_prompt_at = time.monotonic()
|
||||
last_state_log = time.monotonic()
|
||||
|
||||
while time.monotonic() < deadline:
|
||||
cur_url = page.url
|
||||
|
||||
# 已不在登录页(用户可能已点登录完成跳转)
|
||||
if not _is_login_page_url_for_user(cur_url):
|
||||
logger.info(
|
||||
"platform_collector: URL 已离开登录页(%s),登录完成", cur_url,
|
||||
)
|
||||
return True
|
||||
|
||||
# 探测 input(多 selector 优先级)
|
||||
user_val, pass_val = "", ""
|
||||
user_loc, pass_loc = None, None
|
||||
for sel in ("#jc-user", "input[type='tel']", "input[autocomplete='username']"):
|
||||
loc = page.locator(sel).first
|
||||
if loc.count() > 0:
|
||||
user_loc = loc
|
||||
user_val = (loc.input_value() or "").strip()
|
||||
break
|
||||
for sel in ("#jc-pass", "input[type='password']", "input[autocomplete='current-password']"):
|
||||
loc = page.locator(sel).first
|
||||
if loc.count() > 0:
|
||||
pass_loc = loc
|
||||
pass_val = (loc.input_value() or "").strip()
|
||||
break
|
||||
|
||||
now = time.monotonic()
|
||||
if now - last_state_log >= HUMAN_PROMPT_INTERVAL:
|
||||
last_state_log = now
|
||||
logger.info(
|
||||
"platform_collector: 等待用户输入 user=%r pass=%r (URL=%s)",
|
||||
bool(user_val), bool(pass_val), cur_url,
|
||||
)
|
||||
|
||||
# input 都有值 → 自动点登录按钮(v1.1.26 关键:用户最新要求自动点)
|
||||
if user_val and pass_val:
|
||||
btn = None
|
||||
btn_text_used = ""
|
||||
for txt in ("Sign in", "登录", "Login", "signin"):
|
||||
loc = page.locator(f"button:has-text('{txt}')").first
|
||||
if loc.count() > 0:
|
||||
btn = loc
|
||||
btn_text_used = txt
|
||||
break
|
||||
if btn is None:
|
||||
btn = page.locator("button[type='submit']").first
|
||||
btn_text_used = "submit"
|
||||
if btn is not None and btn.count() > 0:
|
||||
logger.info(
|
||||
"platform_collector: ✅ 账号密码已填好(user=%r),自动点 %s 按钮",
|
||||
user_val, btn_text_used,
|
||||
)
|
||||
btn.click()
|
||||
return True
|
||||
logger.warning("platform_collector: input 有值但找不到登录按钮")
|
||||
|
||||
# 提醒用户
|
||||
if now - last_prompt_at >= HUMAN_PROMPT_INTERVAL:
|
||||
last_prompt_at = now
|
||||
remaining = int(deadline - now)
|
||||
if user_val and pass_val:
|
||||
# 上面 click 应该已经 return,不会到这里
|
||||
print(
|
||||
f"[登录] ✅ 账号密码已填好(user={user_val[:3]}***),"
|
||||
f"请点「登录」按钮,剩余 {remaining}s..."
|
||||
)
|
||||
else:
|
||||
print(
|
||||
f"[登录] 等待用户输入账号密码... 剩余 {remaining}s"
|
||||
)
|
||||
|
||||
time.sleep(HUMAN_POLL_INTERVAL)
|
||||
|
||||
logger.warning(
|
||||
"platform_collector: %ds 内用户未跳转(最后 URL=%s)", wait_sec, page.url,
|
||||
)
|
||||
return False
|
||||
|
||||
|
||||
def _is_login_page_url_for_user(url: str) -> bool:
|
||||
"""判断 URL 是否还在登录页(用于 _wait_for_login_inputs_filled 退出条件)。"""
|
||||
u = url.lower()
|
||||
if "/login" in u or "/signin" in u or "/auth/" in u:
|
||||
return True
|
||||
# 根域本身也是登录页(demo 数据未提交前的初始态)
|
||||
return u.rstrip("/").endswith("sandbox.jc2009.com")
|
||||
|
||||
|
||||
def _wait_for_manual_login(page, wait_sec: int = HUMAN_WAIT_TIMEOUT) -> bool:
|
||||
"""手工登录 fallback:轮询 _is_logged_in,每 HUMAN_POLL_INTERVAL 秒一次。
|
||||
"""手工登录 fallback(v1.1.19 重写,v1.1.20 加强,v1.1.21 修订):
|
||||
等用户在浏览器完成登录。
|
||||
|
||||
与 skill-template task_rpa.py::_ensure_logged_in 设计对齐:
|
||||
- 不自动破解,等用户在浏览器手工完成登录
|
||||
- 屏幕打印 [登录] 提示
|
||||
- 超时抛 LoginTimeoutError(ERROR:LOGIN_TIMEOUT)
|
||||
判定「登录完成」的多个信号(满足任一即视为完成):
|
||||
1. URL 含 `/app/` 且稳定 URL_STABLE_REQUIRED_S 秒(v1.1.19 旧逻辑)
|
||||
2. URL 已离开 `/login` / `/signin` / `/auth/` 等登录路径(v1.1.20 新增)
|
||||
—— 即便 sandbox 未跳到 /app/,只要离开登录页就认为登录成功
|
||||
3. 页面出现 sidebar 元素 `a[href$='/hr/contracts']`(v1.1.20 新增)
|
||||
—— 说明已加载到 SPA 主页
|
||||
|
||||
自动 fill demo/123456 失败时调用此函数。
|
||||
- **不做任何 goto / reload / click**,避免打断用户输入
|
||||
- 每 HUMAN_PROMPT_INTERVAL 秒打印一次剩余时间(默认 15s)
|
||||
- 每 HUMAN_POLL_INTERVAL 秒检查一次(默认 5s)
|
||||
- 超时抛 LoginTimeoutError。
|
||||
"""
|
||||
print(
|
||||
f"[登录] 自动登录失败,请在浏览器中完成登录"
|
||||
f"[登录] 请在浏览器中完成登录"
|
||||
f"(sandbox demo/123456 或真实账号),最多等待 {wait_sec} 秒..."
|
||||
)
|
||||
deadline = time.monotonic() + wait_sec
|
||||
attempt = 0
|
||||
while time.monotonic() < deadline:
|
||||
attempt += 1
|
||||
last_prompt_at = time.monotonic()
|
||||
last_url = page.url
|
||||
last_url_change_at = time.monotonic()
|
||||
|
||||
def _is_login_page_url(url: str) -> bool:
|
||||
u = url.lower()
|
||||
return (
|
||||
"/login" in u or "/signin" in u or "/auth/" in u
|
||||
or u.rstrip("/").endswith("sandbox.jc2009.com")
|
||||
)
|
||||
|
||||
def _sidebar_present() -> bool:
|
||||
try:
|
||||
return page.locator("a[href$='/hr/contracts']").count() > 0
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
while True:
|
||||
remaining = deadline - time.monotonic()
|
||||
if remaining <= 0:
|
||||
break
|
||||
|
||||
cur_url = page.url
|
||||
if cur_url != last_url:
|
||||
logger.info(
|
||||
"platform_collector: URL 变化 %s → %s", last_url, cur_url,
|
||||
)
|
||||
last_url = cur_url
|
||||
last_url_change_at = time.monotonic()
|
||||
|
||||
# 信号 1:URL 跳到 /app/... 并稳定
|
||||
if _is_logged_in(page):
|
||||
print(f"[登录] 检测到登录成功(attempt {attempt})")
|
||||
stable_for = time.monotonic() - last_url_change_at
|
||||
if stable_for >= URL_STABLE_REQUIRED_S:
|
||||
print(
|
||||
f"[登录] ✅ 检测到登录完成(URL 含 /app/,稳定 {stable_for:.1f}s)"
|
||||
)
|
||||
return True
|
||||
logger.info(
|
||||
"platform_collector: URL 含 /app/ 但还在过渡,等稳定 "
|
||||
"(current=%s, stable_for=%.1fs)", cur_url, stable_for,
|
||||
)
|
||||
# 信号 2:URL 已离开登录页(即使不在 /app/,比如卡在中间过渡页也算)
|
||||
elif not _is_login_page_url(cur_url):
|
||||
stable_for = time.monotonic() - last_url_change_at
|
||||
if stable_for >= URL_STABLE_REQUIRED_S:
|
||||
print(
|
||||
f"[登录] ✅ 检测到登录完成(URL 已离开登录页 → {cur_url},"
|
||||
f"稳定 {stable_for:.1f}s)"
|
||||
)
|
||||
return True
|
||||
logger.info(
|
||||
"platform_collector: URL 已离开登录页但还没稳定,等 "
|
||||
"(current=%s, stable_for=%.1fs)", cur_url, stable_for,
|
||||
)
|
||||
# 信号 3:sidebar 已出现
|
||||
elif _sidebar_present():
|
||||
print(
|
||||
f"[登录] ✅ 检测到登录完成(sidebar 已出现,URL={cur_url})"
|
||||
)
|
||||
return True
|
||||
else:
|
||||
# 还在登录页:每 HUMAN_PROMPT_INTERVAL 秒提醒用户一次
|
||||
now = time.monotonic()
|
||||
if now - last_prompt_at >= HUMAN_PROMPT_INTERVAL:
|
||||
last_prompt_at = now
|
||||
print(
|
||||
f"[登录] 等待用户输入... 剩余 {int(remaining)}s "
|
||||
f"(当前 URL={cur_url})"
|
||||
)
|
||||
|
||||
time.sleep(HUMAN_POLL_INTERVAL)
|
||||
|
||||
raise LoginTimeoutError(
|
||||
f"自动登录失败,{wait_sec}s 内用户在浏览器中也未完成登录"
|
||||
f"(最后 URL={page.url})"
|
||||
)
|
||||
|
||||
|
||||
def _wait_for_url_stable(
|
||||
page, target_substr: str | tuple[str, ...], max_wait_sec: int = 30,
|
||||
) -> bool:
|
||||
"""v1.1.19:等 URL 含 target_substr,且连续 URL_STABLE_REQUIRED_S 秒不变。
|
||||
|
||||
v1.1.22:target_substr 支持 tuple(任一匹配即视为稳定)。
|
||||
|
||||
用于「已登录态直接跳到目标页面」后等待 SPA 路由稳定。
|
||||
**不做任何 goto / reload**,只 listen URL 变化。
|
||||
|
||||
Returns:
|
||||
True = URL 已稳定到任一目标;False = 超时。
|
||||
"""
|
||||
if isinstance(target_substr, str):
|
||||
targets = (target_substr,)
|
||||
else:
|
||||
targets = target_substr
|
||||
|
||||
deadline = time.monotonic() + max_wait_sec
|
||||
last_url = page.url
|
||||
last_url_change_at = time.monotonic()
|
||||
|
||||
while time.monotonic() < deadline:
|
||||
cur_url = page.url
|
||||
if cur_url != last_url:
|
||||
logger.info(
|
||||
"platform_collector._wait_for_url_stable: URL 变化 %s → %s",
|
||||
last_url, cur_url,
|
||||
)
|
||||
last_url = cur_url
|
||||
last_url_change_at = time.monotonic()
|
||||
|
||||
cur_lower = cur_url.lower()
|
||||
if any(t.lower() in cur_lower for t in targets):
|
||||
stable_for = time.monotonic() - last_url_change_at
|
||||
if stable_for >= URL_STABLE_REQUIRED_S:
|
||||
return True
|
||||
|
||||
time.sleep(URL_STABLE_POLL_S)
|
||||
return False
|
||||
|
||||
|
||||
def _fill_root_login(page) -> bool:
|
||||
"""根域登录页(如还在):fill demo/123456 → click Sign in。返回是否执行了登录。"""
|
||||
user_input = page.locator('#jc-user').first
|
||||
@@ -211,6 +529,76 @@ def _fill_root_login(page) -> bool:
|
||||
return True
|
||||
|
||||
|
||||
def _fill_demo_on_spa_login(page) -> bool:
|
||||
"""v1.1.27:SPA 二级登录页(/industries/campus/smartcampus/login)
|
||||
input 默认**不预填**(与根域不同),必须主动 fill demo/123456 再点登录。
|
||||
|
||||
selector 优先级:
|
||||
- user: `#jc-user` → `input[type='tel']` → `input[autocomplete='username']`
|
||||
- pass: `#jc-pass` → `input[type='password']` → `input[autocomplete='current-password']`
|
||||
按钮文本优先级:Sign in → 登录 → Login → signin → submit[type]。
|
||||
|
||||
返回是否成功点登录按钮。
|
||||
"""
|
||||
# 找 user input
|
||||
user_loc = None
|
||||
for sel in ("#jc-user", "input[type='tel']", "input[autocomplete='username']"):
|
||||
loc = page.locator(sel).first
|
||||
if loc.count() > 0:
|
||||
user_loc = loc
|
||||
break
|
||||
if user_loc is None:
|
||||
logger.warning(
|
||||
"platform_collector._fill_demo_on_spa_login: 找不到 user input "
|
||||
"(#jc-user / tel / autocomplete=username 全 miss)"
|
||||
)
|
||||
return False
|
||||
|
||||
# 找 pass input
|
||||
pass_loc = None
|
||||
for sel in ("#jc-pass", "input[type='password']", "input[autocomplete='current-password']"):
|
||||
loc = page.locator(sel).first
|
||||
if loc.count() > 0:
|
||||
pass_loc = loc
|
||||
break
|
||||
if pass_loc is None:
|
||||
logger.warning(
|
||||
"platform_collector._fill_demo_on_spa_login: 找不到 pass input"
|
||||
)
|
||||
return False
|
||||
|
||||
user_loc.fill("demo")
|
||||
pass_loc.fill("123456")
|
||||
logger.info(
|
||||
"platform_collector: SPA 二级登录页 fill demo/123456 完成"
|
||||
)
|
||||
|
||||
# 找登录按钮
|
||||
btn = None
|
||||
btn_text_used = ""
|
||||
for txt in ("Sign in", "登录", "Login", "signin"):
|
||||
loc = page.locator(f"button:has-text('{txt}')").first
|
||||
if loc.count() > 0:
|
||||
btn = loc
|
||||
btn_text_used = txt
|
||||
break
|
||||
if btn is None:
|
||||
btn = page.locator("button[type='submit']").first
|
||||
btn_text_used = "submit"
|
||||
if btn is None or btn.count() == 0:
|
||||
logger.warning(
|
||||
"platform_collector._fill_demo_on_spa_login: 找不到登录按钮"
|
||||
)
|
||||
return False
|
||||
|
||||
logger.info(
|
||||
"platform_collector: SPA 二级登录页点 %s 按钮(demo/123456)",
|
||||
btn_text_used,
|
||||
)
|
||||
btn.click()
|
||||
return True
|
||||
|
||||
|
||||
def _click_campus_link(page) -> bool:
|
||||
"""在 /select 页面点智慧校园 tile,返回是否点中。"""
|
||||
link = page.locator("a[href*='/industries/campus']").first
|
||||
@@ -384,90 +772,151 @@ def collect_all(
|
||||
# 所有 wait_for_selector 默认 30s,调试时不直观)
|
||||
page.set_default_timeout(DEFAULT_PAGE_TIMEOUT_MS)
|
||||
|
||||
# 1. v1.1.11:先用**最小侵入**方式检测已登录态 —— 直接 goto SPA_HOME_URL
|
||||
# (不要先跳 ROOT_URL + 自动 fill,会破坏已登录态)
|
||||
# v1.1.11 改进:探针结果 + 二次等待 + cookie 检测,三重判断避免误判
|
||||
# 1. v1.1.21:先 goto **根域**(sandbox 登录页),让 SPA hydrate 2 秒
|
||||
logger.info(
|
||||
"platform_collector: 检测登录态,goto SPA_HOME 先探 (当前 URL=%s)",
|
||||
page.url,
|
||||
"platform_collector: 启动,goto ROOT_URL=%s (当前 URL=%s)",
|
||||
ROOT_URL, page.url,
|
||||
)
|
||||
_goto(page, SPA_HOME_URL)
|
||||
# v1.1.11:等 sandbox SPA 异步验证 cookie(POST_LOGIN_SETTLE_MS → SPA_HYDRATION_MS)
|
||||
page.wait_for_timeout(SPA_HYDRATION_MS)
|
||||
# v1.1.11:二次等待 —— sandbox SPA 有时首次访问会先跳一个中间过渡页
|
||||
# (如 /select),等几秒看 URL 是否稳定到 /app/
|
||||
first_url = page.url
|
||||
page.wait_for_timeout(SPA_HOME_RETRY_DELAY_MS)
|
||||
second_url = page.url
|
||||
if first_url != second_url:
|
||||
logger.info(
|
||||
"platform_collector: 探针 URL 过渡 %s → %s,等稳定", first_url, second_url,
|
||||
)
|
||||
page.wait_for_timeout(SPA_HOME_RETRY_DELAY_MS)
|
||||
_goto(page, ROOT_URL)
|
||||
# 给 React/Vue 框架 2 秒 hydrate 时间,看登录 input 是否就绪
|
||||
page.wait_for_timeout(2_000)
|
||||
|
||||
if _is_logged_in(page):
|
||||
logger.info(
|
||||
"platform_collector: ✅ 已登录态,**完全跳过登录流程**(不弹窗、不自动 fill、不打扰)"
|
||||
",URL=%s", page.url,
|
||||
)
|
||||
else:
|
||||
logger.info(
|
||||
"platform_collector: 未登录态(SPA_HOME 跳回 login, URL=%s),走登录分支",
|
||||
"platform_collector: ✅ 已登录态(URL=%s),跳过登录流程",
|
||||
page.url,
|
||||
)
|
||||
# 2. 未登录:直接 goto sandbox SPA login(绕过 /select 和 root 跳转)
|
||||
_goto(page, SPA_LOGIN_URL)
|
||||
# SPA hydration + framework 初始化时间(修复登录时间太短的核心点)
|
||||
page.wait_for_timeout(SPA_HYDRATION_MS)
|
||||
else:
|
||||
# 2. 等 #jc-user / #jc-pass 出现(最多 2 秒)
|
||||
# 若 sandbox 已预填 demo/123456 → 自动点 Sign in
|
||||
# 若 input 为空 → 等用户输入(**v1.1.24 用户自己点登录按钮**)
|
||||
auto_clicked = _try_auto_sign_in(page, wait_input_sec=2)
|
||||
if auto_clicked:
|
||||
logger.info(
|
||||
"platform_collector: ✅ 自动登录成功(预填 demo),等 URL 跳到 /app/"
|
||||
)
|
||||
elif not headless:
|
||||
# 3. input 为空:等用户输入 → **用户自己点 Sign in**(脚本不抢按钮)
|
||||
if _wait_for_login_inputs_filled(page, wait_sec=human_wait_timeout):
|
||||
logger.info(
|
||||
"platform_collector: ✅ 用户完成输入并跳转,等 URL 跳到 /app/"
|
||||
)
|
||||
else:
|
||||
# 超时未跳转 → fallback 到原来的多信号等待
|
||||
logger.warning(
|
||||
"platform_collector: 用户未跳转,fallback 到 _wait_for_manual_login"
|
||||
)
|
||||
_wait_for_manual_login(page, wait_sec=human_wait_timeout)
|
||||
else:
|
||||
raise LoginTimeoutError(
|
||||
f"headless 模式下未登录,且无法手工登录"
|
||||
f"(请用 --show-browser 重试)"
|
||||
)
|
||||
|
||||
# 3. sandbox SPA 自动登录(demo/123456 已知可用)
|
||||
# 失败 fallback 手工登录等待(轮询 _is_logged_in + 180s 兜底)
|
||||
if not _fill_spa_login(page):
|
||||
# 4. v1.1.27:等 URL 跳到任一「SPA 已路由」状态(最多 30s,**不做任何操作**)
|
||||
# - /app/ = 已到 SPA 主页
|
||||
# - /select = sandbox 中间页
|
||||
# - /login 或 /signin = sandbox 又把人踢回登录页(根域登录后常见,
|
||||
# sandbox 会跳到 /industries/campus/smartcampus/login SPA 二级登录页)
|
||||
# - 如果 URL 跳到 login/signin,**立即**退出 30 秒等待(不再干等),
|
||||
# 让 4.5 处理。否则会等满 30 秒才发现要进 SPA 登录兜底,造成卡顿。
|
||||
spa_url_seen = False
|
||||
if not _wait_for_url_stable(
|
||||
page, ("/app/", "/select", "/login", "/signin"), max_wait_sec=30,
|
||||
):
|
||||
logger.warning(
|
||||
"platform_collector: 登录后 30s 内 URL 未稳定到 /app/、/select、"
|
||||
"/login 或 /signin(当前 URL=%s)",
|
||||
page.url,
|
||||
)
|
||||
|
||||
# 4.5. v1.1.28:URL 落在 SPA 二级登录页 → 主动等 input hydrate(最多 10 秒)
|
||||
# → 主动 fill demo/123456(sandbox SPA 默认 mock)→ 点登录 → 等 URL 跳离
|
||||
# v1.1.27 失败的根因:URL 已稳定在 /login 时,React SPA input 可能还没渲染,
|
||||
# _fill_demo_on_spa_login 找不到 input → False → 卡 180 秒等用户输入。
|
||||
# 修复:先等 input 真正 hydrate 出来,再 fill+click。
|
||||
cur_url_after_stable = page.url.lower()
|
||||
if "/login" in cur_url_after_stable or "/signin" in cur_url_after_stable:
|
||||
logger.warning(
|
||||
"platform_collector: 登录后落在 SPA 二级登录页(%s),"
|
||||
"v1.1.28 触发 SPA 登录兜底(等 input hydrate + fill demo/123456)",
|
||||
page.url,
|
||||
)
|
||||
# 1. 等 input 真正 hydrate 出来(最多 10 秒)
|
||||
try:
|
||||
page.wait_for_selector(
|
||||
"#jc-user, #jc-pass, input[type='tel'], input[type='password']",
|
||||
timeout=10_000,
|
||||
state="attached",
|
||||
)
|
||||
page.wait_for_timeout(500) # 让 React 状态稳定
|
||||
except Exception as e:
|
||||
logger.warning(
|
||||
"platform_collector: 自动 fill 失败(URL=%s),fallback 手工登录等待",
|
||||
page.url,
|
||||
"platform_collector: SPA 登录页 10s 内未渲染 input: %s", e,
|
||||
)
|
||||
|
||||
# 2. fill demo/123456 + 点登录(不管当前 input 有没有值,强制覆盖)
|
||||
spa_filled = _fill_demo_on_spa_login(page)
|
||||
if spa_filled:
|
||||
logger.info(
|
||||
"platform_collector: ✅ SPA 二级登录页 demo/123456 自动填好,"
|
||||
"已点登录,等 URL 跳到 /app/ 或 /select"
|
||||
)
|
||||
# 等 URL 跳离 login(30 秒内跳到 /app/ 或 /select 都算成功)
|
||||
if not _wait_for_url_stable(page, ("/app/", "/select"), max_wait_sec=30):
|
||||
logger.warning(
|
||||
"platform_collector: SPA 登录后 30s 内 URL 未稳定到 /app/或/select"
|
||||
"(当前 URL=%s,demo 可能失效需手工登录)",
|
||||
page.url,
|
||||
)
|
||||
# demo 失效:等用户手工登录(headless 模式直接抛错)
|
||||
if not headless:
|
||||
_wait_for_manual_login(page, wait_sec=human_wait_timeout)
|
||||
else:
|
||||
raise LoginTimeoutError(
|
||||
f"SPA 二级登录页 demo/123456 填了但 URL 不跳离"
|
||||
f"(可能 demo 失效或需人工登录,URL={page.url})"
|
||||
)
|
||||
else:
|
||||
# fill 失败(input 仍未渲染)→ 等用户手工输入
|
||||
logger.warning(
|
||||
"platform_collector: SPA 登录页 _fill_demo 失败"
|
||||
"(input 仍未渲染),fallback 等用户输入"
|
||||
)
|
||||
if not headless:
|
||||
_wait_for_manual_login(page, wait_sec=human_wait_timeout)
|
||||
if _wait_for_login_inputs_filled(page, wait_sec=human_wait_timeout):
|
||||
_wait_for_url_stable(page, ("/app/", "/select"), max_wait_sec=30)
|
||||
else:
|
||||
_wait_for_manual_login(page, wait_sec=human_wait_timeout)
|
||||
else:
|
||||
# headless 模式无法手工登录:抛 LoginTimeoutError 让 CLI 决定
|
||||
raise LoginTimeoutError(
|
||||
f"headless 模式下自动 fill 失败,且无法手工登录"
|
||||
f"(请用 --show-browser 重试)"
|
||||
f"SPA 二级登录页 demo fill 失败且 headless 无法手工登录"
|
||||
f"(URL={page.url})"
|
||||
)
|
||||
else:
|
||||
# 等 SPA 路由稳定(之前这里没有 wait,直接 goto home 会打断 pending 跳转)
|
||||
page.wait_for_timeout(POST_LOGIN_SETTLE_MS)
|
||||
|
||||
# 4. 强制 goto SPA home(如果还停在 login,重试登录一次)
|
||||
for retry in range(SPA_HOME_RETRY_MAX):
|
||||
if _is_logged_in(page):
|
||||
break
|
||||
# 5. v1.1.22:显式跳到 SPA_HOME_URL(/app/home),等侧边栏 link 出现
|
||||
# 避免 sandbox 在 /select 中间页时直接找 sidebar link 找不到
|
||||
if "/app/home" not in page.url.lower():
|
||||
logger.info(
|
||||
"platform_collector: 强制 goto SPA_HOME (retry=%d), 当前 URL=%s",
|
||||
retry, page.url,
|
||||
"platform_collector: 当前 URL=%s,跳到 SPA_HOME_URL=%s",
|
||||
page.url, SPA_HOME_URL,
|
||||
)
|
||||
_goto(page, SPA_HOME_URL)
|
||||
page.wait_for_timeout(SPA_HOME_RETRY_DELAY_MS)
|
||||
if _fill_spa_login(page):
|
||||
page.wait_for_timeout(POST_LOGIN_SETTLE_MS)
|
||||
|
||||
# 4.5. retry 后若仍未登录,headless 模式抛超时
|
||||
if not _is_logged_in(page):
|
||||
if headless:
|
||||
raise LoginTimeoutError(
|
||||
"SPA_HOME retry 全部失败,URL 仍未离开 login 页"
|
||||
page.wait_for_timeout(SPA_HYDRATION_MS)
|
||||
# 等 URL 稳定到 /app/home(最多 15s,**不做任何操作**)
|
||||
if not _wait_for_url_stable(page, "/app/home", max_wait_sec=15):
|
||||
logger.warning(
|
||||
"platform_collector: 15s 内未稳定到 /app/home(当前 URL=%s)",
|
||||
page.url,
|
||||
)
|
||||
_wait_for_manual_login(page, wait_sec=human_wait_timeout)
|
||||
|
||||
# 5. 等 SPA 渲染 + click 侧边栏 hr/contracts(SPA 路由,避免 goto 触发整页 reload 丢 session)
|
||||
# 6. 等 SPA 渲染 + click 侧边栏 hr/contracts(SPA 路由,避免 goto 触发整页 reload 丢 session)
|
||||
try:
|
||||
# dump SPA home 一次(用于调试 selector)
|
||||
page.wait_for_timeout(POST_LOGIN_SETTLE_MS)
|
||||
_dump_page(page, download_dir.parent / "spa_home_dump.html", "SPA home")
|
||||
|
||||
# 等 SPA 真正加载完(侧边栏 link 出现,最长 SIDEBAR_LINK_TIMEOUT_MS)
|
||||
# 修复:之前 15s 太短,sandbox mock 加载慢时直接 fallback 到 pushState → 表格加载不出来
|
||||
try:
|
||||
page.wait_for_selector(
|
||||
"a[href$='/hr/contracts']",
|
||||
|
||||
@@ -6,7 +6,7 @@
|
||||
"""
|
||||
|
||||
SKILL_SLUG = "alert-contract-expiring-hr"
|
||||
SKILL_VERSION = "1.1.19"
|
||||
SKILL_VERSION = "1.1.28"
|
||||
LOG_LOGGER_NAME = "openclaw.skill.alert_contract_expiring_hr"
|
||||
PLATFORM_KIT_MIN_VERSION = "1.2.11"
|
||||
|
||||
@@ -48,7 +48,10 @@ PERFORMANCE_RENEW_MIN = 80 # 全部 >= 80 且逐年上升 → renew
|
||||
# ----- v1.1.6:RPA 超时常量(与 skill-template human_verification.py 对齐)-----
|
||||
# 手工登录等待(HITL fallback)
|
||||
HUMAN_WAIT_TIMEOUT = 180 # 手工登录等待超时(秒)—— skill-template 同名常量
|
||||
HUMAN_POLL_INTERVAL = 2.0 # 轮询间隔(秒)—— skill-template 同名常量
|
||||
HUMAN_POLL_INTERVAL = 5.0 # 轮询间隔(秒)—— v1.1.19 调长:避免打断用户输入
|
||||
HUMAN_PROMPT_INTERVAL = 15.0 # 剩余时间提示间隔(秒)—— v1.1.19 新增
|
||||
URL_STABLE_REQUIRED_S = 2.0 # URL 连续 N 秒不变才算稳定 —— v1.1.19 新增
|
||||
URL_STABLE_POLL_S = 0.5 # URL 稳定探测间隔(秒)
|
||||
|
||||
# Playwright 操作超时(毫秒)
|
||||
DEFAULT_PAGE_TIMEOUT_MS = 15_000 # Playwright 默认操作超时 —— skill-template 同名常量
|
||||
|
||||
@@ -1271,6 +1271,382 @@ def test_enqueue_download_tasks_safe_empty_list() -> tuple[bool, str]:
|
||||
return True, "空列表 no-op OK"
|
||||
|
||||
|
||||
# ---------- v1.1.19: _wait_for_url_stable + login flow ----------
|
||||
|
||||
|
||||
class _FakePage:
|
||||
"""v1.1.19 login flow 测试用的简易 page mock。
|
||||
|
||||
- .url: 字符串
|
||||
- 支持外部修改 .url 模拟用户登录跳转
|
||||
- v1.1.24:新增 .locator() 返回 _FakeLocatorNoOp,让
|
||||
_wait_for_login_inputs_filled 的 selector 探测不抛 AttributeError
|
||||
"""
|
||||
|
||||
def __init__(self, initial_url: str = "about:blank") -> None:
|
||||
self.url = initial_url
|
||||
self.goto_calls: list[str] = []
|
||||
self.reload_calls: int = 0
|
||||
self.click_calls: int = 0
|
||||
|
||||
def locator(self, selector: str): # noqa: ARG002
|
||||
"""v1.1.24 扩展:测试 _wait_for_login_inputs_filled 用,返回空 locator。"""
|
||||
return _FakeLocator(count=0)
|
||||
|
||||
|
||||
def test_wait_for_url_stable_immediate_match() -> tuple[bool, str]:
|
||||
"""_wait_for_url_stable: URL 已含目标 + 稳定 → 立即返回 True。"""
|
||||
from service.platform_collector import _wait_for_url_stable
|
||||
|
||||
page = _FakePage("https://example.com/app/contracts")
|
||||
if not _wait_for_url_stable(page, "/app/", max_wait_sec=5):
|
||||
return False, "URL 已含 /app/ 应立即返回 True"
|
||||
if page.goto_calls or page.reload_calls:
|
||||
return False, f"不应触发 goto/reload: {page.goto_calls}/{page.reload_calls}"
|
||||
return True, "URL 已含 /app/ 立即返回 OK"
|
||||
|
||||
|
||||
def test_wait_for_url_stable_transition() -> tuple[bool, str]:
|
||||
"""_wait_for_url_stable: URL 经历 login → /app/ 过渡,等稳定后返回 True。"""
|
||||
import threading
|
||||
import time as _time
|
||||
|
||||
from service.platform_collector import _wait_for_url_stable
|
||||
|
||||
page = _FakePage("https://example.com/login")
|
||||
|
||||
def flip_to_app():
|
||||
_time.sleep(0.6)
|
||||
page.url = "https://example.com/app/home"
|
||||
|
||||
threading.Thread(target=flip_to_app, daemon=True).start()
|
||||
if not _wait_for_url_stable(page, "/app/", max_wait_sec=5):
|
||||
return False, f"应等 URL 稳定到 /app/,但超时(最终 URL={page.url})"
|
||||
return True, f"URL 过渡后稳定 OK(最终 URL={page.url})"
|
||||
|
||||
|
||||
def test_wait_for_url_stable_timeout() -> tuple[bool, str]:
|
||||
"""_wait_for_url_stable: URL 始终未跳到目标 → 超时返回 False。"""
|
||||
from service.platform_collector import _wait_for_url_stable
|
||||
|
||||
page = _FakePage("https://example.com/login")
|
||||
if _wait_for_url_stable(page, "/app/", max_wait_sec=2):
|
||||
return False, "URL 永远含 login,应超时返回 False"
|
||||
return True, "超时返回 False OK"
|
||||
|
||||
|
||||
def test_wait_for_manual_login_no_goto() -> tuple[bool, str]:
|
||||
"""_wait_for_manual_login: 等待期间 **不应** 触发 goto / reload / click。"""
|
||||
import threading
|
||||
import time as _time
|
||||
|
||||
from service import platform_collector as pc
|
||||
|
||||
# platform_collector 在 import 时已经把常量复制到模块 globals;
|
||||
# 直接覆盖模块属性最稳。缩短 HUMAN_POLL_INTERVAL 和 URL_STABLE_REQUIRED_S,
|
||||
# 让 fake_user 0.8s 翻 URL + 0.3s 稳定窗口能凑进 wait_sec=5。
|
||||
orig_poll = pc.HUMAN_POLL_INTERVAL
|
||||
orig_stable = pc.URL_STABLE_REQUIRED_S
|
||||
pc.HUMAN_POLL_INTERVAL = 0.3
|
||||
pc.URL_STABLE_REQUIRED_S = 0.3
|
||||
try:
|
||||
page = _FakePage("https://example.com/login")
|
||||
|
||||
def fake_user_login():
|
||||
_time.sleep(0.8)
|
||||
page.url = "https://example.com/app/home"
|
||||
|
||||
threading.Thread(target=fake_user_login, daemon=True).start()
|
||||
try:
|
||||
pc._wait_for_manual_login(page, wait_sec=5)
|
||||
except Exception as e:
|
||||
return False, f"应等用户登录成功,但抛异常: {e}"
|
||||
|
||||
if page.goto_calls:
|
||||
return False, f"等待期间不应 goto: {page.goto_calls}"
|
||||
if page.reload_calls:
|
||||
return False, f"等待期间不应 reload: {page.reload_calls}"
|
||||
return True, "等用户登录期间无打断 OK"
|
||||
finally:
|
||||
pc.HUMAN_POLL_INTERVAL = orig_poll
|
||||
pc.URL_STABLE_REQUIRED_S = orig_stable
|
||||
|
||||
|
||||
# ---------- v1.1.24: _try_auto_sign_in SPA 二级登录页(中文「登录」按钮)----------
|
||||
|
||||
|
||||
class _FakeLocator:
|
||||
"""v1.1.24 _try_auto_sign_in 测试用的简易 Playwright Locator mock。
|
||||
|
||||
支持的 API:
|
||||
- .count() -> int
|
||||
- .input_value() -> str
|
||||
- .click() -> None
|
||||
- .first -> 自身
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
count: int = 0,
|
||||
value: str = "",
|
||||
click_log: list | None = None,
|
||||
) -> None:
|
||||
self._count = count
|
||||
self._value = value
|
||||
self._click_log = click_log if click_log is not None else []
|
||||
|
||||
@property
|
||||
def first(self) -> "_FakeLocator":
|
||||
return self
|
||||
|
||||
def count(self) -> int:
|
||||
return self._count
|
||||
|
||||
def input_value(self) -> str:
|
||||
return self._value
|
||||
|
||||
def click(self) -> None:
|
||||
self._click_log.append(self._value or "clicked")
|
||||
|
||||
def fill(self, value: str) -> None:
|
||||
"""v1.1.27:fill 后 value 立即更新。"""
|
||||
self._value = value
|
||||
|
||||
|
||||
class _FakeLocatorRouter:
|
||||
"""根据 selector 返回对应 _FakeLocator 的路由器。
|
||||
|
||||
selectors: dict[selector_str, _FakeLocator]
|
||||
"""
|
||||
|
||||
def __init__(self, selectors: dict[str, _FakeLocator], page) -> None:
|
||||
self._selectors = selectors
|
||||
self._page = page
|
||||
|
||||
def __call__(self, sel: str):
|
||||
"""page.locator(sel) 形式调用。"""
|
||||
if sel in self._selectors:
|
||||
return self._selectors[sel]
|
||||
return _FakeLocator(count=0)
|
||||
|
||||
def __getattr__(self, sel: str):
|
||||
# loc.page 已存在则按 selector 路由
|
||||
if sel in self._selectors:
|
||||
return self._selectors[sel]
|
||||
# fallback:未注册的 selector 都返回空 locator
|
||||
return _FakeLocator(count=0)
|
||||
|
||||
|
||||
class _FakePlaywrightPage:
|
||||
"""v1.1.24 _try_auto_sign_in 测试用的 page mock。
|
||||
|
||||
usage:
|
||||
page = _FakePlaywrightPage(
|
||||
url="https://sandbox.jc2009.com/industries/campus/smartcampus/login",
|
||||
selectors={
|
||||
"input[type='tel']": _FakeLocator(count=1, value="18677308191"),
|
||||
"input[type='password']": _FakeLocator(count=1, value="***"),
|
||||
"button:has-text('登录')": _FakeLocator(count=1, click_log=click_log),
|
||||
},
|
||||
wait_for_selector_log=wait_log,
|
||||
)
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
url: str = "about:blank",
|
||||
selectors: dict[str, _FakeLocator] | None = None,
|
||||
wait_for_selector_log: list | None = None,
|
||||
wait_selector_should_raise: bool = False,
|
||||
) -> None:
|
||||
self.url = url
|
||||
self._selectors = selectors or {}
|
||||
self._wait_log = wait_for_selector_log if wait_for_selector_log is not None else []
|
||||
self._wait_should_raise = wait_selector_should_raise
|
||||
|
||||
def wait_for_selector(self, selector: str, timeout: int = 0) -> None:
|
||||
self._wait_log.append(selector)
|
||||
if self._wait_should_raise:
|
||||
raise Exception("fake: timeout")
|
||||
|
||||
@property
|
||||
def locator(self):
|
||||
return _FakeLocatorRouter(self._selectors, self)
|
||||
|
||||
|
||||
def test_try_auto_sign_in_spa_login_chinese_button() -> tuple[bool, str]:
|
||||
"""v1.1.24:SPA 二级登录页(/industries/campus/smartcampus/login),
|
||||
input 用 `input[type='tel']` / `input[type='password']`,
|
||||
按钮中文「登录」。预填有值 → 自动点登录。
|
||||
"""
|
||||
from service.platform_collector import _try_auto_sign_in
|
||||
|
||||
click_log: list[str] = []
|
||||
page = _FakePlaywrightPage(
|
||||
url="https://sandbox.jc2009.com/industries/campus/smartcampus/login",
|
||||
selectors={
|
||||
"input[type='tel']": _FakeLocator(count=1, value="18677308191"),
|
||||
"input[type='password']": _FakeLocator(count=1, value="dummy"),
|
||||
"button:has-text('登录')": _FakeLocator(count=1, click_log=click_log),
|
||||
},
|
||||
)
|
||||
if not _try_auto_sign_in(page, wait_input_sec=2):
|
||||
return False, "SPA 二级登录页预填应自动点 登录"
|
||||
if not click_log:
|
||||
return False, f"未触发点击: {click_log}"
|
||||
return True, f"SPA 二级登录页中文按钮自动点 OK(click_log={click_log})"
|
||||
|
||||
|
||||
def test_try_auto_sign_in_root_sign_in_english_button() -> tuple[bool, str]:
|
||||
"""v1.1.24(向后兼容):根域登录页(https://sandbox.jc2009.com/),
|
||||
input 用 `#jc-user` / `#jc-pass`,按钮英文「Sign in」。预填有值 → 自动点。
|
||||
"""
|
||||
from service.platform_collector import _try_auto_sign_in
|
||||
|
||||
click_log: list[str] = []
|
||||
page = _FakePlaywrightPage(
|
||||
url="https://sandbox.jc2009.com/",
|
||||
selectors={
|
||||
"#jc-user": _FakeLocator(count=1, value="demo"),
|
||||
"#jc-pass": _FakeLocator(count=1, value="123456"),
|
||||
"button:has-text('Sign in')": _FakeLocator(count=1, click_log=click_log),
|
||||
},
|
||||
)
|
||||
if not _try_auto_sign_in(page, wait_input_sec=2):
|
||||
return False, "根域登录页预填应自动点 Sign in"
|
||||
if not click_log:
|
||||
return False, f"未触发点击: {click_log}"
|
||||
return True, f"根域登录页 Sign in 自动点 OK(click_log={click_log})"
|
||||
|
||||
|
||||
def test_try_auto_sign_in_no_input_returns_false() -> tuple[bool, str]:
|
||||
"""v1.1.24:input 为空(用户还没输入)→ 返回 False,不点按钮。"""
|
||||
from service.platform_collector import _try_auto_sign_in
|
||||
|
||||
click_log: list[str] = []
|
||||
page = _FakePlaywrightPage(
|
||||
url="https://sandbox.jc2009.com/industries/campus/smartcampus/login",
|
||||
selectors={
|
||||
"input[type='tel']": _FakeLocator(count=1, value=""), # 空
|
||||
"input[type='password']": _FakeLocator(count=1, value="123"),
|
||||
"button:has-text('登录')": _FakeLocator(count=1, click_log=click_log),
|
||||
},
|
||||
)
|
||||
if _try_auto_sign_in(page, wait_input_sec=2):
|
||||
return False, "user 空时不应自动点"
|
||||
if click_log:
|
||||
return False, f"未填完不应点击: {click_log}"
|
||||
return True, "input 空返回 False 不点 OK"
|
||||
|
||||
|
||||
def test_try_auto_sign_in_selector_timeout_returns_false() -> tuple[bool, str]:
|
||||
"""v1.1.24:等 input 超时(如 URL 已是 SPA home,无登录 input)→ 返回 False。"""
|
||||
from service.platform_collector import _try_auto_sign_in
|
||||
|
||||
click_log: list[str] = []
|
||||
page = _FakePlaywrightPage(
|
||||
url="https://sandbox.jc2009.com/industries/campus/smartcampus/app/home",
|
||||
selectors={}, # 没有任何 input
|
||||
wait_selector_should_raise=True,
|
||||
)
|
||||
if _try_auto_sign_in(page, wait_input_sec=1):
|
||||
return False, "无登录 input 应返回 False"
|
||||
if click_log:
|
||||
return False, f"无 input 不应点击: {click_log}"
|
||||
return True, "无登录 input 超时返回 False OK"
|
||||
|
||||
|
||||
def test_wait_for_login_inputs_filled_auto_click_v1_26() -> tuple[bool, str]:
|
||||
"""v1.1.26:`_wait_for_login_inputs_filled` **自动点登录按钮**——
|
||||
按用户最新要求:「到了 /industries/campus/smartcampus/login 整个页面,
|
||||
有账号和密码可用直接点登录跳转」。
|
||||
|
||||
input 有值时点击 `button:has-text('登录')` 或 `Sign in`(多按钮文本优先级)。
|
||||
"""
|
||||
from service import platform_collector as pc
|
||||
|
||||
orig_poll = pc.HUMAN_POLL_INTERVAL
|
||||
orig_prompt = pc.HUMAN_PROMPT_INTERVAL
|
||||
pc.HUMAN_POLL_INTERVAL = 0.2
|
||||
pc.HUMAN_PROMPT_INTERVAL = 0.2
|
||||
try:
|
||||
click_log: list[str] = []
|
||||
page = _FakePlaywrightPage(
|
||||
url="https://sandbox.jc2009.com/industries/campus/smartcampus/login",
|
||||
selectors={
|
||||
"input[type='tel']": _FakeLocator(count=1, value="18677308191"),
|
||||
"input[type='password']": _FakeLocator(count=1, value="dummy"),
|
||||
"button:has-text('登录')": _FakeLocator(count=1, click_log=click_log),
|
||||
},
|
||||
)
|
||||
|
||||
result = pc._wait_for_login_inputs_filled(page, wait_sec=5)
|
||||
if not result:
|
||||
return False, "input 有值应自动点登录 → return True"
|
||||
|
||||
if not click_log:
|
||||
return False, f"v1.1.26 应自动点登录按钮,但 click_log={click_log}"
|
||||
return True, f"v1.1.26 自动点登录按钮 OK(click_log={click_log})"
|
||||
finally:
|
||||
pc.HUMAN_POLL_INTERVAL = orig_poll
|
||||
pc.HUMAN_PROMPT_INTERVAL = orig_prompt
|
||||
|
||||
|
||||
def test_fill_demo_on_spa_login_basic_v1_27() -> tuple[bool, str]:
|
||||
"""v1.1.27:`_fill_demo_on_spa_login` ——SPA 二级登录页 input 默认不预填,
|
||||
必须主动 fill demo/123456 再点登录按钮。
|
||||
"""
|
||||
click_log: list[str] = []
|
||||
page = _FakePlaywrightPage(
|
||||
url="https://sandbox.jc2009.com/industries/campus/smartcampus/login",
|
||||
selectors={
|
||||
"#jc-user": _FakeLocator(count=1, value=""), # 空 → 要 fill
|
||||
"#jc-pass": _FakeLocator(count=1, value=""),
|
||||
"button:has-text('Sign in')": _FakeLocator(count=1, click_log=click_log),
|
||||
},
|
||||
)
|
||||
from service.platform_collector import _fill_demo_on_spa_login
|
||||
|
||||
if not _fill_demo_on_spa_login(page):
|
||||
return False, "SPA login fill demo 应成功"
|
||||
if not click_log:
|
||||
return False, f"应点 Sign in,但 click_log={click_log}"
|
||||
return True, f"SPA 二级登录页 demo/123456 自动 fill+点 OK(click_log={click_log})"
|
||||
|
||||
|
||||
def test_fill_demo_on_spa_login_chinese_button_v1_27() -> tuple[bool, str]:
|
||||
"""v1.1.27:中文「登录」按钮(用户截图版本)也能 fill demo/123456 并点。"""
|
||||
click_log: list[str] = []
|
||||
page = _FakePlaywrightPage(
|
||||
url="https://sandbox.jc2009.com/industries/campus/smartcampus/login",
|
||||
selectors={
|
||||
"input[type='tel']": _FakeLocator(count=1, value=""),
|
||||
"input[type='password']": _FakeLocator(count=1, value=""),
|
||||
"button:has-text('登录')": _FakeLocator(count=1, click_log=click_log),
|
||||
},
|
||||
)
|
||||
from service.platform_collector import _fill_demo_on_spa_login
|
||||
|
||||
if not _fill_demo_on_spa_login(page):
|
||||
return False, "SPA login 中文按钮 fill demo 应成功"
|
||||
if not click_log:
|
||||
return False, f"应点 登录,但 click_log={click_log}"
|
||||
return True, f"SPA 二级登录页中文「登录」按钮 OK(click_log={click_log})"
|
||||
|
||||
|
||||
def test_fill_demo_on_spa_login_no_input_returns_false_v1_27() -> tuple[bool, str]:
|
||||
"""v1.1.27:找不到 user input(页面不是 login 页)→ 返回 False。"""
|
||||
page = _FakePlaywrightPage(
|
||||
url="https://sandbox.jc2009.com/industries/campus/smartcampus/app/home",
|
||||
selectors={},
|
||||
)
|
||||
from service.platform_collector import _fill_demo_on_spa_login
|
||||
|
||||
if _fill_demo_on_spa_login(page):
|
||||
return False, "无 login input 应返回 False"
|
||||
return True, "无 login input 返回 False OK"
|
||||
|
||||
|
||||
# ---------- v1.1.18: show-config / logs-list ----------
|
||||
|
||||
|
||||
@@ -1540,6 +1916,18 @@ def main() -> int:
|
||||
_run("download_tasks/enqueue_safe_idempotent", test_enqueue_download_tasks_safe_idempotent)
|
||||
_run("download_tasks/enqueue_safe_fk_failure", test_enqueue_download_tasks_safe_fk_failure)
|
||||
_run("download_tasks/enqueue_safe_empty_list", test_enqueue_download_tasks_safe_empty_list)
|
||||
_run("login_flow/wait_url_stable_immediate", test_wait_for_url_stable_immediate_match)
|
||||
_run("login_flow/wait_url_stable_transition", test_wait_for_url_stable_transition)
|
||||
_run("login_flow/wait_url_stable_timeout", test_wait_for_url_stable_timeout)
|
||||
_run("login_flow/wait_manual_login_no_goto", test_wait_for_manual_login_no_goto)
|
||||
_run("login_flow/auto_sign_in_spa_chinese_button", test_try_auto_sign_in_spa_login_chinese_button)
|
||||
_run("login_flow/auto_sign_in_root_english_button", test_try_auto_sign_in_root_sign_in_english_button)
|
||||
_run("login_flow/auto_sign_in_no_input_returns_false", test_try_auto_sign_in_no_input_returns_false)
|
||||
_run("login_flow/auto_sign_in_selector_timeout_returns_false", test_try_auto_sign_in_selector_timeout_returns_false)
|
||||
_run("login_flow/wait_login_inputs_filled_auto_click_v1_26", test_wait_for_login_inputs_filled_auto_click_v1_26)
|
||||
_run("login_flow/fill_demo_on_spa_login_basic_v1_27", test_fill_demo_on_spa_login_basic_v1_27)
|
||||
_run("login_flow/fill_demo_on_spa_login_chinese_button_v1_27", test_fill_demo_on_spa_login_chinese_button_v1_27)
|
||||
_run("login_flow/fill_demo_on_spa_login_no_input_returns_false_v1_27", test_fill_demo_on_spa_login_no_input_returns_false_v1_27)
|
||||
_run("show_config/masks_tokens", test_show_config_masks_tokens)
|
||||
_run("show_config/skips_empty", test_show_config_skips_empty)
|
||||
_run("show_config/skips_comments", test_show_config_skips_comments)
|
||||
|
||||
Reference in New Issue
Block a user