Compare commits

...

3 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
10 changed files with 876 additions and 103 deletions

View File

@@ -86,3 +86,20 @@ INPUT_DIR=
# 失败最大重试次数(默认 3) # 失败最大重试次数(默认 3)
DOWNLOAD_QUEUE_MAX_ATTEMPTS= 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,31 @@
本文件对应技能市场 **「更新日志」** Tab。 本文件对应技能市场 **「更新日志」** 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 ## 1.1.27
- **修复「到 SPA 二级登录页卡 30 秒」问题**: - **修复「到 SPA 二级登录页卡 30 秒」问题**:

View File

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

View File

@@ -83,7 +83,8 @@
"type": "cli", "type": "cli",
"command": "generate-renewal-task", "command": "generate-renewal-task",
"args": ["--contract-id", "{row.合同ID}"] "args": ["--contract-id", "{row.合同ID}"]
} },
"v1_1_30_notes": "contract-id 缺失时返回 JSON 提示(missing_arg),不再 argparse 报错。"
}, },
{ {
"id": "dispatch-task-notice", "id": "dispatch-task-notice",
@@ -99,7 +100,8 @@
"type": "cli", "type": "cli",
"command": "dispatch-task-notice", "command": "dispatch-task-notice",
"args": ["--unit-id", "{form.unit_id}", "--month", "{form.month}"] "args": ["--unit-id", "{form.unit_id}", "--month", "{form.month}"]
} },
"v1_1_30_notes": "unit-id 缺失时返回 JSON 提示(missing_arg),不再 argparse 报错。"
}, },
{ {
"id": "mark-contract-renewed", "id": "mark-contract-renewed",
@@ -116,7 +118,8 @@
"type": "cli", "type": "cli",
"command": "mark-contract-renewed", "command": "mark-contract-renewed",
"args": ["--contract-id", "{row.合同ID}", "--material-id", "{form.material_id}"] "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", "id": "render-html",
@@ -128,7 +131,8 @@
"type": "cli", "type": "cli",
"command": "render-html", "command": "render-html",
"args": ["--task-id", "{form.task_id}"] "args": ["--task-id", "{form.task_id}"]
} },
"v1_1_30_notes": "task-id 缺失时返回 JSON 提示(missing_arg),不再 argparse 报错。"
}, },
{ {
"id": "show-config", "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: def cmd_generate_task(args: argparse.Namespace) -> int:
contract_id = args.contract_id contract_id = args.contract_id
task_id = args.task_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() conn = get_connection()
try: try:
result = task_generator.regenerate_task_for_contract_id(conn, contract_id, task_id) 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: 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() conn = get_connection()
try: try:
result = dispatcher.dispatch(conn, args.unit_id, args.month or "") 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: 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() conn = get_connection()
try: try:
from db import repository as repo 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: 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() conn = get_connection()
try: try:
output_path = html_renderer.render_task_html(conn, args.task_id) output_path = html_renderer.render_task_html(conn, args.task_id)
@@ -380,28 +415,30 @@ def build_parser() -> argparse.ArgumentParser:
sp = sub.add_parser( sp = sub.add_parser(
"generate-renewal-task", help="为指定合同生成续聘任务书", "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.add_argument("--task-id", default=None, help="可选:指定 task_id(默认自动生成)")
sp.set_defaults(handler=cmd_generate_task) sp.set_defaults(handler=cmd_generate_task)
sp = sub.add_parser( sp = sub.add_parser(
"dispatch-task-notice", help="按单位/月份派发续聘通知(dry-run)", "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.add_argument("--month", default="", help="月份 YYYY-MM(留空 = 全部)")
sp.set_defaults(handler=cmd_dispatch) sp.set_defaults(handler=cmd_dispatch)
sp = sub.add_parser( sp = sub.add_parser(
"mark-contract-renewed", help="标记合同已续签", "mark-contract-renewed", help="标记合同已续签",
) )
sp.add_argument("--contract-id", required=True, help="合同 ID") sp.add_argument("--contract-id", default="", help="合同 ID(必填,由调用方传入)")
sp.add_argument("--material-id", required=True, help="续签材料 ID") sp.add_argument("--material-id", default="", help="续签材料 ID(必填,由调用方传入)")
sp.set_defaults(handler=cmd_mark_renewed) sp.set_defaults(handler=cmd_mark_renewed)
sp = sub.add_parser( sp = sub.add_parser(
"render-html", help="渲染任务书 HTML", "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.set_defaults(handler=cmd_render_html)
sp = sub.add_parser( 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() 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: def _goto(page, url: str, timeout: int = GOTO_TIMEOUT_MS) -> None:
"""带 networkidle / domcontentloaded 双兜底的 goto。""" """带 networkidle / domcontentloaded 双兜底的 goto。"""
try: try:
@@ -510,14 +568,29 @@ def _wait_for_url_stable(
def _fill_root_login(page) -> bool: 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 user_input = page.locator('#jc-user').first
if user_input.count() == 0: if user_input.count() == 0:
return False 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() page.locator('button[type="submit"]:has-text("Sign in")').first.click()
logger.info("platform_collector: 根域自动登录 demo/123456...") logger.info("platform_collector: 根域自动登录(user 来自 .env)...")
try: try:
page.wait_for_url( page.wait_for_url(
lambda u: "/home" in u or "/select" in u or "/industries/" in u, lambda u: "/home" in u or "/select" in u or "/industries/" in u,
@@ -529,18 +602,46 @@ def _fill_root_login(page) -> bool:
return True return True
def _fill_demo_on_spa_login(page) -> bool: def _fill_spa_login_from_env(page) -> bool:
"""v1.1.27:SPA 二级登录页(/industries/campus/smartcampus/login) """v1.1.31/32:SPA 二级登录页(/industries/campus/smartcampus/login)
input 默认**不预填**(与根域不同),必须主动 fill demo/123456 再点登录。 按用户要求「密码不能直接放到代码」—— 从 .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 优先级: selector 优先级:
- user: `#jc-user` → `input[type='tel']` → `input[autocomplete='username']` - user: `#jc-user` → `input[type='tel']` → `input[autocomplete='username']`
- pass: `#jc-pass` → `input[type='password']` → `input[autocomplete='current-password']` - pass: `#jc-pass` → `input[type='password']` → `input[autocomplete='current-password']`
按钮文本优先级:Sign in → 登录 → Login → signin → submit[type]。 按钮文本优先级:Sign in → 登录 → Login → signin → submit[type]。
返回是否成功点登录按钮。 返回是否成功点登录按钮(**无 .env 凭证时返回 False,让用户手工输入**)。
""" """
# 找 user input # 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 user_loc = None
for sel in ("#jc-user", "input[type='tel']", "input[autocomplete='username']"): for sel in ("#jc-user", "input[type='tel']", "input[autocomplete='username']"):
loc = page.locator(sel).first loc = page.locator(sel).first
@@ -549,12 +650,12 @@ def _fill_demo_on_spa_login(page) -> bool:
break break
if user_loc is None: if user_loc is None:
logger.warning( logger.warning(
"platform_collector._fill_demo_on_spa_login: 找不到 user input " "platform_collector._fill_spa_login_from_env: 找不到 user input "
"(#jc-user / tel / autocomplete=username 全 miss)" "(#jc-user / tel / autocomplete=username 全 miss)"
) )
return False return False
# 找 pass input # 3. 找 pass input
pass_loc = None pass_loc = None
for sel in ("#jc-pass", "input[type='password']", "input[autocomplete='current-password']"): for sel in ("#jc-pass", "input[type='password']", "input[autocomplete='current-password']"):
loc = page.locator(sel).first loc = page.locator(sel).first
@@ -563,17 +664,17 @@ def _fill_demo_on_spa_login(page) -> bool:
break break
if pass_loc is None: if pass_loc is None:
logger.warning( logger.warning(
"platform_collector._fill_demo_on_spa_login: 找不到 pass input" "platform_collector._fill_spa_login_from_env: 找不到 pass input"
) )
return False return False
user_loc.fill("demo") user_loc.fill(spa_user)
pass_loc.fill("123456") pass_loc.fill(spa_pass)
logger.info( logger.info(
"platform_collector: SPA 二级登录页 fill demo/123456 完成" "platform_collector: SPA 二级登录页 fill 完成(user 来自 .env)"
) )
# 找登录按钮 # 4. 找登录按钮
btn = None btn = None
btn_text_used = "" btn_text_used = ""
for txt in ("Sign in", "登录", "Login", "signin"): for txt in ("Sign in", "登录", "Login", "signin"):
@@ -587,18 +688,138 @@ def _fill_demo_on_spa_login(page) -> bool:
btn_text_used = "submit" btn_text_used = "submit"
if btn is None or btn.count() == 0: if btn is None or btn.count() == 0:
logger.warning( logger.warning(
"platform_collector._fill_demo_on_spa_login: 找不到登录按钮" "platform_collector._fill_spa_login_from_env: 找不到登录按钮"
) )
return False return False
logger.info( logger.info(
"platform_collector: SPA 二级登录页点 %s 按钮(demo/123456)", "platform_collector: SPA 二级登录页点 %s 按钮(user 来自 .env)",
btn_text_used, btn_text_used,
) )
btn.click() btn.click()
return True 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: def _click_campus_link(page) -> bool:
"""在 /select 页面点智慧校园 tile,返回是否点中。""" """在 /select 页面点智慧校园 tile,返回是否点中。"""
link = page.locator("a[href*='/industries/campus']").first link = page.locator("a[href*='/industries/campus']").first
@@ -830,25 +1051,59 @@ def collect_all(
page.url, page.url,
) )
# 4.5. v1.1.24/27:URL 落在 SPA 二级登录页 → 自动填 demo/123456 再登录 # 4.5. v1.1.28/31:URL 落在 SPA 二级登录页 → 主动等 input hydrate(最多 10 秒)
# sandbox SPA 登录页 input 默认**不预填**(与根域不同), # → 从 .env 读凭证 fill(v1.1.31:密码不硬编码)→ 点登录 → 等 URL 跳离
# 所以 _try_auto_sign_in 不会自动点 —— 必须显式 fill demo/123456。 # v1.1.27 失败的根因:URL 已稳定在 /login 时,React SPA input 可能还没渲染,
# 找不到 input → False → 卡 180 秒等用户输入。
# 修复:先等 input 真正 hydrate 出来,再 fill+click。
cur_url_after_stable = page.url.lower() cur_url_after_stable = page.url.lower()
if "/login" in cur_url_after_stable or "/signin" in cur_url_after_stable: if "/login" in cur_url_after_stable or "/signin" in cur_url_after_stable:
logger.warning( logger.warning(
"platform_collector: 登录后落在 SPA 二级登录页(%s)," "platform_collector: 登录后落在 SPA 二级登录页(%s),"
"v1.1.27 触发 demo/123456 自动 fill", "v1.1.31 触发 SPA 登录兜底(等 input hydrate + 从 .env 读凭证 fill)",
page.url, page.url,
) )
spa_filled = _fill_demo_on_spa_login(page) # 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: if spa_filled:
logger.info( logger.info(
"platform_collector: ✅ SPA 二级登录页 demo/123456 自动填好," "platform_collector: ✅ SPA 二级登录页(来自 .env)已自动填好并点登录,"
"已点登录,等 URL 跳到 /app/ 或 /select" "等 URL 跳到 /app/ 或 /select"
) )
_wait_for_url_stable(page, ("/app/", "/select"), max_wait_sec=30) # 等 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: else:
# demo fill 失败(input 不存在)→ 等用户手工输入 raise LoginTimeoutError(
f"SPA 二级登录页 .env 凭证 fill 后 URL 不跳离"
f"(可能凭证失效,URL={page.url})"
)
else:
# .env 无凭证 / input 仍未渲染 → 等用户手工输入
logger.warning(
"platform_collector: SPA 登录页自动 fill 失败"
"(.env 无凭证或 input 未渲染),fallback 等用户输入"
)
if not headless: if not headless:
if _wait_for_login_inputs_filled(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) _wait_for_url_stable(page, ("/app/", "/select"), max_wait_sec=30)
@@ -856,12 +1111,13 @@ def collect_all(
_wait_for_manual_login(page, wait_sec=human_wait_timeout) _wait_for_manual_login(page, wait_sec=human_wait_timeout)
else: else:
raise LoginTimeoutError( raise LoginTimeoutError(
f"SPA 二级登录页 demo fill 失败且 headless 无法手工登录" f"SPA 二级登录页 .env 无凭证且 headless 无法手工登录"
f"(URL={page.url})" f"(URL={page.url})"
) )
# 5. v1.1.22:显式跳到 SPA_HOME_URL(/app/home),等侧边栏 link 出现 # 5. v1.1.22:显式跳到 SPA_HOME_URL(/app/home),等侧边栏 link 出现
# 避免 sandbox 在 /select 中间页时直接找 sidebar link 找不到 # 避免 sandbox 在 /select 中间页时直接找 sidebar link 找不到
# v1.1.32 优化:URL 不在 /app/home(可能在 /select 中间页)就 goto(5s 等)
if "/app/home" not in page.url.lower(): if "/app/home" not in page.url.lower():
logger.info( logger.info(
"platform_collector: 当前 URL=%s,跳到 SPA_HOME_URL=%s", "platform_collector: 当前 URL=%s,跳到 SPA_HOME_URL=%s",
@@ -869,10 +1125,10 @@ def collect_all(
) )
_goto(page, SPA_HOME_URL) _goto(page, SPA_HOME_URL)
page.wait_for_timeout(SPA_HYDRATION_MS) page.wait_for_timeout(SPA_HYDRATION_MS)
# 等 URL 稳定到 /app/home(最多 15s,**不做任何操作**) # v1.1.32 优化:等 URL 稳定到 /app/home(最多 5s,比 15s 快很多)
if not _wait_for_url_stable(page, "/app/home", max_wait_sec=15): if not _wait_for_url_stable(page, "/app/home", max_wait_sec=5):
logger.warning( logger.warning(
"platform_collector: 15s 内未稳定到 /app/home(当前 URL=%s)", "platform_collector: 5s 内未稳定到 /app/home(当前 URL=%s)",
page.url, page.url,
) )
@@ -918,6 +1174,15 @@ def collect_all(
" '/industries/campus/smartcampus/app/hr/contracts'" " '/industries/campus/smartcampus/app/hr/contracts'"
) )
page.wait_for_timeout(PUSHSTATE_SETTLE_MS) 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: except Exception as e:
logger.warning("SPA pushState 失败: %s", e) logger.warning("SPA pushState 失败: %s", e)
_goto(page, PLATFORM_URL) _goto(page, PLATFORM_URL)
@@ -937,6 +1202,8 @@ def collect_all(
# 5.6. 兜底刷新:仅当等表格超时后才触发;刷新后可能被踢回 login,需重新登录 # 5.6. 兜底刷新:仅当等表格超时后才触发;刷新后可能被踢回 login,需重新登录
# 修复:之前没检查 URL,刷新后停在 login 页继续往后走,找不到表格就 dump 报错 # 修复:之前没检查 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(): def _try_reload_after_table_timeout():
logger.warning( logger.warning(
"platform_collector: 表格未在 %dms 内出现,触发兜底刷新", "platform_collector: 表格未在 %dms 内出现,触发兜底刷新",
@@ -961,8 +1228,23 @@ def collect_all(
"platform_collector: 刷新后回到登录页(%s),重新登录", "platform_collector: 刷新后回到登录页(%s),重新登录",
page.url, page.url,
) )
_goto(page, SPA_LOGIN_URL) # v1.1.32:先等 input hydrate,再走 .env 凭证 fill
page.wait_for_timeout(POST_RELOAD_LOGIN_HYDRATION_MS) 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 _fill_spa_login(page):
if not headless: if not headless:
_wait_for_manual_login(page, wait_sec=human_wait_timeout) _wait_for_manual_login(page, wait_sec=human_wait_timeout)
@@ -972,6 +1254,9 @@ def collect_all(
) )
else: else:
page.wait_for_timeout(POST_LOGIN_SETTLE_MS) page.wait_for_timeout(POST_LOGIN_SETTLE_MS)
else:
# .env 凭证 fill 成功 → 等 URL 跳离 login
_wait_for_url_stable(page, ("/app/", "/select"), max_wait_sec=15)
_goto(page, PLATFORM_URL) _goto(page, PLATFORM_URL)
page.wait_for_timeout(POST_RELOGIN_GOTO_WAIT_MS) page.wait_for_timeout(POST_RELOGIN_GOTO_WAIT_MS)
@@ -1324,6 +1609,35 @@ def _enqueue_download_tasks_safe(
owns_conn = conn is None owns_conn = conn is None
if owns_conn: if owns_conn:
conn = get_connection() 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: try:
for cid in contract_ids: for cid in contract_ids:
try: try:

View File

@@ -6,7 +6,7 @@
""" """
SKILL_SLUG = "alert-contract-expiring-hr" SKILL_SLUG = "alert-contract-expiring-hr"
SKILL_VERSION = "1.1.27" SKILL_VERSION = "1.1.30"
LOG_LOGGER_NAME = "openclaw.skill.alert_contract_expiring_hr" LOG_LOGGER_NAME = "openclaw.skill.alert_contract_expiring_hr"
PLATFORM_KIT_MIN_VERSION = "1.2.11" 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_INITIAL_WAIT_MS = 3_000 # 表格初始等待(React + mock 填充)
TABLE_POLL_INTERVAL_MS = 2_000 # 表格轮询间隔 TABLE_POLL_INTERVAL_MS = 1_500 # v1.1.32 优化:表格轮询间隔(2s → 1.5s)
TABLE_POLL_MAX_ATTEMPTS = 20 # 表格轮询最大次数(合计 ~40s) TABLE_POLL_MAX_ATTEMPTS = 10 # v1.1.32 优化:轮询最大次数(20 → 10,合计 ~15s)
TABLE_TR_TIMEOUT_MS = 3_000 # 单次 wait_for_selector 超时 TABLE_TR_TIMEOUT_MS = 3_000 # 单次 wait_for_selector 超时
FULL_RENDER_WAIT_MS = 1_500 # 完整 DOM 渲染后等渲染结束 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 后等 PUSHSTATE_SETTLE_MS = 1_500 # SPA pushState 后等
SIDEBAR_CLICK_URL_TIMEOUT_MS = 10_000 # click 后等 URL 变 /hr/contracts 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 后备用
RELOAD_NETWORKIDLE_TIMEOUT_MS = 30_000 # page.reload networkidle 超时 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 行为正确" 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 解析 ---------- # ---------- db.repository:v1.1.2 新增的 sandbox card 解析 ----------
def test_parse_sandbox_card_row_format() -> tuple[bool, str]: 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" 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 ---------- # ---------- v1.1.19: _wait_for_url_stable + login flow ----------
@@ -1423,18 +1544,22 @@ class _FakeLocatorRouter:
self._selectors = selectors self._selectors = selectors
self._page = page 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): def __call__(self, sel: str):
"""page.locator(sel) 形式调用。""" """page.locator(sel) 形式调用。"""
if sel in self._selectors: return self._resolve(sel)
return self._selectors[sel]
return _FakeLocator(count=0)
def __getattr__(self, sel: str): def __getattr__(self, sel: str):
# loc.page 已存在则按 selector 路由 # loc.page 已存在则按 selector 路由
if sel in self._selectors: return self._resolve(sel)
return self._selectors[sel]
# fallback:未注册的 selector 都返回空 locator
return _FakeLocator(count=0)
class _FakePlaywrightPage: class _FakePlaywrightPage:
@@ -1592,30 +1717,52 @@ def test_wait_for_login_inputs_filled_auto_click_v1_26() -> tuple[bool, str]:
pc.HUMAN_PROMPT_INTERVAL = orig_prompt pc.HUMAN_PROMPT_INTERVAL = orig_prompt
def test_fill_demo_on_spa_login_basic_v1_27() -> tuple[bool, str]: def test_fill_spa_login_from_env_basic_v1_31() -> tuple[bool, str]:
"""v1.1.27:`_fill_demo_on_spa_login` ——SPA 二级登录页 input 默认不预填, """v1.1.31:`_fill_spa_login_from_env` ——从 .env 读 SMARTCAMPUS_LOGIN_USER /
必须主动 fill demo/123456 再点登录按钮。 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] = [] click_log: list[str] = []
page = _FakePlaywrightPage( page = _FakePlaywrightPage(
url="https://sandbox.jc2009.com/industries/campus/smartcampus/login", url="https://sandbox.jc2009.com/industries/campus/smartcampus/login",
selectors={ selectors={
"#jc-user": _FakeLocator(count=1, value=""), # 空 → 要 fill "#jc-user": _FakeLocator(count=1, value=""),
"#jc-pass": _FakeLocator(count=1, value=""), "#jc-pass": _FakeLocator(count=1, value=""),
"button:has-text('Sign in')": _FakeLocator(count=1, click_log=click_log), "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): if not pc._fill_spa_login_from_env(page):
return False, "SPA login fill demo 应成功" return False, "SPA login fill(来自 .env)应成功"
if not click_log: if not click_log:
return False, f"应点 Sign in,但 click_log={click_log}" return False, f"应点 Sign in,但 click_log={click_log}"
return True, f"SPA 二级登录页 demo/123456 自动 fill+点 OK(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_demo_on_spa_login_chinese_button_v1_27() -> tuple[bool, str]: def test_fill_spa_login_from_env_chinese_button_v1_31() -> tuple[bool, str]:
"""v1.1.27:中文「登录」按钮(用户截图版本)也能 fill demo/123456 并点。""" """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] = [] click_log: list[str] = []
page = _FakePlaywrightPage( page = _FakePlaywrightPage(
url="https://sandbox.jc2009.com/industries/campus/smartcampus/login", url="https://sandbox.jc2009.com/industries/campus/smartcampus/login",
@@ -1625,26 +1772,248 @@ def test_fill_demo_on_spa_login_chinese_button_v1_27() -> tuple[bool, str]:
"button:has-text('登录')": _FakeLocator(count=1, click_log=click_log), "button:has-text('登录')": _FakeLocator(count=1, click_log=click_log),
}, },
) )
from service.platform_collector import _fill_demo_on_spa_login if not pc._fill_spa_login_from_env(page):
return False, "SPA login 中文按钮 fill 应成功"
if not _fill_demo_on_spa_login(page):
return False, "SPA login 中文按钮 fill demo 应成功"
if not click_log: if not click_log:
return False, f"应点 登录,但 click_log={click_log}" return False, f"应点 登录,但 click_log={click_log}"
return True, f"SPA 二级登录页中文「登录」按钮 OK(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_demo_on_spa_login_no_input_returns_false_v1_27() -> tuple[bool, str]: def test_fill_spa_login_from_env_no_credentials_returns_false_v1_31() -> tuple[bool, str]:
"""v1.1.27:找不到 user input(页面不是 login 页)→ 返回 False。""" """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( page = _FakePlaywrightPage(
url="https://sandbox.jc2009.com/industries/campus/smartcampus/app/home", url="https://sandbox.jc2009.com/industries/campus/smartcampus/app/home",
selectors={}, selectors={},
) )
from service.platform_collector import _fill_demo_on_spa_login if pc._fill_spa_login_from_env(page):
if _fill_demo_on_spa_login(page):
return False, "无 login input 应返回 False" return False, "无 login input 应返回 False"
return True, "无 login input 返回 False OK" 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 ---------- # ---------- v1.1.18: show-config / logs-list ----------
@@ -1877,6 +2246,7 @@ def main() -> int:
_run("constants/skill_md_slug_alignment", test_skill_md_slug_alignment) _run("constants/skill_md_slug_alignment", test_skill_md_slug_alignment)
_run("assets/data_management_actions", test_data_management_actions) _run("assets/data_management_actions", test_data_management_actions)
_run("cli/collect_platform_browser_default", test_collect_platform_defaults_to_browser) _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_row", test_parse_sandbox_card_row_format)
_run("db/parse_sandbox_card_narrative", test_parse_sandbox_card_narrative_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) _run("db/parse_sandbox_card_empty_invalid", test_parse_sandbox_card_empty_and_invalid)
@@ -1916,6 +2286,7 @@ def main() -> int:
_run("download_tasks/enqueue_safe_idempotent", test_enqueue_download_tasks_safe_idempotent) _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_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_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_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_transition", test_wait_for_url_stable_transition)
_run("login_flow/wait_url_stable_timeout", test_wait_for_url_stable_timeout) _run("login_flow/wait_url_stable_timeout", test_wait_for_url_stable_timeout)
@@ -1925,9 +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_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/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/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_spa_login_from_env_basic_v1_31", test_fill_spa_login_from_env_basic_v1_31)
_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_spa_login_from_env_chinese_button_v1_31", test_fill_spa_login_from_env_chinese_button_v1_31)
_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("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/masks_tokens", test_show_config_masks_tokens)
_run("show_config/skips_empty", test_show_config_skips_empty) _run("show_config/skips_empty", test_show_config_skips_empty)
_run("show_config/skips_comments", test_show_config_skips_comments) _run("show_config/skips_comments", test_show_config_skips_comments)