Compare commits

...

4 Commits

Author SHA1 Message Date
689810eaeb chore: auto release commit (2026-09-15 02:56:48)
Some checks failed
技能自动化发布 / release (push) Failing after 4s
2026-09-15 02:56:49 +08:00
7584e69714 chore: auto release commit (2026-09-15 01:39:08)
Some checks failed
技能自动化发布 / release (push) Failing after 6s
2026-09-15 01:39:08 +08:00
d3fee3f46c chore: auto release commit (2026-09-15 01:30:18)
Some checks failed
技能自动化发布 / release (push) Failing after 4s
2026-09-15 01:30:18 +08:00
ebea77b5f2 chore: auto release commit (2026-09-15 01:27:04)
Some checks failed
技能自动化发布 / release (push) Failing after 4s
2026-09-15 01:27:05 +08:00
10 changed files with 984 additions and 62 deletions

View File

@@ -86,3 +86,20 @@ INPUT_DIR=
# 失败最大重试次数(默认 3)
DOWNLOAD_QUEUE_MAX_ATTEMPTS=
# ----- v1.1.31:SmartCampus HR 平台登录凭证(不硬编码到代码)-----
# 根域登录页(https://sandbox.jc2009.com/)和 SPA 二级登录页
# (/industries/campus/smartcampus/login)共用同一组凭证。
# 留空 → 脚本不自动 fill,等用户在浏览器手工输入。
SMARTCAMPUS_LOGIN_USER=
SMARTCAMPUS_LOGIN_PASSWORD=
# ----- v1.1.32:目标行业 tile(DEV MODE 面板)-----
# SPA 登录页右侧 DEV MODE 面板包含多个行业 tile(教师通用 / 人事处 / 财务处 / 科研处),
# 脚本自动点对应 tile 的 Sign in 把 demo 凭据搬到主表单。可选:
# hr → 人事处(默认)
# finance → 财务处
# research → 科研处
# demo → 教师(通用)
# 留空 → 默认 hr,且会探测主表单已预填的 user 值自动选择。
SMARTCAMPUS_LOGIN_INDUSTRY=

View File

@@ -2,6 +2,40 @@
本文件对应技能市场 **「更新日志」** Tab。
## 1.1.30
- **修复 jiangchang 客户端 arguments 丢失导致 argparse SystemExit 问题**:
- `generate-renewal-task --contract-id` / `dispatch-task-notice --unit-id` / `mark-contract-renewed --contract-id --material-id` / `render-html --task-id` 4 个 sync action 的 `--xxx` 参数从 `required=True` 改为 `default=""`
- CLI handler 检测到空值返回 JSON `{"error": "missing_arg", "detail": ..., "hint": ...}` + exit=1(不再 argparse 报错退出,宿主能拿到结构化错误)
- `assets/actions.json` 4 个 sync action 追加 `v1_1_30_notes` 字段说明新行为
- 1 个新测试 `sync_actions_tolerate_missing_args_v1_30` 覆盖 4 个 action 的 missing_arg JSON
## 1.1.29
- **修复「老 DB 没 download_tasks 表导致入队失败」问题**:
- 用户本地 `data/alert-contract-expiring-hr.db` 是 v1.1.17 之前 init 的(只有前 6 张表),没跑过 v1.1.18 的 init-db → 没 `download_tasks` 表
- `_enqueue_download_tasks_safe` 在 `owns_conn=True` 路径下,先 idempotent `CREATE TABLE IF NOT EXISTS download_tasks ...` + 2 索引,再 INSERT
- 解决 `OperationalError: no such table: download_tasks` + 下载队列 `failed=3`
- 新增测试 `enqueue_safe_auto_create_table_v1_29` 模拟「老 DB 没建过 download_tasks」场景(用 monkeypatch `rp.get_db_path` + `db.connection.get_db_path`),合计 67/67
## 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 整个页面,有账号和密码可用直接点登录跳转」

View File

@@ -1,7 +1,7 @@
---
name: 合同到期智能预警与续聘辅助
description: "当用户说要扫描即将到期的合同、查 90/60/30 天预警名单、给某位教师生成续聘任务书、下发单位版续聘清单、从 SmartCampus HR 平台抓合同数据、解析续聘材料 PDF/Word、查看本月合同到期人员清单时,加载本技能。支持定时扫描(每日凌晨)+ 数据管理 + Agent 三入口,覆盖合同到期智能预警与续聘辅助全流程。"
version: 1.1.26
version: 1.1.30
author: 深圳匠厂科技有限公司
metadata:
openclaw:

View File

@@ -83,7 +83,8 @@
"type": "cli",
"command": "generate-renewal-task",
"args": ["--contract-id", "{row.合同ID}"]
}
},
"v1_1_30_notes": "contract-id 缺失时返回 JSON 提示(missing_arg),不再 argparse 报错。"
},
{
"id": "dispatch-task-notice",
@@ -99,7 +100,8 @@
"type": "cli",
"command": "dispatch-task-notice",
"args": ["--unit-id", "{form.unit_id}", "--month", "{form.month}"]
}
},
"v1_1_30_notes": "unit-id 缺失时返回 JSON 提示(missing_arg),不再 argparse 报错。"
},
{
"id": "mark-contract-renewed",
@@ -116,7 +118,8 @@
"type": "cli",
"command": "mark-contract-renewed",
"args": ["--contract-id", "{row.合同ID}", "--material-id", "{form.material_id}"]
}
},
"v1_1_30_notes": "contract-id/material-id 缺失时返回 JSON 提示(missing_arg),不再 argparse 报错。"
},
{
"id": "render-html",
@@ -128,7 +131,8 @@
"type": "cli",
"command": "render-html",
"args": ["--task-id", "{form.task_id}"]
}
},
"v1_1_30_notes": "task-id 缺失时返回 JSON 提示(missing_arg),不再 argparse 报错。"
},
{
"id": "show-config",

Binary file not shown.

After

Width:  |  Height:  |  Size: 589 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 156 KiB

View File

@@ -127,6 +127,14 @@ def cmd_scan(args: argparse.Namespace) -> int:
def cmd_generate_task(args: argparse.Namespace) -> int:
contract_id = args.contract_id
task_id = args.task_id
# v1.1.30:友好提示 —— 空 contract_id(客户端 arguments 丢失场景)
if not contract_id:
print(json.dumps({
"error": "missing_arg",
"detail": "缺少必填参数 --contract-id。请在调用方传入合同 ID。",
"hint": "可先运行 `collect-from-platform` 抓取合同台账,或在数据管理 Tab 选中一行后再调用。",
}, ensure_ascii=False, indent=2))
return 1
conn = get_connection()
try:
result = task_generator.regenerate_task_for_contract_id(conn, contract_id, task_id)
@@ -141,6 +149,14 @@ def cmd_generate_task(args: argparse.Namespace) -> int:
def cmd_dispatch(args: argparse.Namespace) -> int:
# v1.1.30:友好提示 —— 空 unit_id
if not args.unit_id:
print(json.dumps({
"error": "missing_arg",
"detail": "缺少必填参数 --unit-id。请在调用方传入二级单位 ID。",
"hint": "可参考 contracts 表的 unit_id 字段;或用 --month 限定月份范围。",
}, ensure_ascii=False, indent=2))
return 1
conn = get_connection()
try:
result = dispatcher.dispatch(conn, args.unit_id, args.month or "")
@@ -151,6 +167,17 @@ def cmd_dispatch(args: argparse.Namespace) -> int:
def cmd_mark_renewed(args: argparse.Namespace) -> int:
# v1.1.30:友好提示 —— 空 contract_id 或 material_id
if not args.contract_id or not args.material_id:
print(json.dumps({
"error": "missing_arg",
"detail": (
f"缺少必填参数 {'--contract-id ' if not args.contract_id else ''}"
f"{'--material-id ' if not args.material_id else ''}(由调用方传入)。"
),
"hint": "合同 ID 来自 contracts 表,材料 ID 来自 materials_content 表。",
}, ensure_ascii=False, indent=2))
return 1
conn = get_connection()
try:
from db import repository as repo
@@ -176,6 +203,14 @@ def cmd_mark_renewed(args: argparse.Namespace) -> int:
def cmd_render_html(args: argparse.Namespace) -> int:
# v1.1.30:友好提示 —— 空 task_id
if not args.task_id:
print(json.dumps({
"error": "missing_arg",
"detail": "缺少必填参数 --task-id。请在调用方传入任务 ID。",
"hint": "可先运行 generate-renewal-task 生成任务,再用其 task_id 调用。",
}, ensure_ascii=False, indent=2))
return 1
conn = get_connection()
try:
output_path = html_renderer.render_task_html(conn, args.task_id)
@@ -380,28 +415,30 @@ def build_parser() -> argparse.ArgumentParser:
sp = sub.add_parser(
"generate-renewal-task", help="为指定合同生成续聘任务书",
)
sp.add_argument("--contract-id", required=True, help="合同 ID")
# v1.1.30:去掉 required=True,给 contract_id 默认空串占位
# (jiangchang 客户端偶发 arguments 丢失时空参数不会让 argparse 直接报错退出)
sp.add_argument("--contract-id", default="", help="合同 ID(必填,由调用方传入)")
sp.add_argument("--task-id", default=None, help="可选:指定 task_id(默认自动生成)")
sp.set_defaults(handler=cmd_generate_task)
sp = sub.add_parser(
"dispatch-task-notice", help="按单位/月份派发续聘通知(dry-run)",
)
sp.add_argument("--unit-id", required=True, help="二级单位 ID")
sp.add_argument("--unit-id", default="", help="二级单位 ID(必填,由调用方传入)")
sp.add_argument("--month", default="", help="月份 YYYY-MM(留空 = 全部)")
sp.set_defaults(handler=cmd_dispatch)
sp = sub.add_parser(
"mark-contract-renewed", help="标记合同已续签",
)
sp.add_argument("--contract-id", required=True, help="合同 ID")
sp.add_argument("--material-id", required=True, help="续签材料 ID")
sp.add_argument("--contract-id", default="", help="合同 ID(必填,由调用方传入)")
sp.add_argument("--material-id", default="", help="续签材料 ID(必填,由调用方传入)")
sp.set_defaults(handler=cmd_mark_renewed)
sp = sub.add_parser(
"render-html", help="渲染任务书 HTML",
)
sp.add_argument("--task-id", required=True, help="任务 ID")
sp.add_argument("--task-id", default="", help="任务 ID(必填,由调用方传入)")
sp.set_defaults(handler=cmd_render_html)
sp = sub.add_parser(

View File

@@ -127,6 +127,64 @@ def _safe_getenv_dir(env_key: str, default_rel: str) -> Path:
return (get_data_dir() / default_rel).resolve()
def _get_env(env_key: str, default: str = "") -> str:
"""v1.1.31/32:读 env 值,进程 env 优先,缺则从 .env 文件读(手动解析)。
v1.1.32 关键修复:之前只读 os.environ,但 .env 是用户手动维护的文件,
python 进程默认不会自动加载它,导致 SMARTCAMPUS_LOGIN_USER/PASSWORD 永远读不到
(jiangchang 宿主未必会注入 .env)。手动解析 .env 兜底。
"""
# 1. 进程 env 优先(jiangchang 注入或系统 env 优先)
val = os.environ.get(env_key)
if val is not None and val != "":
return val
# 2. 从 .env 文件手动解析(一次性缓存)
global _ENV_CACHE_LOADED, _ENV_CACHE
if not _ENV_CACHE_LOADED:
_ENV_CACHE_LOADED = True
_ENV_CACHE = _load_env_file()
return _ENV_CACHE.get(env_key, default)
_ENV_CACHE: dict[str, str] = {}
_ENV_CACHE_LOADED = False
def _load_env_file() -> dict[str, str]:
"""手动解析 .env 文件(不依赖 python-dotenv)。
规则:
- 跳过注释(# 开头)和空行
- 形如 `KEY=value`,value 可带 "..." 或 '...' 引号(自动去引号)
- value 含 `=` 时只 split 一次
- 找不到 .env 返回空 dict
"""
from util.runtime_paths import get_env_path
env_path = get_env_path()
if not env_path.exists():
return {}
out: dict[str, str] = {}
try:
for raw in env_path.read_text(encoding="utf-8").splitlines():
line = raw.strip()
if not line or line.startswith("#"):
continue
if "=" not in line:
continue
k, v = line.split("=", 1)
k = k.strip()
v = v.strip()
# 去引号
if len(v) >= 2 and v[0] == v[-1] and v[0] in ("'", '"'):
v = v[1:-1]
out[k] = v
except Exception as e:
logger.warning("platform_collector: 解析 .env 失败(继续走空 dict): %s", e)
return out
def _goto(page, url: str, timeout: int = GOTO_TIMEOUT_MS) -> None:
"""带 networkidle / domcontentloaded 双兜底的 goto。"""
try:
@@ -510,14 +568,29 @@ def _wait_for_url_stable(
def _fill_root_login(page) -> bool:
"""根域登录页(如还在):fill demo/123456 → click Sign in。返回是否执行了登录。"""
"""v1.1.31:根域登录页(如还在):从 .env 读 SMARTCAMPUS_LOGIN_USER /
SMARTCAMPUS_LOGIN_PASSWORD(不硬编码 demo/123456)→ fill → click Sign in。
无 .env 凭证时 return False,调用方走「用户手工登录」分支。
"""
user_input = page.locator('#jc-user').first
if user_input.count() == 0:
return False
user_input.fill("demo")
page.locator('#jc-pass').first.fill("123456")
root_user = _get_env("SMARTCAMPUS_LOGIN_USER", "").strip()
root_pass = _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: 根域自动登录 demo/123456...")
logger.info("platform_collector: 根域自动登录(user 来自 .env)...")
try:
page.wait_for_url(
lambda u: "/home" in u or "/select" in u or "/industries/" in u,
@@ -529,6 +602,224 @@ def _fill_root_login(page) -> bool:
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,让用户手工输入**)。
"""
# 1. 从 .env 读凭证(不硬编码)
spa_user = _get_env("SMARTCAMPUS_LOGIN_USER", "").strip()
spa_pass = _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)
"""
explicit = _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
@@ -743,47 +1034,90 @@ def collect_all(
f"(请用 --show-browser 重试)"
)
# 4. 登录完成 → 等 URL 真正稳定到 /app/ 或 /select(最多 30s,**不做任何操作**)
# v1.1.22:sandbox 登录后会先跳 /select,再自动跳 /app/home,
# 所以两个 URL 都算"SPA 已路由"
if not _wait_for_url_stable(page, ("/app/", "/select"), max_wait_sec=30):
# 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(当前 URL=%s)",
"platform_collector: 登录后 30s 内 URL 未稳定到 /app/、/select、"
"/login 或 /signin(当前 URL=%s)",
page.url,
)
# 不强行 reload —— 让用户决定是否手工刷新
# 4.5. v1.1.24:sandbox 可能登录后把人踢回 SPA 二级登录页
# (/industries/campus/smartcampus/login)。如果 URL 落在
# 这个二级登录页,再走一次 _try_auto_sign_in(已经扩展支持
# 中文「登录」按钮);若 SPA 登录页 input 有预填账号 → 自动点;
# 若 input 空 → 等用户输入并**自己点登录**(脚本不抢按钮)。
# 4.5. v1.1.28/31:URL 落在 SPA 二级登录页 → 主动等 input hydrate(最多 10 秒)
# → 从 .env 读凭证 fill(v1.1.31:密码不硬编码)→ 点登录 → 等 URL 跳离
# v1.1.27 失败的根因:URL 已稳定在 /login 时,React SPA input 可能还没渲染,
# 找不到 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: 登录后仍落在登录页(%s),"
"v1.1.24 触发 SPA 二级登录兜底",
"platform_collector: 登录后落在 SPA 二级登录页(%s),"
"v1.1.31 触发 SPA 登录兜底(等 input hydrate + 从 .env 读凭证 fill)",
page.url,
)
spa_auto = _try_auto_sign_in(page, wait_input_sec=2)
if spa_auto:
# 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: SPA 登录页 10s 内未渲染 input: %s", e,
)
# 2. 从 .env 读凭证 fill + 点登录
spa_filled = _fill_spa_login_from_env(page)
if spa_filled:
logger.info(
"platform_collector: ✅ SPA 二级登录页已自动登录,"
"platform_collector: ✅ SPA 二级登录页(来自 .env)已自动填好并点登录,"
"等 URL 跳到 /app/ 或 /select"
)
_wait_for_url_stable(page, ("/app/", "/select"), max_wait_sec=30)
elif not headless:
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)
# 等 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,.env 凭证可能失效)",
page.url,
)
# 凭证失效:等用户手工登录(headless 模式直接抛错)
if not headless:
_wait_for_manual_login(page, wait_sec=human_wait_timeout)
else:
raise LoginTimeoutError(
f"SPA 二级登录页 .env 凭证 fill 后 URL 不跳离"
f"(可能凭证失效,URL={page.url})"
)
else:
raise LoginTimeoutError(
f"SPA 二级登录页 input 空且 headless 无法手工登录(URL={page.url})"
# .env 无凭证 / input 仍未渲染 → 等用户手工输入
logger.warning(
"platform_collector: SPA 登录页自动 fill 失败"
"(.env 无凭证或 input 未渲染),fallback 等用户输入"
)
if not headless:
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:
raise LoginTimeoutError(
f"SPA 二级登录页 .env 无凭证且 headless 无法手工登录"
f"(URL={page.url})"
)
# 5. v1.1.22:显式跳到 SPA_HOME_URL(/app/home),等侧边栏 link 出现
# 避免 sandbox 在 /select 中间页时直接找 sidebar link 找不到
# v1.1.32 优化:URL 不在 /app/home(可能在 /select 中间页)就 goto(5s 等)
if "/app/home" not in page.url.lower():
logger.info(
"platform_collector: 当前 URL=%s,跳到 SPA_HOME_URL=%s",
@@ -791,10 +1125,10 @@ def collect_all(
)
_goto(page, SPA_HOME_URL)
page.wait_for_timeout(SPA_HYDRATION_MS)
# 等 URL 稳定到 /app/home(最多 15s,**不做任何操作**)
if not _wait_for_url_stable(page, "/app/home", max_wait_sec=15):
# v1.1.32 优化:等 URL 稳定到 /app/home(最多 5s,比 15s 快很多)
if not _wait_for_url_stable(page, "/app/home", max_wait_sec=5):
logger.warning(
"platform_collector: 15s 内未稳定到 /app/home(当前 URL=%s)",
"platform_collector: 5s 内未稳定到 /app/home(当前 URL=%s)",
page.url,
)
@@ -840,6 +1174,15 @@ def collect_all(
" '/industries/campus/smartcampus/app/hr/contracts'"
)
page.wait_for_timeout(PUSHSTATE_SETTLE_MS)
# v1.1.32:pushState 后 URL 可能没切(sandbox mock 偶发),
# 检查 URL 真到 /hr/contracts 才算成功,否则 fallback goto
if "/hr/contracts" not in page.url.lower():
logger.warning(
"platform_collector: pushState 后 URL=%s 未到 /hr/contracts,"
"fallback goto PLATFORM_URL",
page.url,
)
_goto(page, PLATFORM_URL)
except Exception as e:
logger.warning("SPA pushState 失败: %s", e)
_goto(page, PLATFORM_URL)
@@ -859,6 +1202,8 @@ def collect_all(
# 5.6. 兜底刷新:仅当等表格超时后才触发;刷新后可能被踢回 login,需重新登录
# 修复:之前没检查 URL,刷新后停在 login 页继续往后走,找不到表格就 dump 报错
# v1.1.32 优化:刷回 login 时用 _fill_spa_login_from_env(从 .env 读),
# 不再调硬编码 demo/123456 的 _fill_spa_login(与 .env 凭证一致)。
def _try_reload_after_table_timeout():
logger.warning(
"platform_collector: 表格未在 %dms 内出现,触发兜底刷新",
@@ -883,17 +1228,35 @@ def collect_all(
"platform_collector: 刷新后回到登录页(%s),重新登录",
page.url,
)
_goto(page, SPA_LOGIN_URL)
page.wait_for_timeout(POST_RELOAD_LOGIN_HYDRATION_MS)
if not _fill_spa_login(page):
if not headless:
_wait_for_manual_login(page, wait_sec=human_wait_timeout)
# v1.1.32:先等 input hydrate,再走 .env 凭证 fill
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)
except Exception as e:
logger.warning(
"platform_collector: 刷新后 login 页 10s 内未渲染 input: %s", e,
)
if not _fill_spa_login_from_env(page):
# v1.1.32:.env 无凭证时 fallback 到硬编码 demo(保持 v1.1.10 兼容)
logger.warning(
"platform_collector: .env 无凭证,fallback 硬编码 demo/123456"
)
if not _fill_spa_login(page):
if not headless:
_wait_for_manual_login(page, wait_sec=human_wait_timeout)
else:
raise LoginTimeoutError(
"刷新后回到 login 页且 headless 无法手工登录"
)
else:
raise LoginTimeoutError(
"刷新后回到 login 页且 headless 无法手工登录"
)
page.wait_for_timeout(POST_LOGIN_SETTLE_MS)
else:
page.wait_for_timeout(POST_LOGIN_SETTLE_MS)
# .env 凭证 fill 成功 → 等 URL 跳离 login
_wait_for_url_stable(page, ("/app/", "/select"), max_wait_sec=15)
_goto(page, PLATFORM_URL)
page.wait_for_timeout(POST_RELOGIN_GOTO_WAIT_MS)
@@ -1246,6 +1609,35 @@ def _enqueue_download_tasks_safe(
owns_conn = conn is None
if owns_conn:
conn = get_connection()
# v1.1.29:idempotent 表创建 —— 用户没跑 init-db 时,download_tasks 表不存在,
# 首次采集直接建表(CREATE TABLE IF NOT EXISTS)而不是报 "no such table"
try:
conn.executescript(
"""
CREATE TABLE IF NOT EXISTS download_tasks (
task_id TEXT PRIMARY KEY,
contract_id TEXT NOT NULL,
kind TEXT NOT NULL DEFAULT 'platform_button',
status TEXT NOT NULL DEFAULT 'pending',
attempts INTEGER NOT NULL DEFAULT 0,
error_msg TEXT,
enqueued_at TEXT NOT NULL,
started_at TEXT,
finished_at TEXT,
source_path TEXT,
result_path TEXT,
FOREIGN KEY (contract_id) REFERENCES contracts(contract_id)
);
CREATE INDEX IF NOT EXISTS idx_dlq_status ON download_tasks(status);
CREATE INDEX IF NOT EXISTS idx_dlq_contract ON download_tasks(contract_id);
"""
)
conn.commit()
except Exception as ensure_err:
logger.warning(
"platform_collector: ensure download_tasks 表创建失败(继续走入队流程): %s",
ensure_err,
)
try:
for cid in contract_ids:
try:

View File

@@ -6,7 +6,7 @@
"""
SKILL_SLUG = "alert-contract-expiring-hr"
SKILL_VERSION = "1.1.26"
SKILL_VERSION = "1.1.30"
LOG_LOGGER_NAME = "openclaw.skill.alert_contract_expiring_hr"
PLATFORM_KIT_MIN_VERSION = "1.2.11"
@@ -68,16 +68,16 @@ SPA_HOME_RETRY_MAX = 2 # SPA_HOME retry 最大次数
# 表格等待(毫秒)
TABLE_INITIAL_WAIT_MS = 3_000 # 表格初始等待(React + mock 填充)
TABLE_POLL_INTERVAL_MS = 2_000 # 表格轮询间隔
TABLE_POLL_MAX_ATTEMPTS = 20 # 表格轮询最大次数(合计 ~40s)
TABLE_POLL_INTERVAL_MS = 1_500 # v1.1.32 优化:表格轮询间隔(2s → 1.5s)
TABLE_POLL_MAX_ATTEMPTS = 10 # v1.1.32 优化:轮询最大次数(20 → 10,合计 ~15s)
TABLE_TR_TIMEOUT_MS = 3_000 # 单次 wait_for_selector 超时
FULL_RENDER_WAIT_MS = 1_500 # 完整 DOM 渲染后等渲染结束
# 侧边栏/卡片(毫秒)
SIDEBAR_LINK_TIMEOUT_MS = 30_000 # 等侧边栏 hr/contracts link
SIDEBAR_LINK_TIMEOUT_MS = 10_000 # v1.1.32 优化:等侧边栏 link(30s → 10s;sandbox 通常 2-5s)
PUSHSTATE_SETTLE_MS = 1_500 # SPA pushState 后等
SIDEBAR_CLICK_URL_TIMEOUT_MS = 10_000 # click 后等 URL 变 /hr/contracts
URL_AFTER_LOGIN_TIMEOUT_MS = 20_000 # 根域 login 后等跳转
URL_AFTER_LOGIN_TIMEOUT_MS = 10_000 # v1.1.32 优化:根域 login 后等跳转(20s → 10s)
# reload 后备用
RELOAD_NETWORKIDLE_TIMEOUT_MS = 30_000 # page.reload networkidle 超时

View File

@@ -284,6 +284,53 @@ def test_collect_platform_defaults_to_browser() -> tuple[bool, str]:
return True, "默认浏览器模式,--headless 和 --show-browser 行为正确"
def test_sync_actions_tolerate_missing_args_v1_30() -> tuple[bool, str]:
"""v1.1.30:4 个 sync action(generate-renewal-task / dispatch-task-notice /
mark-contract-renewed / render-html)的必填参数全部不传时,
argparse 不应 SystemExit,而是给空字符串默认值;handler 检测到空值后返回
JSON {error: missing_arg, hint: ...} 友好提示,**不再 argparse 错误退出**。
解决问题:jiangchang 客户端 arguments 丢失时调用方收到 argparse 错误而误判失败。
"""
from cli.app import build_parser
parser = build_parser()
# 1. 无参调用不应 argparse SystemExit(注意:之前是 required=True 会抛错)
try:
for cmd in (
"generate-renewal-task",
"dispatch-task-notice",
"mark-contract-renewed",
"render-html",
):
parser.parse_args([cmd]) # 不传任何参数
except SystemExit as e:
return False, f"argparse 在无参调用时 SystemExit({e.code}),应给默认值"
# 2. handler 检测空值应返回 JSON 提示而非抛异常
import argparse
from cli.app import (
cmd_dispatch, cmd_generate_task, cmd_mark_renewed, cmd_render_html,
)
empty_args = argparse.Namespace(
contract_id="", unit_id="", month="", material_id="", task_id="",
)
try:
rc1 = cmd_generate_task(empty_args)
rc2 = cmd_dispatch(empty_args)
rc3 = cmd_mark_renewed(empty_args)
rc4 = cmd_render_html(empty_args)
except Exception as e:
return False, f"handler 应返回 JSON 提示而非抛异常: {e}"
for name, rc in (("generate", rc1), ("dispatch", rc2), ("mark_renewed", rc3), ("render", rc4)):
if rc != 1:
return False, f"{name} 空参数应返回 exit=1,实际 {rc}"
return True, "4 个 sync action 空参数容忍 OK(exit=1 + JSON 提示)"
# ---------- db.repository:v1.1.2 新增的 sandbox card 解析 ----------
def test_parse_sandbox_card_row_format() -> tuple[bool, str]:
@@ -1271,6 +1318,80 @@ def test_enqueue_download_tasks_safe_empty_list() -> tuple[bool, str]:
return True, "空列表 no-op OK"
def test_enqueue_download_tasks_safe_auto_create_table_v1_29() -> tuple[bool, str]:
"""v1.1.29:用户没跑 init-db 时,首次调用 `_enqueue_download_tasks_safe`
不传 conn(owns_conn=True),函数内部 idempotent 创建 download_tasks 表
而非抛 "no such table"。
"""
import os
import sqlite3
import tempfile
from pathlib import Path
from service.platform_collector import _enqueue_download_tasks_safe
# 1. 临时文件 DB,**只建 contracts 表**(模拟老 DB 没 download_tasks)
with tempfile.TemporaryDirectory() as tmp:
tmp_db = Path(tmp) / "old.db"
# 重定向 JIANGCHANG_DATA_ROOT 或 get_db_path 用临时路径
# 这里用 monkeypatch get_db_path + get_connection
from util import runtime_paths as rp
from db import connection as dbconn
orig_rp_db_path = rp.get_db_path
orig_dbconn_db_path = dbconn.get_db_path
rp.get_db_path = lambda: tmp_db # type: ignore[assignment]
dbconn.get_db_path = lambda: tmp_db # type: ignore[assignment]
try:
# 模拟老 DB:只有 contracts 表(v1.1.17 之前 init 过)
_tmp_conn = sqlite3.connect(str(tmp_db))
try:
_tmp_conn.executescript(
"""
CREATE TABLE contracts (
contract_id TEXT PRIMARY KEY,
teacher_name TEXT, unit_id TEXT, unit_name TEXT,
position TEXT, contract_type TEXT, contract_start TEXT,
contract_end TEXT, trial_period_end TEXT,
status TEXT DEFAULT '可续聘',
last_material_id TEXT, renewed_at TEXT
);
INSERT INTO contracts
(contract_id, teacher_name, unit_id, unit_name,
position, contract_type, contract_start,
contract_end, trial_period_end)
VALUES ('ENQ-CREATE-1', '测试', 'U1', '物理学院',
'实验师', '专任教师聘用合同',
'2024-01-01', '2026-12-31', '2024-04-01');
"""
)
_tmp_conn.commit()
finally:
_tmp_conn.close()
# 2. 不传 conn 调用(owns_conn=True,会走 get_connection → 触发 ensure 建表)
summary = _enqueue_download_tasks_safe(["ENQ-CREATE-1"])
if summary.get("failed", 0) > 0:
return False, f"应自动建表后入队成功,但 failed={summary}"
# 3. 验证表确实被自动创建了
conn = sqlite3.connect(str(tmp_db))
try:
count = conn.execute(
"SELECT COUNT(*) FROM download_tasks"
).fetchone()[0]
finally:
conn.close()
if count != 1:
return False, f"download_tasks 应有 1 行,实际 {count}"
return True, f"ensure 建表 + 入队 OK(enqueued={summary.get('enqueued')})"
finally:
rp.get_db_path = orig_rp_db_path # type: ignore[assignment]
dbconn.get_db_path = orig_dbconn_db_path # type: ignore[assignment]
# ---------- v1.1.19: _wait_for_url_stable + login flow ----------
@@ -1408,6 +1529,10 @@ class _FakeLocator:
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 的路由器。
@@ -1419,18 +1544,22 @@ class _FakeLocatorRouter:
self._selectors = selectors
self._page = page
def _resolve(self, sel: str) -> _FakeLocator:
# v1.1.32:支持「,」分隔的多 selector 联合(任一命中即返回)
# 例:'*:has-text("DEV MODE"), *:text-is("DEV MODE")'
for part in sel.split(","):
part = part.strip()
if part in self._selectors:
return self._selectors[part]
return _FakeLocator(count=0)
def __call__(self, sel: str):
"""page.locator(sel) 形式调用。"""
if sel in self._selectors:
return self._selectors[sel]
return _FakeLocator(count=0)
return self._resolve(sel)
def __getattr__(self, sel: str):
# loc.page 已存在则按 selector 路由
if sel in self._selectors:
return self._selectors[sel]
# fallback:未注册的 selector 都返回空 locator
return _FakeLocator(count=0)
return self._resolve(sel)
class _FakePlaywrightPage:
@@ -1588,6 +1717,305 @@ def test_wait_for_login_inputs_filled_auto_click_v1_26() -> tuple[bool, str]:
pc.HUMAN_PROMPT_INTERVAL = orig_prompt
def test_fill_spa_login_from_env_basic_v1_31() -> tuple[bool, str]:
"""v1.1.31:`_fill_spa_login_from_env` ——从 .env 读 SMARTCAMPUS_LOGIN_USER /
SMARTCAMPUS_LOGIN_PASSWORD fill 再点登录按钮(密码不硬编码到代码)。
"""
import os
from service import platform_collector as pc
orig_u, orig_p = os.environ.get("SMARTCAMPUS_LOGIN_USER"), os.environ.get("SMARTCAMPUS_LOGIN_PASSWORD")
os.environ["SMARTCAMPUS_LOGIN_USER"] = "hr_demo"
os.environ["SMARTCAMPUS_LOGIN_PASSWORD"] = "hr_secret_xxx"
try:
click_log: list[str] = []
page = _FakePlaywrightPage(
url="https://sandbox.jc2009.com/industries/campus/smartcampus/login",
selectors={
"#jc-user": _FakeLocator(count=1, value=""),
"#jc-pass": _FakeLocator(count=1, value=""),
"button:has-text('Sign in')": _FakeLocator(count=1, click_log=click_log),
},
)
if not pc._fill_spa_login_from_env(page):
return False, "SPA login fill(来自 .env)应成功"
if not click_log:
return False, f"应点 Sign in,但 click_log={click_log}"
return True, f"SPA 二级登录页 .env 凭证自动 fill+点 OK(click_log={click_log})"
finally:
if orig_u is None:
os.environ.pop("SMARTCAMPUS_LOGIN_USER", None)
else:
os.environ["SMARTCAMPUS_LOGIN_USER"] = orig_u
if orig_p is None:
os.environ.pop("SMARTCAMPUS_LOGIN_PASSWORD", None)
else:
os.environ["SMARTCAMPUS_LOGIN_PASSWORD"] = orig_p
def test_fill_spa_login_from_env_chinese_button_v1_31() -> tuple[bool, str]:
"""v1.1.31:中文「登录」按钮也能 fill .env 凭证并点。"""
import os
from service import platform_collector as pc
orig_u, orig_p = os.environ.get("SMARTCAMPUS_LOGIN_USER"), os.environ.get("SMARTCAMPUS_LOGIN_PASSWORD")
os.environ["SMARTCAMPUS_LOGIN_USER"] = "hr_demo"
os.environ["SMARTCAMPUS_LOGIN_PASSWORD"] = "hr_secret_xxx"
try:
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),
},
)
if not pc._fill_spa_login_from_env(page):
return False, "SPA login 中文按钮 fill 应成功"
if not click_log:
return False, f"应点 登录,但 click_log={click_log}"
return True, f"SPA 二级登录页中文「登录」按钮 OK(click_log={click_log})"
finally:
if orig_u is None:
os.environ.pop("SMARTCAMPUS_LOGIN_USER", None)
else:
os.environ["SMARTCAMPUS_LOGIN_USER"] = orig_u
if orig_p is None:
os.environ.pop("SMARTCAMPUS_LOGIN_PASSWORD", None)
else:
os.environ["SMARTCAMPUS_LOGIN_PASSWORD"] = orig_p
def test_fill_spa_login_from_env_no_credentials_returns_false_v1_31() -> tuple[bool, str]:
"""v1.1.31:.env 未配置 SMARTCAMPUS_LOGIN_USER / PASSWORD → 返回 False,不自动 fill。
按用户要求「密码不能直接放到代码」,无 .env 凭证时等用户手工输入。
v1.1.32 修复:清空进程 env 同时清空 _ENV_CACHE(_get_env 会从 .env 文件 fallback)。
"""
import os
from service import platform_collector as pc
orig_u = os.environ.pop("SMARTCAMPUS_LOGIN_USER", None)
orig_p = os.environ.pop("SMARTCAMPUS_LOGIN_PASSWORD", None)
# v1.1.32:清空 .env 缓存 + 临时把 .env 路径指向不存在的文件
orig_cache = pc._ENV_CACHE.copy()
orig_cache_loaded = pc._ENV_CACHE_LOADED
orig_get_env_path = pc.get_env_path if hasattr(pc, "get_env_path") else None
from pathlib import Path
pc._ENV_CACHE = {}
pc._ENV_CACHE_LOADED = True
pc.get_env_path = lambda: Path("Z:/nonexistent-venv-test/.env") # type: ignore[assignment]
try:
click_log: list[str] = []
page = _FakePlaywrightPage(
url="https://sandbox.jc2009.com/industries/campus/smartcampus/login",
selectors={
"#jc-user": _FakeLocator(count=1, value=""),
"#jc-pass": _FakeLocator(count=1, value=""),
"button:has-text('Sign in')": _FakeLocator(count=1, click_log=click_log),
},
)
if pc._fill_spa_login_from_env(page):
return False, "无 .env 凭证时不应自动 fill"
if click_log:
return False, f"无 .env 凭证时不应点登录按钮,但 click_log={click_log}"
return True, "无 .env 凭证返回 False 不点登录 OK"
finally:
if orig_u is not None:
os.environ["SMARTCAMPUS_LOGIN_USER"] = orig_u
if orig_p is not None:
os.environ["SMARTCAMPUS_LOGIN_PASSWORD"] = orig_p
pc._ENV_CACHE = orig_cache
pc._ENV_CACHE_LOADED = orig_cache_loaded
if orig_get_env_path is not None:
pc.get_env_path = orig_get_env_path # type: ignore[assignment]
def test_fill_spa_login_from_env_no_input_returns_false_v1_31() -> tuple[bool, str]:
"""v1.1.31:找不到 user input(页面不是 login 页)→ 返回 False。"""
import os
from service import platform_collector as pc
orig_u, orig_p = os.environ.get("SMARTCAMPUS_LOGIN_USER"), os.environ.get("SMARTCAMPUS_LOGIN_PASSWORD")
os.environ["SMARTCAMPUS_LOGIN_USER"] = "hr_demo"
os.environ["SMARTCAMPUS_LOGIN_PASSWORD"] = "hr_secret_xxx"
try:
page = _FakePlaywrightPage(
url="https://sandbox.jc2009.com/industries/campus/smartcampus/app/home",
selectors={},
)
if pc._fill_spa_login_from_env(page):
return False, "无 login input 应返回 False"
return True, "无 login input 返回 False OK"
finally:
if orig_u is None:
os.environ.pop("SMARTCAMPUS_LOGIN_USER", None)
else:
os.environ["SMARTCAMPUS_LOGIN_USER"] = orig_u
if orig_p is None:
os.environ.pop("SMARTCAMPUS_LOGIN_PASSWORD", None)
else:
os.environ["SMARTCAMPUS_LOGIN_PASSWORD"] = orig_p
# ---------- v1.1.32: DEV MODE 面板对应行业 Sign in 点击 ----------
class _CountingLocator(_FakeLocator):
"""v1.1.32:扩展 _FakeLocator,支持 nth() 返回子 locator。"""
def __init__(self, *args, parent_count: int = 0, **kwargs) -> None:
super().__init__(*args, **kwargs)
self._parent_count = parent_count
def nth(self, n: int) -> "_FakeLocator":
return _FakeLocator(
count=1 if n < self._parent_count else 0,
value=self._value,
click_log=self._click_log,
)
def _make_dev_mode_page(
sign_in_count: int = 4,
with_dev_marker: bool = True,
main_form_filled: bool = True,
) -> _FakePlaywrightPage:
"""v1.1.32:构造 DEV MODE 面板的 fake page。
Args:
sign_in_count: Sign in 按钮总数(教师通用=1 + N 个行业 tile)
with_dev_marker: 是否包含 DEV MODE 标志
main_form_filled: 主表单是否已预填 demo(左侧教师登录表单)
"""
selectors: dict[str, _FakeLocator] = {}
if with_dev_marker:
# DEV MODE 面板标志:模拟一个含 'DEV MODE' 文本的元素
selectors["*:has-text('DEV MODE')"] = _FakeLocator(count=1)
# Sign in 按钮(DEV MODE 面板里每个 tile 一个 + 主表单的 Sign in = sign_in_count + 1)
click_log: list[str] = []
selectors["button:has-text('Sign in')"] = _CountingLocator(
count=sign_in_count, parent_count=sign_in_count, click_log=click_log,
)
# 主表单
user_val = "demo" if main_form_filled else ""
selectors["#jc-user"] = _FakeLocator(count=1, value=user_val)
selectors["#jc-pass"] = _FakeLocator(count=1, value="123456" if main_form_filled else "")
return _FakePlaywrightPage(
url="https://sandbox.jc2009.com/industries/campus/smartcampus/login",
selectors=selectors,
)
def test_dev_mode_panel_open_detection_v1_32() -> tuple[bool, str]:
"""v1.1.32:DEV MODE 面板展开(≥2 Sign in + DEV MODE 标志)→ _is_dev_mode_panel_open 返回 True。"""
from service import platform_collector as pc
page = _make_dev_mode_page(sign_in_count=4, with_dev_marker=True)
if not pc._is_dev_mode_panel_open(page):
return False, "DEV MODE 面板展开但 _is_dev_mode_panel_open 返回 False"
return True, "DEV MODE 面板检测 OK"
def test_dev_mode_panel_closed_detection_v1_32() -> tuple[bool, str]:
"""v1.1.32:DEV MODE 面板未展开(<2 Sign in 或无 DEV MODE 标志)→ 返回 False。"""
from service import platform_collector as pc
# 场景 A:只有 1 个 Sign in 按钮(主表单),无 DEV MODE 标志
page_a = _FakePlaywrightPage(
url="https://sandbox.jc2009.com/industries/campus/smartcampus/login",
selectors={
"button:has-text('Sign in')": _FakeLocator(count=1),
"#jc-user": _FakeLocator(count=1, value="demo"),
"#jc-pass": _FakeLocator(count=1, value="123456"),
},
)
if pc._is_dev_mode_panel_open(page_a):
return False, "1 个 Sign in 不应判定为 DEV MODE 展开"
# 场景 B:≥2 Sign in 但无 DEV MODE 标志
page_b = _FakePlaywrightPage(
url="https://sandbox.jc2009.com/industries/campus/smartcampus/login",
selectors={
"button:has-text('Sign in')": _FakeLocator(count=3),
},
)
if pc._is_dev_mode_panel_open(page_b):
return False, "无 DEV MODE 标志不应判定为 DEV MODE 展开"
return True, "DEV MODE 面板关闭检测 OK"
def test_click_dev_mode_sign_in_for_industry_v1_32() -> tuple[bool, str]:
"""v1.1.32:点击 DEV MODE 面板对应行业 tile 的 Sign in。
无 SMARTCAMPUS_LOGIN_INDUSTRY env 时默认 hr → 点第 2 个 Sign in(人事处 tile)。
"""
import os
from service import platform_collector as pc
orig_industry = os.environ.pop("SMARTCAMPUS_LOGIN_INDUSTRY", None)
try:
page = _make_dev_mode_page(sign_in_count=4)
# _click_dev_mode_sign_in_for_industry 不依赖 _is_dev_mode_panel_open
# 这里直接测点击逻辑
if not pc._click_dev_mode_sign_in_for_industry(page, "hr"):
return False, "点 hr tile Sign in 应成功"
# 验证 click 落在 Sign in 按钮上(点击会让 click_log 多 1 条)
all_sign_in = page._selectors["button:has-text('Sign in')"]
if len(all_sign_in._click_log) < 1:
return False, f"应点击 Sign in,但 click_log={all_sign_in._click_log}"
return True, f"DEV MODE hr tile Sign in 点击 OK(click_log={all_sign_in._click_log})"
finally:
if orig_industry is not None:
os.environ["SMARTCAMPUS_LOGIN_INDUSTRY"] = orig_industry
def test_fill_spa_login_from_env_clicks_dev_mode_sign_in_first_v1_32() -> tuple[bool, str]:
"""v1.1.32:DEV MODE 面板展开时,_fill_spa_login_from_env 应先点对应行业 Sign in
再走原 fill + 点主表单 Sign in 流程(让对应行业 demo 凭据填到主表单)。
"""
import os
from service import platform_collector as pc
orig_u = os.environ.get("SMARTCAMPUS_LOGIN_USER")
orig_p = os.environ.get("SMARTCAMPUS_LOGIN_PASSWORD")
orig_industry = os.environ.get("SMARTCAMPUS_LOGIN_INDUSTRY")
os.environ["SMARTCAMPUS_LOGIN_USER"] = "hr_demo"
os.environ["SMARTCAMPUS_LOGIN_PASSWORD"] = "hr_secret_xxx"
os.environ["SMARTCAMPUS_LOGIN_INDUSTRY"] = "finance"
try:
# 主表单 input 留空(DEV MODE 面板存在时,主表单可能没填)
page = _make_dev_mode_page(sign_in_count=4, main_form_filled=False)
if not pc._fill_spa_login_from_env(page):
return False, "_fill_spa_login_from_env 应成功(DEV MODE + .env)"
all_sign_in = page._selectors["button:has-text('Sign in')"]
# 期望点击次数 ≥ 2:1 次 DEV MODE finance tile + 1 次主表单 Sign in
if len(all_sign_in._click_log) < 2:
return False, (
f"应先点 DEV MODE 行业 Sign in 再点主表单 Sign in,"
f"但 click_log={all_sign_in._click_log}(期望 ≥2 次)"
)
return True, (
f"SPA 登录页 DEV MODE + .env fill 流程 OK(click_log={all_sign_in._click_log})"
)
finally:
if orig_u is None:
os.environ.pop("SMARTCAMPUS_LOGIN_USER", None)
else:
os.environ["SMARTCAMPUS_LOGIN_USER"] = orig_u
if orig_p is None:
os.environ.pop("SMARTCAMPUS_LOGIN_PASSWORD", None)
else:
os.environ["SMARTCAMPUS_LOGIN_PASSWORD"] = orig_p
if orig_industry is None:
os.environ.pop("SMARTCAMPUS_LOGIN_INDUSTRY", None)
else:
os.environ["SMARTCAMPUS_LOGIN_INDUSTRY"] = orig_industry
# ---------- v1.1.18: show-config / logs-list ----------
@@ -1818,6 +2246,7 @@ def main() -> int:
_run("constants/skill_md_slug_alignment", test_skill_md_slug_alignment)
_run("assets/data_management_actions", test_data_management_actions)
_run("cli/collect_platform_browser_default", test_collect_platform_defaults_to_browser)
_run("cli/sync_actions_tolerate_missing_args_v1_30", test_sync_actions_tolerate_missing_args_v1_30)
_run("db/parse_sandbox_card_row", test_parse_sandbox_card_row_format)
_run("db/parse_sandbox_card_narrative", test_parse_sandbox_card_narrative_format)
_run("db/parse_sandbox_card_empty_invalid", test_parse_sandbox_card_empty_and_invalid)
@@ -1857,6 +2286,7 @@ 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("download_tasks/enqueue_safe_auto_create_table_v1_29", test_enqueue_download_tasks_safe_auto_create_table_v1_29)
_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)
@@ -1866,6 +2296,14 @@ def main() -> int:
_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_spa_login_from_env_basic_v1_31", test_fill_spa_login_from_env_basic_v1_31)
_run("login_flow/fill_spa_login_from_env_chinese_button_v1_31", test_fill_spa_login_from_env_chinese_button_v1_31)
_run("login_flow/fill_spa_login_from_env_no_credentials_returns_false_v1_31", test_fill_spa_login_from_env_no_credentials_returns_false_v1_31)
_run("login_flow/fill_spa_login_from_env_no_input_returns_false_v1_31", test_fill_spa_login_from_env_no_input_returns_false_v1_31)
_run("login_flow/dev_mode_panel_open_detection_v1_32", test_dev_mode_panel_open_detection_v1_32)
_run("login_flow/dev_mode_panel_closed_detection_v1_32", test_dev_mode_panel_closed_detection_v1_32)
_run("login_flow/click_dev_mode_sign_in_for_industry_v1_32", test_click_dev_mode_sign_in_for_industry_v1_32)
_run("login_flow/fill_spa_login_from_env_clicks_dev_mode_sign_in_first_v1_32", test_fill_spa_login_from_env_clicks_dev_mode_sign_in_first_v1_32)
_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)