chore: auto release commit (2026-09-15 03:29:00)
All checks were successful
技能自动化发布 / release (push) Successful in 11s

This commit is contained in:
2026-09-15 03:29:00 +08:00
parent 689810eaeb
commit 22b42f8d30
9 changed files with 2063 additions and 1738 deletions

View File

@@ -0,0 +1,743 @@
"""Playwright RPA:SmartCampus HR 平台登录流程。
v1.1.31:从 platform_collector.py 拆分出来的纯登录流程子模块。
包含 12 个函数:
- _try_auto_sign_in / _wait_for_login_inputs_filled / _is_login_page_url_for_user
- _wait_for_manual_login / _wait_for_url_stable / _fill_root_login
- _fill_spa_login_from_env / _is_dev_mode_panel_open / _resolve_target_industry
- _click_dev_mode_sign_in_for_industry / _click_campus_link / _fill_spa_login
设计要点(保持 v1.1.30 不变):
- 所有 facade 公开符号(HUMAN_POLL_INTERVAL / LoginTimeoutError / _goto / _dump_page /
_is_logged_in / _is_spa_routed)通过 `from service import platform_collector as _pc` lazy import
→ 保证测试 `pc.HUMAN_POLL_INTERVAL = 0.001` monkey-patch 生效
- `HUMAN_WAIT_TIMEOUT` / `SPA_LOGIN_*` / `SIDEBAR_*` / `URL_STABLE_*` 等不会动态 patch 的常量
从 `util.constants` 顶部 import
"""
from __future__ import annotations
import time
from util.constants import (
HUMAN_PROMPT_INTERVAL,
HUMAN_WAIT_TIMEOUT,
SIDEBAR_LINK_TIMEOUT_MS,
SPA_LOGIN_INPUT_TIMEOUT_MS,
SPA_LOGIN_SETTLE_MS,
SPA_LOGIN_URL_TIMEOUT_MS,
URL_AFTER_LOGIN_TIMEOUT_MS,
URL_STABLE_POLL_S,
)
from util.logging_setup import get_skill_logger
logger = get_skill_logger()
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`。
"""
# Lazy import: 测试通过 pc.HUMAN_POLL_INTERVAL = X 加速,需要 facade attr lookup
from service import platform_collector as _pc
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(_pc.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(v1.1.19 重写,v1.1.20 加强,v1.1.21 修订):
等用户在浏览器完成登录。
判定「登录完成」的多个信号(满足任一即视为完成):
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 主页
- **不做任何 goto / reload / click**,避免打断用户输入
- 每 HUMAN_PROMPT_INTERVAL 秒打印一次剩余时间(默认 15s)
- 每 HUMAN_POLL_INTERVAL 秒检查一次(默认 5s)
- 超时抛 LoginTimeoutError。
"""
# Lazy import: 读 facade constants 与异常类,便于测试 patch
from service import platform_collector as _pc
print(
f"[登录] 请在浏览器中完成登录"
f"(sandbox demo/123456 或真实账号),最多等待 {wait_sec} 秒..."
)
deadline = time.monotonic() + wait_sec
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 _pc._is_logged_in(page):
stable_for = time.monotonic() - last_url_change_at
if stable_for >= _pc.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 >= _pc.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 >= _pc.HUMAN_PROMPT_INTERVAL:
last_prompt_at = now
print(
f"[登录] 等待用户输入... 剩余 {int(remaining)}s "
f"(当前 URL={cur_url})"
)
time.sleep(_pc.HUMAN_POLL_INTERVAL)
raise _pc.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 = 超时。
"""
# Lazy import: 读 URL_STABLE_REQUIRED_S 便于测试 patch
from service import platform_collector as _pc
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 >= _pc.URL_STABLE_REQUIRED_S:
return True
time.sleep(URL_STABLE_POLL_S)
return False
def _fill_root_login(page) -> bool:
"""v1.1.31:根域登录页(如还在):从 .env 读 SMARTCAMPUS_LOGIN_USER /
SMARTCAMPUS_LOGIN_PASSWORD(不硬编码 demo/123456)→ fill → click Sign in。
无 .env 凭证时 return False,调用方走「用户手工登录」分支。
"""
# Lazy import: 读 _get_env(保留在 facade,测试通过 pc._ENV_CACHE 隔离)
from service import platform_collector as _pc
user_input = page.locator('#jc-user').first
if user_input.count() == 0:
return False
root_user = _pc._get_env("SMARTCAMPUS_LOGIN_USER", "").strip()
root_pass = _pc._get_env("SMARTCAMPUS_LOGIN_PASSWORD", "").strip()
if not root_user or not root_pass:
logger.info(
"platform_collector._fill_root_login: .env 未配置 "
"SMARTCAMPUS_LOGIN_USER / SMARTCAMPUS_LOGIN_PASSWORD,"
"**不自动 fill**,等用户手工输入"
)
return False
user_input.fill(root_user)
page.locator('#jc-pass').first.fill(root_pass)
page.locator('button[type="submit"]:has-text("Sign in")').first.click()
logger.info("platform_collector: 根域自动登录(user 来自 .env)...")
try:
page.wait_for_url(
lambda u: "/home" in u or "/select" in u or "/industries/" in u,
timeout=URL_AFTER_LOGIN_TIMEOUT_MS,
)
except Exception as e:
logger.warning("等根域登录跳转超时: %s", e)
logger.info("platform_collector: 根域登录后 URL=%s", page.url)
return True
def _fill_spa_login_from_env(page) -> bool:
"""v1.1.31/32:SPA 二级登录页(/industries/campus/smartcampus/login)
按用户要求「密码不能直接放到代码」—— 从 .env 读 SMARTCAMPUS_LOGIN_USER /
SMARTCAMPUS_LOGIN_PASSWORD(无值时不自动 fill,等用户手工输入)。
v1.1.32:sandbox SPA 登录页默认会把「教师(通用)」demo 凭据预填到主表单 input。
但用户实际要登录的可能是「人事处」「财务处」等不同行业 tile;
按用户截图场景,需先在右侧 DEV MODE 面板里点对应 tile 的 Sign in,
让 SPA 把对应行业的 demo 凭据搬到主表单,再走原 fill+click 流程。
行业优先级:.env SMARTCAMPUS_LOGIN_INDUSTRY(默认 hr)→ 探测页面已预填的 user 值。
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]。
返回是否成功点登录按钮(**无 .env 凭证时返回 False,让用户手工输入**)。
"""
# Lazy import: 读 _get_env(facade 持有 _ENV_CACHE)
from service import platform_collector as _pc
# 1. 从 .env 读凭证(不硬编码)
spa_user = _pc._get_env("SMARTCAMPUS_LOGIN_USER", "").strip()
spa_pass = _pc._get_env("SMARTCAMPUS_LOGIN_PASSWORD", "").strip()
if not spa_user or not spa_pass:
logger.info(
"platform_collector._fill_spa_login_from_env: .env 未配置 "
"SMARTCAMPUS_LOGIN_USER / SMARTCAMPUS_LOGIN_PASSWORD,"
"**不自动 fill**,等用户在浏览器手工输入"
)
return False
# 1.5 v1.1.32:检测 DEV MODE 面板是否展开 → 是否需要先点对应行业的 Sign in
# sandbox 行为:教师(通用)demo 总是预填到主表单,登录其他行业(hr/finance/research)
# 需要在右侧 DEV MODE 面板里点对应 tile 的 Sign in。
if _is_dev_mode_panel_open(page):
target_industry = _resolve_target_industry(page)
logger.info(
"platform_collector: DEV MODE 面板展开,目标行业=%s", target_industry,
)
_click_dev_mode_sign_in_for_industry(page, target_industry)
# 2. 找 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_spa_login_from_env: 找不到 user input "
"(#jc-user / tel / autocomplete=username 全 miss)"
)
return False
# 3. 找 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_spa_login_from_env: 找不到 pass input"
)
return False
user_loc.fill(spa_user)
pass_loc.fill(spa_pass)
logger.info(
"platform_collector: SPA 二级登录页 fill 完成(user 来自 .env)"
)
# 4. 找登录按钮
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_spa_login_from_env: 找不到登录按钮"
)
return False
logger.info(
"platform_collector: SPA 二级登录页点 %s 按钮(user 来自 .env)",
btn_text_used,
)
btn.click()
return True
def _is_dev_mode_panel_open(page) -> bool:
"""v1.1.32:检测 sandbox SPA 登录页右侧 DEV MODE 面板是否展开。
面板里有「DEV MODE」文字 + 多个 tile(教师通用 / 人事处 / 财务处 / 科研处),
每个 tile 有自己的 Fill + Sign in 按钮。
返回 True 表示面板可见且可点对应行业的 Sign in。
实现策略(避免依赖 inner_text——fake page mock 不一定支持):
- 用 selector 探测至少 2 个 Sign in 按钮(教师/人事处/财务处/科研处 tile)
- 至少 1 个含「DEV MODE」文本的元素
"""
try:
# 1. 至少 2 个 Sign in 按钮(教师通用 + 至少 1 个行业 tile)
sign_in_btns = page.locator("button:has-text('Sign in')")
if sign_in_btns.count() < 2:
return False
# 2. DEV MODE 标志:找含 'DEV MODE' 文本的元素
dev_marker = page.locator(
"*:has-text('DEV MODE'), *:text-is('DEV MODE')"
).first
return dev_marker.count() > 0
except Exception:
return False
def _resolve_target_industry(page) -> str:
"""v1.1.32:解析要登录的目标行业。
优先级:
1. .env SMARTCAMPUS_LOGIN_INDUSTRY(可选,默认 'hr')
2. 探测页面:若主表单已预填 user 是 'hr' / 'finance' / 'research' / 'demo'
→ 直接用该值(说明用户已点过对应 tile 的 Fill)
"""
# Lazy import: 读 _get_env 拿 SMARTCAMPUS_LOGIN_INDUSTRY
from service import platform_collector as _pc
explicit = _pc._get_env("SMARTCAMPUS_LOGIN_INDUSTRY", "").strip().lower()
if explicit:
return explicit
# fallback:探测主表单 input 已预填的 user 值
try:
for sel in ("#jc-user", "input[type='tel']", "input[autocomplete='username']"):
loc = page.locator(sel).first
if loc.count() > 0:
cur = (loc.input_value() or "").strip().lower()
if cur in ("hr", "finance", "research", "demo"):
return cur
except Exception:
pass
return "hr"
def _click_dev_mode_sign_in_for_industry(page, industry: str) -> bool:
"""v1.1.32:在 DEV MODE 面板里点对应行业 tile 的 Sign in 按钮。
sandbox 行为:每个 tile 是一段包含「行业名 + Email + Password + Fill + Sign in」的 div,
用户点 Sign in 会让对应行业的 demo 凭据填到主表单。
selector 策略(简化版,避免依赖 inner_text——fake page mock 不一定支持):
- 行业显示名 → 用 :has-text + Sign in 按钮组合定位
- 兜底:用 nth 索引(教师通用=1 / 人事处=2 / 财务处=3 / 科研处=4)
返回是否成功点击(找不到时返回 False,由调用方决定 fallback)。
"""
if not industry:
return False
# 行业显示名 → 可能的中文标签
industry_aliases = {
"hr": ("人事处",),
"finance": ("财务处",),
"research": ("科研处",),
"demo": ("教师", "通用", "Teacher"),
}
aliases = industry_aliases.get(industry, (industry,))
# 1. 模式 A:找包含「行业中文标签」的容器 + 容器内 Sign in 按钮
for alias in aliases:
try:
# Playwright 支持 :has() / :has-text() / :text-matches()
tile_btn = page.locator(
f"div:has-text('{alias}'):has(button:has-text('Sign in')) "
f"button:has-text('Sign in')"
).first
if tile_btn.count() > 0:
logger.info(
"platform_collector: DEV MODE 点 %s tile 的 Sign in 按钮",
alias,
)
tile_btn.click()
return True
except Exception as e:
logger.warning(
"platform_collector: DEV MODE %s Sign in 点击失败: %s", alias, e,
)
# 2. 兜底:直接找所有 Sign in 按钮,按 industry 顺序选
# hr(人事处) 排第 2 个,finance(财务处) 排第 3 个,research(科研处) 排第 4 个
industry_order = {"demo": 1, "hr": 2, "finance": 3, "research": 4}
if industry in industry_order:
idx = industry_order[industry]
try:
all_sign_in = page.locator("button:has-text('Sign in')")
cnt = all_sign_in.count()
if cnt >= idx:
logger.info(
"platform_collector: DEV MODE 用兜底策略点第 %d 个 Sign in(%s)",
idx, industry,
)
all_sign_in.nth(idx - 1).click()
return True
except Exception as e:
logger.warning(
"platform_collector: DEV MODE 兜底点击失败: %s", e,
)
logger.warning(
"platform_collector: DEV MODE 找不到 industry=%s 的 Sign in 按钮", industry,
)
return False
def _click_campus_link(page) -> bool:
"""在 /select 页面点智慧校园 tile,返回是否点中。"""
link = page.locator("a[href*='/industries/campus']").first
if link.count() == 0:
logger.warning("platform_collector: 未找到 智慧校园 link")
return False
link.click()
logger.info("platform_collector: click 智慧校园 链接")
try:
page.wait_for_url(lambda u: "/industries/campus" in u, timeout=SIDEBAR_LINK_TIMEOUT_MS // 2)
except Exception as e:
logger.warning("等 sandbox 路径超时: %s;当前 URL=%s", e, page.url)
return True
def _fill_spa_login(page, settle_ms: int = SPA_LOGIN_SETTLE_MS) -> bool:
"""sandbox SPA 自动登录(如还在登录页):fill demo/123456 → click 登录 → 等跳转。
返回是否执行了登录(True = 已点登录按钮且 URL 已跳离 login;调用方可继续后续流程)。
失败时**不抛异常**,由调用方决定是否 fallback 到 _wait_for_manual_login。
v1.1.10 关键修复:**先检测 _is_logged_in,已登录态直接 return True 不再 fill**。
之前没这个检查,profile 复用时会被强制 demo fill 破坏 session。
v1.1.11 关键修复:**sandbox SPA 登录 input 实际是 `type="tel"` + 无 autocomplete**,
之前 selector `input[autocomplete="username"]` 永远找不到,自动 fill 100% 失败。
改用 sandbox 实际 selector `#jc-user` / `#jc-pass`(来自 root 登录页)+ `button[type="submit"]`。
修复要点(v1.1.5/v1.1.6 登录时间太短导致采集不到数据):
- 默认 settle_ms = SPA_LOGIN_SETTLE_MS(4000ms),给 SPA 异步渲染 + token 写入 cookie 时间
- 等 username input 的 timeout = SPA_LOGIN_INPUT_TIMEOUT_MS(20s)
- click 按钮后 wait_for_url timeout = SPA_LOGIN_URL_TIMEOUT_MS(30s)
- 多处过渡路径(/login, /signin, /auth/)一并判断
"""
# Lazy import: 读 _is_logged_in(facade 持有,便于测试 mock)
from service import platform_collector as _pc
# v1.1.10:先检查是否已登录(profile cookie 复用场景),已登录直接 return True
if _pc._is_logged_in(page):
logger.info(
"platform_collector: 已登录态(profile cookie 复用),跳过自动 fill"
)
return True
url_lower = page.url.lower()
if "/login" not in url_lower and "/signin" not in url_lower and "/auth/" not in url_lower:
return False
# v1.1.11:sandbox SPA 登录页 input 是 #jc-user / #jc-pass(与 root 登录页同 ID)
# 等输入框出现(带超时,避免 sandbox mock 慢启动卡死)
try:
page.wait_for_selector(
'#jc-user, input[type="tel"], input[autocomplete="username"]',
timeout=SPA_LOGIN_INPUT_TIMEOUT_MS,
)
except Exception as e:
logger.warning("SPA login input 未出现: %s;当前 URL=%s", e, page.url)
return False
# v1.1.11:按优先级选 selector —— sandbox 实际用 #jc-user
user_sel = None
for sel in ('#jc-user', 'input[type="tel"]', 'input[autocomplete="username"]'):
loc = page.locator(sel).first
if loc.count() > 0:
user_sel = sel
break
if not user_sel:
logger.warning("SPA login input selector 全 miss(#jc-user / type=tel / autocomplete=username)")
return False
page.locator(user_sel).first.fill("demo")
# 密码 selector 同样按优先级
pass_sel = None
for sel in ('#jc-pass', 'input[type="password"]', 'input[autocomplete="current-password"]'):
loc = page.locator(sel).first
if loc.count() > 0:
pass_sel = sel
break
if not pass_sel:
logger.warning("SPA password input selector 全 miss")
return False
page.locator(pass_sel).first.fill("123456")
page.locator(
'button:has-text("登录"), button[type="submit"]:has-text("Sign in"), '
'button[type="submit"]'
).first.click()
logger.info("platform_collector: sandbox SPA 自动登录 (user_sel=%s pass_sel=%s)", user_sel, pass_sel)
page.wait_for_timeout(settle_ms)
try:
page.wait_for_url(
lambda u: "/login" not in u.lower()
and "/signin" not in u.lower()
and "/auth/" not in u.lower(),
timeout=SPA_LOGIN_URL_TIMEOUT_MS,
)
except Exception as e:
logger.warning("SPA 登录超时: %s;当前 URL=%s", e, page.url)
return False
# v1.1.10:fill 后再检测 _is_logged_in,避免 sandbox demo 凭据失效时
# 被误判为「已登录」(URL 跳 /app/home 但 cookie 没写入)
if not _pc._is_logged_in(page):
logger.warning(
"platform_collector: SPA URL 跳离 login 但 _is_logged_in=False,登录可能失败"
)
return False
logger.info("platform_collector: SPA 登录后 URL=%s", page.url)
return True