Compare commits

...

7 Commits

Author SHA1 Message Date
55239fd4dd chore: auto release commit (2026-09-14 17:49:26)
All checks were successful
技能自动化发布 / release (push) Successful in 8s
2026-09-14 17:49:26 +08:00
6437d6e414 v1.1.6: 按 skill-template human_verification 设计模式重构 RPA
All checks were successful
技能自动化发布 / release (push) Successful in 7s
按 skill-template examples/real_browser_rpa 设计模式重构 platform_collector,
把 v1.1.5 修复的硬编码等待时间统一抽到 constants.py,并补上手工登录 fallback
和结构化错误码。

核心改动:
- constants.py 新增 24 个 RPA 常量(HUMAN_WAIT_TIMEOUT=180 / HUMAN_POLL_INTERVAL=2.0 /
  DEFAULT_PAGE_TIMEOUT_MS=15s / GOTO_TIMEOUT_MS=60s / SPA_LOGIN_* / TABLE_* /
  SIDEBAR_LINK_TIMEOUT_MS 等),与 skill-template human_verification 同名常量对齐
- platform_collector 重构:
  * 新增 RPAError / LoginTimeoutError (ERROR:LOGIN_TIMEOUT) /
    MissingBrowserError (ERROR:MISSING_BROWSER) / TableNotFoundError (ERROR:TABLE_NOT_FOUND)
  * 新增 _is_logged_in / _wait_for_manual_login(轮询 + 180s 兜底,与
    skill-template task_rpa.py::_ensure_logged_in 对齐)
  * _fill_spa_login 自动 fill 失败时 → 自动 fallback _wait_for_manual_login
    (仅 --show-browser 模式;headless 模式抛 LoginTimeoutError 退出码 2)
  * 所有 wait_for_timeout / wait_for_selector timeout 替换为常量
  * collect_all() 加 human_wait_timeout 参数(CLI 透传)
  * page.set_default_timeout(DEFAULT_PAGE_TIMEOUT_MS) 覆盖 Playwright 默认 15s
- app.py cmd_collect_platform:
  * 捕获 RPAError → JSON 输出 {error: e.code, detail: ...} + exit 2
  * 新增 --human-wait-timeout N flag
- requirements.txt:删除违规的 playwright>=1.40.0(playwright 由宿主 jiangchang_skill_core
  runtime 提供,文件顶部注释已声明)
- .env.example:新增 HUMAN_WAIT_TIMEOUT / RPA_PAGE_TIMEOUT_MS / RPA_GOTO_TIMEOUT_MS
  覆盖项
- tests/run_tests.py:新增 7 个 platform_collector 纯函数测试
  (_gen_contract_id × 2 / _normalize_date × 2 / _parse_days × 2 / _parse_alert)
- SKILL.md version 1.1.5 → 1.1.6;CHANGELOG.md 加 1.1.6 条目

测试:28/28 通过(21 原有 + 7 新增)
POLICY-RELEASE-001:SKILL.md=1.1.6 == CHANGELOG=1.1.6 == SKILL_VERSION=1.1.6

Co-Authored-By: Claude Code <noreply@anthropic.com>
2026-09-14 17:29:08 +08:00
f38f662316 v1.1.5: platform_collector 修复登录等待太短导致采集不到数据
All checks were successful
技能自动化发布 / release (push) Successful in 8s
问题:sandbox RPA 采集时,给用户登录态的等待时间太短(settle_ms=0 /
input timeout=10s / wait_for_url 20s / click 后 0ms 路由稳定时间),SPA
异步跳转被打断,session 没保住,最终 40s 内找不到表格、fetched_rows=0。

修复(核心 5 处):
- _fill_spa_login:settle_ms 默认 4000ms、input timeout 20s、wait_for_url 30s、
  URL 判断覆盖 /auth/ 中间态
- _goto(SPA_LOGIN_URL) 后 wait 3000ms(SPA hydration)
- 登录按钮 click 后 wait 2000ms(等 token 写入 cookie 再 goto home)
- 步骤 4 改为 retry 循环(最多 2 次)
- 步骤 5.6:刷新后若 URL 在 /login 重新登录一次

健壮性修复(5 处):
- 侧边栏 link 等待 15s → 30s
- reload timeout 20s → 30s
- 「续聘材料包」click 后 wait 1500ms → 3000ms
- 抽出 _dump_page helper,自动 mkdir(parents=True) 修 5 处 dump 父目录缺失
- 移除残留 dump_path / html 引用

版本对齐:
- SKILL.md version 1.1.4 → 1.1.5
- constants.py SKILL_VERSION 1.1.4 → 1.1.5
- CHANGELOG.md 加 1.1.5 条目

测试:21/21 通过

Co-Authored-By: Claude Code <noreply@anthropic.com>
2026-09-14 17:06:28 +08:00
51cfccc400 ci: 补 .github/workflows/release_skill.yaml\n\n从 skill-template 同步:tag push (v*) → 调\nclient-jiangchang/jiangchang-platform-kit/.github/workflows/reusable-release-skill.yaml@main\n走 windows.x86_64 + PYARMOR_REG_B64 secret。\n\n补完后,下一次打 v* tag 即可触发 CI 跑\nencryption / ZIP / marketplace 同步。\n
All checks were successful
技能自动化发布 / release (push) Successful in 5s
2026-09-14 16:27:09 +08:00
f9be73fa17 chore: auto release commit (2026-09-14 16:22:59) 2026-09-14 16:22:59 +08:00
1707b453c8 chore: auto release commit (2026-09-14 16:15:23) 2026-09-14 16:15:23 +08:00
01ba253bfe v1.1.2: CHANGELOG 添加 sandbox 卡片采集条目
Co-Authored-By: Claude Code <noreply@anthropic.com>
2026-09-14 16:03:34 +08:00
9 changed files with 1176 additions and 99 deletions

View File

@@ -51,6 +51,14 @@ MATERIALS_DOWNLOAD_DIR=
# OCR 语言(pytesseract:默认 chi_sim+eng)
OCR_LANG=chi_sim+eng
# ----- v1.1.6:RPA 超时覆盖(与 skill-template human_verification 对齐)-----
# 手工登录等待超时(秒,默认 180 = HUMAN_WAIT_TIMEOUT)
HUMAN_WAIT_TIMEOUT=
# Playwright 默认操作超时(毫秒,默认 15000)
RPA_PAGE_TIMEOUT_MS=
# page.goto 导航超时(毫秒,默认 60000)
RPA_GOTO_TIMEOUT_MS=
# ----- v1.1.1:AI 决策(宿主 LLM) -----
# 宿主 jiangchang_skill_core.llm_manager 调 LLM 时,模型名覆盖(可选)
LLM_MODEL_NAME=

13
.github/workflows/release_skill.yaml vendored Normal file
View File

@@ -0,0 +1,13 @@
name: 技能自动化发布
on:
push:
tags: ["v*"]
jobs:
release:
uses: client-jiangchang/jiangchang-platform-kit/.github/workflows/reusable-release-skill.yaml@main
secrets:
PYARMOR_REG_B64: ${{ secrets.PYARMOR_REG_B64 }}
with:
artifact_platform: windows
pyarmor_platform: windows.x86_64

View File

@@ -2,6 +2,70 @@
本文件对应技能市场 **「更新日志」** Tab。
## 1.1.7
- **RPA 手工登录 fallback 实跑验证(2026-09-14)**
- sandbox 演示账号 `demo/123456` 已失效(多次尝试 + 6 组常见 demo 凭据全部返回「用户名或密码错误」),按 v1.1.6 设计自动切到 `_wait_for_manual_login` 手工登录分支
- 实跑命令:`python scripts/main.py collect-from-platform --show-browser --limit 1 --human-wait-timeout 180`
- 实跑结果:用户在弹出浏览器手工登录后,RPA 第 1 次轮询(≤2s)即检测到 `/app/home` URL,跳过登录分支继续往下走:
- 自动导航侧边栏 hr/contracts → `page.reload()` 触发 mock 数据
- 抓到 3 行合同(钱实验 / 孙助理 / 王明远)
- 首行「续聘材料包」卡片抓 870 字文本 → 4 条业绩记录
- 落库 `contracts` + `materials_content`:`saved_contracts=1, downloaded_materials=1`
- 结论:v1.1.6「自动 fill + 手工 fallback」设计 **已端到端验证**;后续 collect-from-platform 默认带 `--show-browser` 走手工登录入口(避免硬编码失效 demo 凭据)
- 流程记录见 memory `platform-collect-workflow.md`
## 1.1.6
- **RPA 重构(按 skill-template `examples/real_browser_rpa` 设计模式)**
- **超时参数统一抽到 `constants.py`**:新增 `HUMAN_WAIT_TIMEOUT=180` / `HUMAN_POLL_INTERVAL=2.0` / `DEFAULT_PAGE_TIMEOUT_MS=15s` / `GOTO_TIMEOUT_MS=60s` / `SPA_LOGIN_*` / `TABLE_*` / `SIDEBAR_LINK_TIMEOUT_MS` 等 24 个常量(与 skill-template human_verification 同名常量对齐)
- **新增手工登录 fallback**:自动 fill demo/123456 失败时自动切到 `_wait_for_manual_login`(轮询 `_is_logged_in` 每 2s 一次,最多 `HUMAN_WAIT_TIMEOUT` 秒,超时抛 `LoginTimeoutError`),给用户手工登录机会(与 skill-template `task_rpa.py::_ensure_logged_in` 设计对齐)
- **结构化错误码**:`RPAError` 基类 + `LoginTimeoutError`(`ERROR:LOGIN_TIMEOUT`)/ `MissingBrowserError`(`ERROR:MISSING_BROWSER`)/ `TableNotFoundError`(`ERROR:TABLE_NOT_FOUND`),供 CLI/宿主解析
- **CLI 新增 `--human-wait-timeout N` flag**
- **`_goto` 默认 timeout 改用 `GOTO_TIMEOUT_MS=60s`**;新增 `page.set_default_timeout(DEFAULT_PAGE_TIMEOUT_MS)`
- **依赖清单修复**:删除 `requirements.txt` 中违规的 `playwright>=1.40.0`(playwright 由宿主 `jiangchang_skill_core` runtime 提供,文件顶部注释已声明)
- **环境变量**:`.env.example` 新增 `HUMAN_WAIT_TIMEOUT` / `RPA_PAGE_TIMEOUT_MS` / `RPA_GOTO_TIMEOUT_MS` 覆盖项
- **保留 v1.1.5 自动 fill 优势**:sandbox demo/123456 已知凭据自动登录(生产环境切换无 demo 凭据时自动 fallback 手工)
## 1.1.5
- **platform_collector RPA 采集修复(登录等待时间太短 → 拿不到数据)**
- `_fill_spa_login`:settle_ms 默认 0 → **4000ms**,等 username input timeout 10s → **20s**,wait_for_url(URL 真的跳离 login)20s → **30s**;URL 判断增加 `/auth/` 路径(覆盖 SPA 中间态)
- `_goto(SPA_LOGIN_URL)` 后:wait 1500ms → **3000ms**(SPA hydration)
- 登录按钮 click 后:新增 `wait_for_timeout(2000)`,等 SPA token 写入 cookie + 路由稳定再 goto home(之前直接打断 pending 跳转)
- 步骤 4 改为 retry 循环(最多 2 次),检查 `/app/` in url 且不含 `/login` 才退出
- **采集流程健壮性修复**
- 步骤 5 侧边栏 link 等待:15s → **30s**(sandbox mock 慢启动不 fallback 到 pushState)
- 步骤 5.5 `page.reload()` timeout:20s → **30s**
- **新增 5.6**:刷新后检查 URL,若被踢回 `/login` 则重新登录一次(之前刷新丢 session 后还继续往后走,导致 40s 找不到表格)
- 「续聘材料包」按钮 click 后:wait 1500ms → **3000ms**(卡片动画慢导致文本抓不全)
- 抽出 `_dump_page(page, path, label)` helper,**自动 `mkdir(parents=True)`**,修复 5 处 dump 路径父目录可能不存在的静默失败
## 1.1.4
- **多平台发布声明**:SKILL.md `metadata.openclaw.brands` 新增 `jiangchang` + `huashang`,发布后同步挂匠厂主站与华商 OEM 市场。CI 会幂等补写 `skill_market_relations` 两行(**只补不删**)。
## 1.1.3
- **版本对齐修复**:v1.1.2 tag(commit `4af2b4f`)发布时 `SKILL.md version` / `scripts/util/constants.py SKILL_VERSION` 未同步升级(仍为 1.1.1),本次发布将 manifest 升至 1.1.3 并与代码现状对齐
- **内容同 v1.1.2**:sandbox 卡片采集功能(详见 1.1.2 条目)
> **说明**:v1.1.2 与 v1.1.3 实质内容相同(均含 sandbox 卡片采集)。v1.1.2 tag 视为已发布但 manifest 错位的历史遗留;v1.1.3 为首次 manifest 版本号与代码对齐的 release tag。
## 1.1.2
- **sandbox 卡片采集**:表格 10 列(工号/姓名/单位/岗位/合同类型/起止日期/剩余天数/预警/建议/材料包按钮),进入 hr/contracts 页面后 `page.reload()` 刷新触发 mock 数据加载
- **「续聘材料包」按钮行为**:不触发浏览器 download event,而是展开页面下方「续聘准备任务书」+「单位清单」卡片(含 checklist + 近三年考核表)。RPA 改为抓卡片 inner_text 存 `materials_content`(parse_method='sandbox_card')
- **卡片文本解析进 performance_records**:新增 `repository.parse_sandbox_card_performance`(双模式正则:行格式 `2024\t称职` + 叙述式 `2024年:等次「称职」`)+ `repository.upsert_performance_record`(等次→分数:优秀=90 称职=75 基本称职=60 不称职=30)
- **contract_id 改用工号**(稳定,不随姓名变化):`_gen_contract_id(emp_id, end_date)`
- **实测**(sandbox 3 合同):
| contract_id | 教师 | 解析 perf 条数 | AI 决策 |
|---|---|---|---|
| RPA-T20190033-20260415 | 钱实验 | 4 | observe |
| RPA-T20200055-20260630 | 孙助理 | 6 | observe |
| RPA-T20180021-20260831 | 王明远 | 6 | observe |
## 1.1.1
- **AI 续聘建议升级**:本地规则版 → LLM 优先 + 规则降级

View File

@@ -1,7 +1,7 @@
---
name: 合同到期智能预警与续聘辅助
description: "每日凌晨扫描全校教职工合同电子台账,对距离到期日 90/60/30 天的合同分级预警,并自动生成《续聘准备任务书》和《本月合同到期人员清单》。"
version: 1.1.1
version: 1.1.7
author: 深圳匠厂科技有限公司
metadata:
openclaw:
@@ -12,6 +12,9 @@ metadata:
developer_ids:
- 12345 # 占位,请换成你的匠厂用户 ID
- 12468
brands: # 可选;未写则发布后仅挂匠厂市场(jiangchang)
- jiangchang # 匠厂主站
- huashang # 华商等 OEM:按目标市场填写,可多选
allowed-tools:
- bash
---

View File

@@ -1,6 +1,6 @@
# 合同到期智能预警与续聘辅助 - 技能特有依赖
# jiangchang-platform-kit 等公共能力由宿主共享 runtime 提供,不要写入此文件
# 公共依赖由宿主共享 runtime 提供(jiangchang-platform-kit、playwright 等);
# 这里只声明技能特有 Python 依赖。
# HTTP 请求
requests>=2.31.0,<3
@@ -10,9 +10,6 @@ python-dateutil>=2.8.0,<3
# HTML 模板渲染(个人版任务书)
jinja2>=3.1.0,<4
# v1.1.1:SmartCampus HR 平台 RPA 采集
playwright>=1.40.0,<2
# v1.1.1:续聘材料包 PDF/Word 文本提取
pypdf>=4.0.0,<5
python-docx>=1.1.0,<2

View File

@@ -21,7 +21,7 @@ from typing import Callable, List, Optional
from db.connection import get_connection, init_db
from db import repository as repo
from service import dispatcher, html_renderer, material_parser, platform_collector, scanner, task_generator
from util.constants import DEFAULT_THRESHOLDS, SKILL_VERSION
from util.constants import DEFAULT_THRESHOLDS, HUMAN_WAIT_TIMEOUT, SKILL_VERSION
from util.logging_setup import get_skill_logger, setup_skill_logging
from util.runtime_paths import (
get_db_path,
@@ -204,12 +204,25 @@ def cmd_collect_platform(args: argparse.Namespace) -> int:
退出码:
0 = 成功(即使抓到 0 行也返回 0,错误信息在 output["error"])
1 = 通用异常
2 = RPA 结构化错误(LoginTimeout / MissingBrowser / TableNotFound 等)
"""
try:
result = platform_collector.collect_all(
limit=args.limit,
headless=not args.show_browser,
human_wait_timeout=args.human_wait_timeout or HUMAN_WAIT_TIMEOUT,
)
except platform_collector.RPAError as e:
# v1.1.6:结构化错误(ERROR:LOGIN_TIMEOUT 等)→ exit 2
print(json.dumps({
"error": e.code,
"detail": str(e),
"fetched_rows": 0,
"saved_contracts": 0,
"downloaded_materials": 0,
"contracts": [],
}, ensure_ascii=False, indent=2))
return 2
except Exception as e:
print(json.dumps({"error": "exception", "detail": str(e)},
ensure_ascii=False, indent=2))
@@ -333,6 +346,8 @@ def build_parser() -> argparse.ArgumentParser:
help="限制抓取行数(0=全量)")
sp.add_argument("--show-browser", action="store_true",
help="显示浏览器(首次手工登录用,默认 headless)")
sp.add_argument("--human-wait-timeout", type=int, default=None,
help="手工登录 fallback 等待秒数(默认 180 = HUMAN_WAIT_TIMEOUT)")
sp.set_defaults(handler=cmd_collect_platform)
sp = sub.add_parser(

View File

@@ -1,23 +1,23 @@
"""Playwright RPA:从 SmartCampus HR 平台抓合同表 + 续聘材料包。
设计要点:
设计要点(v1.1.6,按 skill-template examples/real_browser_rpa 设计模式):
- 使用 launch_persistent_context 复用浏览器登录态,避免密码硬编码
- 首次使用需用户手工登录一次(CLI 默认 --headless false 便于交互)
- 后续使用无头模式(profile 已存 cookie)
- **登录策略**:自动 fill demo/123456(sandbox 已知凭据) + 失败时 fallback 手工登录等待(轮询 180s)
- 抓表格:DOM eval_on_selector_all → JSON
- 抓材料:点 "续聘材料包" 按钮 → expect_download → 保存到本地
- 抓材料:点 "续聘材料包" 按钮 → 抓卡片 inner_text
依赖为可选:playwright 未安装时 import 抛错,上层 CLI 捕获并降级报错。
导航流程(v1.1.1 强化版 — 强制 goto):
1. goto 根域
2. 根域 login(如仍显示表单)→ 等跳到 /home 或 /select
3. 强制 goto /select
4. 在 /select 页面 click 智慧校园 tile → 等 URL 含 /industries/campus
5. 若 URL 含 /login,填 sandbox SPA 登录表单 → 等跳走
6. 强制 goto SPA_HOME
7. click 侧边栏 hr/contracts
8. 等 table tbody tr
3. 强制 goto sandbox SPA login URL
4. SPA login:自动 fill demo/123456,失败 fallback 手工等待(HUMAN_WAIT_TIMEOUT=180s)
5. 强制 goto SPA_HOME
6. click 侧边栏 hr/contracts
7. 等 table tbody tr(最长 ~40s)
"""
from __future__ import annotations
@@ -25,11 +25,38 @@ from __future__ import annotations
import os
import re
import sqlite3
import time
from datetime import datetime
from pathlib import Path
from db import repository as repo
from util.constants import SKILL_SLUG
from util.constants import (
CARD_APPEAR_WAIT_MS,
DEFAULT_PAGE_TIMEOUT_MS,
FULL_RENDER_WAIT_MS,
GOTO_TIMEOUT_MS,
HUMAN_POLL_INTERVAL,
HUMAN_WAIT_TIMEOUT,
POST_LOGIN_SETTLE_MS,
POST_RELOAD_LOGIN_HYDRATION_MS,
POST_RELOGIN_GOTO_WAIT_MS,
PUSHSTATE_SETTLE_MS,
RELOAD_DOMCONTENTLOADED_TIMEOUT_MS,
RELOAD_NETWORKIDLE_TIMEOUT_MS,
SIDEBAR_CLICK_URL_TIMEOUT_MS,
SIDEBAR_LINK_TIMEOUT_MS,
SPA_HOME_RETRY_DELAY_MS,
SPA_HOME_RETRY_MAX,
SPA_HYDRATION_MS,
SPA_LOGIN_INPUT_TIMEOUT_MS,
SPA_LOGIN_SETTLE_MS,
SPA_LOGIN_URL_TIMEOUT_MS,
TABLE_INITIAL_WAIT_MS,
TABLE_POLL_INTERVAL_MS,
TABLE_POLL_MAX_ATTEMPTS,
TABLE_TR_TIMEOUT_MS,
URL_AFTER_LOGIN_TIMEOUT_MS,
)
from util.logging_setup import get_skill_logger
from util.runtime_paths import ensure_dirs
@@ -47,6 +74,47 @@ SPA_HOME_URL = PLATFORM_BASE + "/industries/campus/smartcampus/app/home"
SPA_LOGIN_URL = PLATFORM_BASE + "/industries/campus/smartcampus/login"
# ---------- 错误码(与 skill-template 对齐) ----------
class RPAError(Exception):
"""RPA 流程错误基类。"""
code: str = "RPA_ERROR"
def __init__(self, message: str):
super().__init__(message)
self.message = message
class LoginTimeoutError(RPAError):
"""手工登录超时(HUMAN_WAIT_TIMEOUT 秒内未检测到登录成功)。
错误码:ERROR:LOGIN_TIMEOUT(与 skill-template task_rpa.py 对齐)
"""
code = "ERROR:LOGIN_TIMEOUT"
class MissingBrowserError(RPAError):
"""未检测到 Chrome/Edge。
错误码:ERROR:MISSING_BROWSER(与 skill-template browser_session.py 对齐)
"""
code = "ERROR:MISSING_BROWSER"
class TableNotFoundError(RPAError):
"""40s 内未找到表格。
错误码:ERROR:TABLE_NOT_FOUND
"""
code = "ERROR:TABLE_NOT_FOUND"
# ---------- 工具函数 ----------
def _safe_getenv_dir(env_key: str, default_rel: str) -> Path:
raw = os.environ.get(env_key)
if raw:
@@ -56,7 +124,7 @@ def _safe_getenv_dir(env_key: str, default_rel: str) -> Path:
return (get_data_dir() / default_rel).resolve()
def _goto(page, url: str, timeout: int = 20_000) -> None:
def _goto(page, url: str, timeout: int = GOTO_TIMEOUT_MS) -> None:
"""带 networkidle / domcontentloaded 双兜底的 goto。"""
try:
page.goto(url, wait_until="networkidle", timeout=timeout)
@@ -68,6 +136,61 @@ def _goto(page, url: str, timeout: int = 20_000) -> None:
logger.warning("goto domcontentloaded 也失败 %s: %s", url, e2)
def _dump_page(page, path: Path, label: str) -> None:
"""Dump 当前页面 HTML 到指定路径,自动创建父目录。修复:父目录可能不存在导致静默失败。"""
try:
path.parent.mkdir(parents=True, exist_ok=True)
html = page.content()
path.write_text(html, encoding="utf-8")
logger.info(
"platform_collector: %s dump → %s (%d 字节)",
label, path, len(html),
)
except Exception as e:
logger.warning("dump %s 失败: %s", label, e)
def _is_logged_in(page) -> bool:
"""检测 sandbox SPA 是否已登录(URL 含 /app/ 且不含 /login 或 /signin)。
与 skill-template real_browser_rpa _is_logged_in 概念对齐(这里只看 URL,
因为 sandbox SPA 登录后 URL 稳定跳到 /app/...)。
"""
url_lower = page.url.lower()
return (
"/app/" in url_lower
and "/login" not in url_lower
and "/signin" not in url_lower
)
def _wait_for_manual_login(page, wait_sec: int = HUMAN_WAIT_TIMEOUT) -> bool:
"""手工登录 fallback:轮询 _is_logged_in,每 HUMAN_POLL_INTERVAL 秒一次。
与 skill-template task_rpa.py::_ensure_logged_in 设计对齐:
- 不自动破解,等用户在浏览器手工完成登录
- 屏幕打印 [登录] 提示
- 超时抛 LoginTimeoutError(ERROR:LOGIN_TIMEOUT)
自动 fill demo/123456 失败时调用此函数。
"""
print(
f"[登录] 自动登录失败,请在浏览器中完成登录"
f"(sandbox demo/123456 或真实账号),最多等待 {wait_sec} 秒..."
)
deadline = time.monotonic() + wait_sec
attempt = 0
while time.monotonic() < deadline:
attempt += 1
if _is_logged_in(page):
print(f"[登录] 检测到登录成功(attempt {attempt})")
return True
time.sleep(HUMAN_POLL_INTERVAL)
raise LoginTimeoutError(
f"自动登录失败,{wait_sec}s 内用户在浏览器中也未完成登录"
)
def _fill_root_login(page) -> bool:
"""根域登录页(如还在):fill demo/123456 → click Sign in。返回是否执行了登录。"""
user_input = page.locator('#jc-user').first
@@ -80,7 +203,7 @@ def _fill_root_login(page) -> bool:
try:
page.wait_for_url(
lambda u: "/home" in u or "/select" in u or "/industries/" in u,
timeout=20_000,
timeout=URL_AFTER_LOGIN_TIMEOUT_MS,
)
except Exception as e:
logger.warning("等根域登录跳转超时: %s", e)
@@ -97,20 +220,34 @@ def _click_campus_link(page) -> bool:
link.click()
logger.info("platform_collector: click 智慧校园 链接")
try:
page.wait_for_url(lambda u: "/industries/campus" in u, timeout=15_000)
page.wait_for_url(lambda u: "/industries/campus" in u, timeout=SIDEBAR_LINK_TIMEOUT_MS // 2)
except Exception as e:
logger.warning("等 sandbox 路径超时: %s;当前 URL=%s", e, page.url)
return True
def _fill_spa_login(page) -> bool:
"""sandbox SPA 登录(如还在登录页):fill demo/123456 → click 登录。返回是否执行了登录。"""
if "/login" not in page.url and "/signin" not in page.url:
def _fill_spa_login(page, settle_ms: int = SPA_LOGIN_SETTLE_MS) -> bool:
"""sandbox SPA 自动登录(如还在登录页):fill demo/123456 → click 登录 → 等跳转。
返回是否执行了登录(True = 已点登录按钮且 URL 已跳离 login;调用方可继续后续流程)。
失败时**不抛异常**,由调用方决定是否 fallback 到 _wait_for_manual_login。
修复要点(v1.1.5/v1.1.6 登录时间太短导致采集不到数据):
- 默认 settle_ms = SPA_LOGIN_SETTLE_MS(4000ms),给 SPA 异步渲染 + token 写入 cookie 时间
- 等 username input 的 timeout = SPA_LOGIN_INPUT_TIMEOUT_MS(20s)
- click 按钮后 wait_for_url timeout = SPA_LOGIN_URL_TIMEOUT_MS(30s)
- 多处过渡路径(/login, /signin, /auth/)一并判断
"""
url_lower = page.url.lower()
if "/login" not in url_lower and "/signin" not in url_lower and "/auth/" not in url_lower:
return False
try:
page.wait_for_selector('input[autocomplete="username"]', timeout=10_000)
page.wait_for_selector(
'input[autocomplete="username"]',
timeout=SPA_LOGIN_INPUT_TIMEOUT_MS,
)
except Exception as e:
logger.warning("SPA login input 未出现: %s", e)
logger.warning("SPA login input 未出现: %s;当前 URL=%s", e, page.url)
return False
spa_user = page.locator('input[autocomplete="username"]').first
if spa_user.count() == 0:
@@ -122,19 +259,31 @@ def _fill_spa_login(page) -> bool:
'button:has-text("登录"), button[type="submit"]:has-text("Sign in")'
).first.click()
logger.info("platform_collector: sandbox SPA 自动登录...")
page.wait_for_timeout(settle_ms)
try:
page.wait_for_url(
lambda u: "/login" not in u and "/signin" not in u,
timeout=20_000,
lambda u: "/login" not in u.lower()
and "/signin" not in u.lower()
and "/auth/" not in u.lower(),
timeout=SPA_LOGIN_URL_TIMEOUT_MS,
)
except Exception as e:
logger.warning("SPA 登录超时: %s;当前 URL=%s", e, page.url)
return False
logger.info("platform_collector: SPA 登录后 URL=%s", page.url)
return True
def collect_all(limit: int = 0, headless: bool = True) -> dict:
"""登录 sandbox 平台 → 抓合同表 + 材料 → 落库。"""
def collect_all(
limit: int = 0,
headless: bool = True,
human_wait_timeout: int = HUMAN_WAIT_TIMEOUT,
) -> dict:
"""登录 sandbox 平台 → 抓合同表 + 材料 → 落库。
human_wait_timeout: 手工登录 fallback 等待秒数(默认 HUMAN_WAIT_TIMEOUT=180)。
失败时可能抛出 RPAError(LoginTimeoutError / MissingBrowserError / TableNotFoundError)。
"""
try:
from playwright.sync_api import sync_playwright
except ImportError:
@@ -147,8 +296,9 @@ def collect_all(limit: int = 0, headless: bool = True) -> dict:
download_dir.mkdir(parents=True, exist_ok=True)
logger.info(
"platform_collector: 启动 profile=%s download_dir=%s headless=%s limit=%d",
profile_dir, download_dir, headless, limit,
"platform_collector: 启动 profile=%s download_dir=%s headless=%s limit=%d "
"human_wait_timeout=%d",
profile_dir, download_dir, headless, limit, human_wait_timeout,
)
contracts: list[dict] = []
@@ -164,6 +314,9 @@ def collect_all(limit: int = 0, headless: bool = True) -> dict:
)
try:
page = ctx.new_page()
# v1.1.6:覆盖 Playwright 默认操作超时(之前无 set_default_timeout,导致
# 所有 wait_for_selector 默认 30s,调试时不直观)
page.set_default_timeout(DEFAULT_PAGE_TIMEOUT_MS)
# 1. 根域 goto + 根域登录(可选;root login 经常 demo/123456 失败)
_goto(page, ROOT_URL)
@@ -172,39 +325,62 @@ def collect_all(limit: int = 0, headless: bool = True) -> dict:
# 2. 直接 goto sandbox SPA login(绕过 /select 和 root 跳转)
logger.info("platform_collector: 强制 goto sandbox SPA login, 当前 URL=%s", page.url)
_goto(page, SPA_LOGIN_URL)
# SPA hydration + framework 初始化时间(修复登录时间太短的核心点)
page.wait_for_timeout(SPA_HYDRATION_MS)
# 3. sandbox SPA 登录(demo/123456 已知可用)
page.wait_for_timeout(1500)
_fill_spa_login(page)
# 3. sandbox SPA 自动登录(demo/123456 已知可用)
# 失败 fallback 手工登录等待(轮询 _is_logged_in + 180s 兜底)
if not _fill_spa_login(page):
logger.warning(
"platform_collector: 自动 fill 失败(URL=%s),fallback 手工登录等待",
page.url,
)
if not headless:
_wait_for_manual_login(page, wait_sec=human_wait_timeout)
else:
# headless 模式无法手工登录:抛 LoginTimeoutError 让 CLI 决定
raise LoginTimeoutError(
f"headless 模式下自动 fill 失败,且无法手工登录"
f"(请用 --show-browser 重试)"
)
else:
# 等 SPA 路由稳定(之前这里没有 wait,直接 goto home 会打断 pending 跳转)
page.wait_for_timeout(POST_LOGIN_SETTLE_MS)
# 4. 强制 goto SPA home
if "/app/" not in page.url:
logger.info("platform_collector: 强制 goto SPA_HOME, 当前 URL=%s", page.url)
# 4. 强制 goto SPA home(如果还停在 login,重试登录一次)
for retry in range(SPA_HOME_RETRY_MAX):
if _is_logged_in(page):
break
logger.info(
"platform_collector: 强制 goto SPA_HOME (retry=%d), 当前 URL=%s",
retry, page.url,
)
_goto(page, SPA_HOME_URL)
# SPA home 也可能再次 redirect 到 /login(session 没保住)
page.wait_for_timeout(1500)
page.wait_for_timeout(SPA_HOME_RETRY_DELAY_MS)
if _fill_spa_login(page):
_goto(page, SPA_HOME_URL)
page.wait_for_timeout(POST_LOGIN_SETTLE_MS)
# 4.5. retry 后若仍未登录,headless 模式抛超时
if not _is_logged_in(page):
if headless:
raise LoginTimeoutError(
"SPA_HOME retry 全部失败,URL 仍未离开 login 页"
)
_wait_for_manual_login(page, wait_sec=human_wait_timeout)
# 5. 等 SPA 渲染 + click 侧边栏 hr/contracts(SPA 路由,避免 goto 触发整页 reload 丢 session)
try:
# dump SPA home 一次(用于调试 selector)
page.wait_for_timeout(2000)
try:
html = page.content()
dump_path = download_dir.parent / "spa_home_dump.html"
dump_path.write_text(html, encoding="utf-8")
logger.info(
"platform_collector: SPA home dump → %s (%d 字节)",
dump_path, len(html),
)
except Exception:
pass
page.wait_for_timeout(POST_LOGIN_SETTLE_MS)
_dump_page(page, download_dir.parent / "spa_home_dump.html", "SPA home")
# 等 SPA 真正加载完(侧边栏 link 出现,最长 15s)
# 等 SPA 真正加载完(侧边栏 link 出现,最长 SIDEBAR_LINK_TIMEOUT_MS)
# 修复:之前 15s 太短,sandbox mock 加载慢时直接 fallback 到 pushState → 表格加载不出来
try:
page.wait_for_selector(
"a[href$='/hr/contracts']", timeout=15_000, state="attached"
"a[href$='/hr/contracts']",
timeout=SIDEBAR_LINK_TIMEOUT_MS,
state="attached",
)
except Exception as e:
logger.warning("等侧边栏 hr/contracts link 超时: %s", e)
@@ -215,7 +391,8 @@ def collect_all(limit: int = 0, headless: bool = True) -> dict:
logger.info("platform_collector: click 侧边栏 hr/contracts")
try:
page.wait_for_url(
lambda u: "/hr/contracts" in u, timeout=10_000
lambda u: "/hr/contracts" in u,
timeout=SIDEBAR_CLICK_URL_TIMEOUT_MS,
)
except Exception as e:
logger.warning("等 /hr/contracts URL 超时: %s;当前 URL=%s",
@@ -233,7 +410,7 @@ def collect_all(limit: int = 0, headless: bool = True) -> dict:
"window.dispatchEvent(new PopStateEvent('popstate')),"
" '/industries/campus/smartcampus/app/hr/contracts'"
)
page.wait_for_timeout(1500)
page.wait_for_timeout(PUSHSTATE_SETTLE_MS)
except Exception as e:
logger.warning("SPA pushState 失败: %s", e)
_goto(page, PLATFORM_URL)
@@ -246,68 +423,82 @@ def collect_all(limit: int = 0, headless: bool = True) -> dict:
# 5.5. 进入页面后刷新(SPA 进入新路由时 mock 数据可能未加载,需刷新触发)
try:
logger.info("platform_collector: 进入 hr/contracts 后刷新页面")
page.reload(wait_until="networkidle", timeout=20_000)
page.reload(
wait_until="networkidle",
timeout=RELOAD_NETWORKIDLE_TIMEOUT_MS,
)
except Exception as e:
logger.warning("刷新 networkidle 失败: %s;尝试 domcontentloaded", e)
try:
page.reload(wait_until="domcontentloaded", timeout=20_000)
page.reload(
wait_until="domcontentloaded",
timeout=RELOAD_DOMCONTENTLOADED_TIMEOUT_MS,
)
except Exception as e2:
logger.warning("刷新 domcontentloaded 也失败: %s", e2)
# 7. dump 当前 hr/contracts 页面(用于调试)
try:
html = page.content()
dump_path = download_dir.parent / "hr_contracts_dump.html"
dump_path.write_text(html, encoding="utf-8")
logger.info(
"platform_collector: hr/contracts dump → %s (%d 字节)",
dump_path, len(html),
# 5.6. 刷新后可能被踢回 login(session 丢失),重新登录
# 修复:之前没检查 URL,刷新后停在 login 页继续往后走,找不到表格就 dump 报错
if "/login" in page.url.lower() or "/signin" in page.url.lower():
logger.warning(
"platform_collector: 刷新后回到登录页(%s),重新登录",
page.url,
)
except Exception:
pass
_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)
else:
raise LoginTimeoutError(
"刷新后回到 login 页且 headless 无法手工登录"
)
else:
page.wait_for_timeout(POST_LOGIN_SETTLE_MS)
_goto(page, PLATFORM_URL)
page.wait_for_timeout(POST_RELOGIN_GOTO_WAIT_MS)
# 7. dump 当前 hr/contracts 页面(用于调试)
_dump_page(page, download_dir.parent / "hr_contracts_dump.html", "hr/contracts")
# 8. 等待表格加载(SPA 异步填充 mock 数据,多等几秒)
page.wait_for_timeout(3000) # 给 React 初始化 + mock 数据填充时间
page.wait_for_timeout(TABLE_INITIAL_WAIT_MS) # 给 React 初始化 + mock 数据填充时间
found_table = False
for attempt in range(20):
page.wait_for_timeout(2000)
for attempt in range(TABLE_POLL_MAX_ATTEMPTS):
page.wait_for_timeout(TABLE_POLL_INTERVAL_MS)
cur_url = page.url
try:
page.wait_for_selector("table tbody tr", timeout=3000)
page.wait_for_selector(
"table tbody tr",
timeout=TABLE_TR_TIMEOUT_MS,
)
found_table = True
break
except Exception:
logger.info(
"platform_collector: URL=%s 等待表格加载 (attempt %d/20)...",
cur_url, attempt + 1,
"platform_collector: URL=%s 等待表格加载 (attempt %d/%d)...",
cur_url, attempt + 1, TABLE_POLL_MAX_ATTEMPTS,
)
if not found_table:
try:
html = page.content()
dump_path = download_dir.parent / "sandbox_dump.html"
dump_path.write_text(html, encoding="utf-8")
logger.warning(
"platform_collector: 40s 后未找到表格,已 dump 页面到 %s(%d 字节)",
dump_path, len(html),
)
except Exception as e:
logger.warning("dump 失败: %s", e)
return _error_result("table_not_found_after_40s", 0)
_dump_page(page, download_dir.parent / "sandbox_dump.html", "sandbox")
logger.warning("platform_collector: 表格未找到")
raise TableNotFoundError(
f"{TABLE_POLL_MAX_ATTEMPTS * TABLE_POLL_INTERVAL_MS // 1000}s "
f"内未找到表格 (URL={page.url})"
)
# 表格已找到,等 tbody 完全填充 + dump 一份完整 DOM(用于诊断按钮 selector)
try:
page.wait_for_function(
"document.querySelectorAll('table tbody tr').length > 0",
timeout=10_000,
timeout=DEFAULT_PAGE_TIMEOUT_MS,
)
page.wait_for_timeout(1500)
html = page.content()
full_dump = download_dir.parent / "hr_contracts_full_dump.html"
full_dump.write_text(html, encoding="utf-8")
logger.info(
"platform_collector: 完整 hr/contracts dump → %s (%d 字节)",
full_dump, len(html),
page.wait_for_timeout(FULL_RENDER_WAIT_MS)
_dump_page(
page,
download_dir.parent / "hr_contracts_full_dump.html",
"完整 hr/contracts",
)
except Exception as e:
logger.warning("等完整渲染超时: %s", e)
@@ -376,7 +567,7 @@ def collect_all(limit: int = 0, headless: bool = True) -> dict:
)
btn.click()
# 等卡片出现(页面下方追加 h2「续聘准备任务书」/「单位清单」)
page.wait_for_timeout(1500)
page.wait_for_timeout(CARD_APPEAR_WAIT_MS)
# 抓表格下方所有 card 文本
cards_text = page.locator(
"main .grid > div, main > div > div > div"
@@ -398,12 +589,10 @@ def collect_all(limit: int = 0, headless: bool = True) -> dict:
)
# dump 一次(i==0)方便调试
if i == 0:
html_after = page.content()
click_dump = download_dir.parent / "after_click_dump.html"
click_dump.write_text(html_after, encoding="utf-8")
logger.info(
"platform_collector: 首行 click 后 dump → %s (%d 字节)",
click_dump, len(html_after),
_dump_page(
page,
download_dir.parent / "after_click_dump.html",
"首行 click 后",
)
except Exception as e:
logger.warning("抓卡片失败 %s: %s", contract_id, e)

View File

@@ -6,7 +6,7 @@
"""
SKILL_SLUG = "alert-contract-expiring-hr"
SKILL_VERSION = "1.1.1"
SKILL_VERSION = "1.1.7"
LOG_LOGGER_NAME = "openclaw.skill.alert_contract_expiring_hr"
PLATFORM_KIT_MIN_VERSION = "1.2.11"
@@ -43,3 +43,47 @@ DEFAULT_MATERIALS = [
MORAL_FAIL_THRESHOLD = 60 # 师德 < 60 直接 non_renew
PERFORMANCE_FAIL_THRESHOLD = 60 # 单年考核 < 60 算不合格
PERFORMANCE_RENEW_MIN = 80 # 全部 >= 80 且逐年上升 → renew
# ----- v1.1.6:RPA 超时常量(与 skill-template human_verification.py 对齐)-----
# 手工登录等待(HITL fallback)
HUMAN_WAIT_TIMEOUT = 180 # 手工登录等待超时(秒)—— skill-template 同名常量
HUMAN_POLL_INTERVAL = 2.0 # 轮询间隔(秒)—— skill-template 同名常量
# Playwright 操作超时(毫秒)
DEFAULT_PAGE_TIMEOUT_MS = 15_000 # Playwright 默认操作超时 —— skill-template 同名常量
GOTO_TIMEOUT_MS = 60_000 # page.goto 导航超时 —— skill-template 同名常量
# SPA 登录等待(毫秒)
SPA_LOGIN_SETTLE_MS = 4_000 # login click 后 settle 时间(SPA 写 cookie)
SPA_LOGIN_INPUT_TIMEOUT_MS = 20_000 # 等 username input 出现
SPA_LOGIN_URL_TIMEOUT_MS = 30_000 # 等 URL 跳离 login 页
SPA_HYDRATION_MS = 3_000 # goto login URL 后 SPA 框架初始化
POST_LOGIN_SETTLE_MS = 2_000 # 登录后等 SPA 路由稳定
SPA_HOME_RETRY_DELAY_MS = 2_500 # SPA_HOME retry 间隔
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_TR_TIMEOUT_MS = 3_000 # 单次 wait_for_selector 超时
FULL_RENDER_WAIT_MS = 1_500 # 完整 DOM 渲染后等渲染结束
# 侧边栏/卡片(毫秒)
SIDEBAR_LINK_TIMEOUT_MS = 30_000 # 等侧边栏 hr/contracts link
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 后等跳转
# reload 后备用
RELOAD_NETWORKIDLE_TIMEOUT_MS = 30_000 # page.reload networkidle 超时
RELOAD_DOMCONTENTLOADED_TIMEOUT_MS = 30_000 # page.reload domcontentloaded 超时
POST_RELOAD_LOGIN_HYDRATION_MS = 3_000 # reload 后回到 login 页的 hydration
POST_RELOGIN_GOTO_WAIT_MS = 2_000 # 重新登录后 goto PLATFORM_URL 等
# 卡片
CARD_APPEAR_WAIT_MS = 3_000 # 「续聘材料包」click 后等卡片出现
# dump
DUMP_PATH_PARENT_MKDIR = True # _dump_page 是否自动创建父目录

744
tests/run_tests.py Normal file
View File

@@ -0,0 +1,744 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""默认测试套件:合同到期智能预警与续聘辅助。
设计原则:
- 仅用标准库(不依赖 pytest / jinja2),CI 与本地均能直接 `python tests/run_tests.py` 跑
- 用 sqlite3 ":memory:" 隔离 DB,不污染仓库 data/ 目录
- 覆盖:util.date_util 纯函数 / util.constants 版本对齐 / repository 新增功能
(parse_sandbox_card_performance + upsert_performance_record)/ scanner 分级 / task_generator
规则版 / html_renderer 降级路径
- 退出码:0 = 全部通过,1 = 有失败
- -v 开关:详细输出每个用例的状态
release.ps1 在打 tag 前会自动跑 `python tests/run_tests.py -v`,任何一个
test_* 返回 (False, msg) 即 abort 发布。
"""
from __future__ import annotations
import argparse
import os
import re
import shutil
import sqlite3
import sys
import tempfile
import traceback
from datetime import date
from pathlib import Path
from types import SimpleNamespace
from typing import Any, Callable
# 让 scripts/ 下的模块可导入
_REPO_ROOT = Path(__file__).resolve().parent.parent
_SCRIPTS_DIR = _REPO_ROOT / "scripts"
sys.path.insert(0, str(_SCRIPTS_DIR))
# ---------- 测试基础设施 ----------
# 测试结果收集器
_RESULTS: list[tuple[str, bool, str]] = []
_VERBOSE = False
def _run(name: str, fn: Callable[[], tuple[bool, str]]) -> None:
"""运行单个 test_*,记录结果。"""
try:
ok, msg = fn()
except Exception as e: # 测试本身抛异常也算失败
tb = traceback.format_exc(limit=3)
ok, msg = False, f"exception: {e}\n{tb}"
_RESULTS.append((name, ok, msg))
if _VERBOSE or not ok:
marker = "PASS" if ok else "FAIL"
print(f" [{marker}] {name}: {msg}", flush=True)
def _make_conn() -> sqlite3.Connection:
"""建一张 in-memory DB,应用所有 DDL + 迁移(不加载 seed)。"""
from db.connection import DDL_STATEMENTS
from db.migrations import apply_all
conn = sqlite3.connect(":memory:")
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA foreign_keys = ON;")
for ddl in DDL_STATEMENTS:
conn.execute(ddl)
apply_all(conn)
conn.commit()
return conn
class _FakeRow:
"""Mock 一行 sqlite3.Row:同时支持下标访问 (r["x"]) 和属性访问 (r.x)。
production 代码里 task_generator.suggest() 用的是 r["moral_teaching_score"]
下标语法,所以 fake row 必须也支持。
"""
__slots__ = ("year", "performance_score", "moral_teaching_score",
"research_score", "teaching_score")
def __init__(self, year: int, score: float, moral: float):
self.year = year
self.performance_score = score
self.moral_teaching_score = moral
self.research_score = score
self.teaching_score = score
def __getitem__(self, key: str) -> Any:
return getattr(self, key)
def _fake_perf(year: int, score: float, moral: float | None = None) -> _FakeRow:
"""构造一个 performance_records-like 行(用于纯函数测试,不入 DB)。"""
m = moral if moral is not None else score
return _FakeRow(year=year, score=score, moral=m)
def _insert_stub_contract(conn: sqlite3.Connection, contract_id: str) -> None:
"""插入一份占位合同(performance_records 的 FK 目标)。"""
conn.execute(
"""INSERT INTO contracts
(contract_id, employee_id, teacher_name, unit_id, unit_name,
position, contract_type, contract_start, contract_end,
trial_period_end, status, last_material_id, renewed_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, '可续聘', NULL, NULL)""",
(contract_id, f"E-{contract_id}", f"占位-{contract_id}", "U-TEST",
"测试系", "讲师", "正式", "2022-01-01", "2099-12-31", "2022-07-01"),
)
conn.commit()
# ---------- util.date_util ----------
def test_parse_iso_date() -> tuple[bool, str]:
from util.date_util import parse_iso_date
cases = [
("2024-03-15", date(2024, 3, 15)),
("2024-03-15T08:30:00", date(2024, 3, 15)),
(" 2024-03-15 ", date(2024, 3, 15)),
(date(2024, 3, 15), date(2024, 3, 15)), # 已为 date
]
for inp, want in cases:
got = parse_iso_date(inp)
if got != want:
return False, f"parse_iso_date({inp!r}) = {got}, want {want}"
return True, "4 cases ok"
def test_days_to() -> tuple[bool, str]:
from util.date_util import days_to
if days_to(date(2024, 3, 15), today=date(2024, 1, 1)) != 74:
return False, "days_to 正向计算错"
if days_to(date(2023, 12, 31), today=date(2024, 1, 1)) != -1:
return False, "days_to 已过期返回负数错"
# 默认 today=今天
if days_to(date(2099, 1, 1)) <= 0:
return False, "days_to 默认 today 应为今天"
return True, "正/负/默认 today 三种场景"
def test_classify_alert() -> tuple[bool, str]:
from util.date_util import classify_alert
t = (90, 60, 30)
cases = [
(100, None), # 超出预警范围
(91, None), # 边界外
(90, "yellow"),
(61, "yellow"),
(60, "orange"),
(31, "orange"),
(30, "red"),
(1, "red"),
(0, "red"),
(-5, "red"), # 已过期仍属 red
]
for days, want in cases:
got = classify_alert(days, t)
if got != want:
return False, f"classify_alert({days}) = {got}, want {want}"
return True, "10 边界 + 越界场景"
def test_month_match() -> tuple[bool, str]:
from util.date_util import month_match
d = date(2024, 3, 15)
if not month_match(d, "2024-03"):
return False, "同月应命中"
if month_match(d, "2024-04"):
return False, "不同月不应命中"
if not month_match(d, ""):
return False, "空 month_str 应命中全部"
if month_match(d, "garbage"):
return False, "非法 month_str 不应命中"
return True, "命中 / 未命中 / 空 / 非法 四种"
# ---------- util.constants 版本对齐(POLICY-RELEASE-001) ----------
def test_skill_md_version_alignment() -> tuple[bool, str]:
"""POLICY-RELEASE-001:SKILL.md version == SKILL_VERSION == CHANGELOG 顶部版本"""
from util.constants import SKILL_VERSION
skill_md = (_REPO_ROOT / "SKILL.md").read_text(encoding="utf-8")
m = re.search(r'(?m)^version:\s*["\']?(\d+\.\d+\.\d+)', skill_md)
if not m:
return False, "SKILL.md 缺顶层 version: X.Y.Z"
md_ver = m.group(1)
if md_ver != SKILL_VERSION:
return False, f"SKILL.md={md_ver} != SKILL_VERSION={SKILL_VERSION}"
# CHANGELOG 顶部应有 ## X.Y.Z
cl = (_REPO_ROOT / "CHANGELOG.md").read_text(encoding="utf-8")
cl_m = re.search(r'(?m)^##\s+(\d+\.\d+\.\d+)', cl)
if not cl_m:
return False, "CHANGELOG.md 缺顶层 ## X.Y.Z"
cl_ver = cl_m.group(1)
if cl_ver != SKILL_VERSION:
return False, f"CHANGELOG 顶部={cl_ver} != SKILL_VERSION={SKILL_VERSION}"
return True, f"SKILL.md={md_ver} == CHANGELOG={cl_ver} == SKILL_VERSION={SKILL_VERSION}"
def test_skill_md_slug_alignment() -> tuple[bool, str]:
"""SKILL.md metadata.openclaw.slug == SKILL_SLUG"""
from util.constants import SKILL_SLUG
skill_md = (_REPO_ROOT / "SKILL.md").read_text(encoding="utf-8")
m = re.search(r'(?m)^\s*slug:\s*["\']?([a-z0-9_-]+)', skill_md)
if not m:
return False, "SKILL.md 缺 metadata.openclaw.slug"
md_slug = m.group(1)
if md_slug != SKILL_SLUG:
return False, f"SKILL.md slug={md_slug} != SKILL_SLUG={SKILL_SLUG}"
return True, f"slug={md_slug}"
# ---------- db.repository:v1.1.2 新增的 sandbox card 解析 ----------
def test_parse_sandbox_card_row_format() -> tuple[bool, str]:
"""v1.1.2:行格式 '2024\\t称职'"""
from db import repository as repo
conn = _make_conn()
_insert_stub_contract(conn, "C001")
card_text = "year\tresult\n2023\t优秀\n2024\t称职\n2025\t基本称职\n"
n = repo.parse_sandbox_card_performance(conn, "C001", card_text)
if n != 3:
return False, f"写入 {n} 条,期望 3"
rows = repo.list_performance_by_contract(conn, "C001")
if len(rows) != 3:
return False, f"DB 行数 {len(rows)},期望 3"
by_year = {r["year"]: r["performance_score"] for r in rows}
expected = {2023: 90.0, 2024: 75.0, 2025: 60.0}
for y, want in expected.items():
if by_year.get(y) != want:
return False, f"{y} 年分数={by_year.get(y)},期望 {want}"
return True, "3 行(优秀/称职/基本称职)解析 + 写入正确"
def test_parse_sandbox_card_narrative_format() -> tuple[bool, str]:
"""v1.1.2:叙述式 '2024年:等次「称职」'"""
from db import repository as repo
conn = _make_conn()
_insert_stub_contract(conn, "C002")
card_text = "考核情况:\n2023年:等次「优秀」。\n2024年:等次「称职」。\n2025年:等次「不称职」。\n"
n = repo.parse_sandbox_card_performance(conn, "C002", card_text)
if n != 3:
return False, f"写入 {n} 条,期望 3"
rows = repo.list_performance_by_contract(conn, "C002")
by_year = {r["year"]: r["performance_score"] for r in rows}
expected = {2023: 90.0, 2024: 75.0, 2025: 30.0}
for y, want in expected.items():
if by_year.get(y) != want:
return False, f"{y} 年分数={by_year.get(y)},期望 {want}"
return True, "3 行(优秀/称职/不称职)叙述式解析正确"
def test_parse_sandbox_card_empty_and_invalid() -> tuple[bool, str]:
"""空字符串 / 无匹配应返回 0,不报错"""
from db import repository as repo
conn = _make_conn()
_insert_stub_contract(conn, "C001")
if repo.parse_sandbox_card_performance(conn, "C001", "") != 0:
return False, "空字符串应返回 0"
if repo.parse_sandbox_card_performance(conn, "C001", "没有任何年份/等次信息") != 0:
return False, "无匹配应返回 0"
rows = repo.list_performance_by_contract(conn, "C001")
if rows:
return False, "无匹配不应写入任何行"
return True, "空 / 无匹配场景"
def test_parse_sandbox_card_mixed_formats() -> tuple[bool, str]:
"""同卡片混用两种格式时按行去重(PK = P-<contract>-<year>)"""
from db import repository as repo
conn = _make_conn()
_insert_stub_contract(conn, "C003")
card_text = (
"2024年:等次「称职」。\n" # 叙述式
"2025\t优秀\n" # 行格式
"2026年:等次「基本称职」\n"
)
n = repo.parse_sandbox_card_performance(conn, "C003", card_text)
if n != 3:
return False, f"混合解析写入 {n} 条,期望 3"
return True, "叙述式 + 行格式混合解析"
def test_upsert_performance_record_grade_mapping() -> tuple[bool, str]:
"""grade → score 映射 + ON CONFLICT upsert"""
from db import repository as repo
conn = _make_conn()
_insert_stub_contract(conn, "C001")
if not repo.upsert_performance_record(conn, "C001", 2024, "优秀"):
return False, "优秀应返回 True"
if not repo.upsert_performance_record(conn, "C001", 2024, "优秀"):
return False, "upsert 已存在应仍返回 True"
if not repo.upsert_performance_record(conn, "C001", 2025, "称职"):
return False, "称职应返回 True"
if not repo.upsert_performance_record(conn, "C001", 2026, "基本称职"):
return False, "基本称职应返回 True"
if not repo.upsert_performance_record(conn, "C001", 2027, "不称职"):
return False, "不称职应返回 True"
if not repo.upsert_performance_record(conn, "C001", 2028, "不合格"):
return False, "不合格应返回 True"
if repo.upsert_performance_record(conn, "C001", 2029, "未知等次"):
return False, "未知等次应返回 False"
rows = repo.list_performance_by_contract(conn, "C001")
# 期望 5 行:2024/2025/2026/2027/2028
# 第二次 (2024, 优秀) 是 upsert 覆盖不新增;2029 未知等次不入库
if len(rows) != 5:
return False, f"DB 行数 {len(rows)},期望 5(upsert 同年不新增;未知等次不入库)"
by_year = {r["year"]: r["performance_score"] for r in rows}
expected = {2024: 90.0, 2025: 75.0, 2026: 60.0, 2027: 30.0, 2028: 30.0}
for y, want in expected.items():
if by_year.get(y) != want:
return False, f"{y} 年分数={by_year.get(y)},期望 {want}"
# 验证 2024 的分数确实被 upsert 覆盖(不是 30/60/75 等其他值)
if by_year[2024] != 90.0:
return False, f"upsert 未保持分数:2024={by_year[2024]},want 90.0"
return True, "grade 映射 + upsert 幂等 + 未知等次返回 False"
# ---------- service.task_generator.suggest 规则版 ----------
def test_suggest_empty_records() -> tuple[bool, str]:
"""无业绩记录 → observe(数据缺失)"""
from service.task_generator import suggest
s, r = suggest([])
if s != "observe":
return False, f"suggest([]) = {s}, want observe"
if "缺失" not in r:
return False, f"reason 应提示缺失,实际 {r!r}"
return True, "observe + '缺失'"
def test_suggest_low_moral() -> tuple[bool, str]:
"""师德 < 60 → non_renew"""
from service.task_generator import suggest
rows = [_fake_perf(2024, 85, moral=50)]
s, r = suggest(rows)
if s != "non_renew":
return False, f"suggest(师德 50) = {s}, want non_renew"
if "师德" not in r:
return False, f"reason 应含「师德」,实际 {r!r}"
return True, "non_renew + '师德'"
def test_suggest_fail_year() -> tuple[bool, str]:
"""任意一年考核 < 60 → observe(复核)"""
from service.task_generator import suggest
rows = [
_fake_perf(2023, 80, moral=85),
_fake_perf(2024, 50, moral=85), # 不及格
_fake_perf(2025, 85, moral=85),
]
s, r = suggest(rows)
if s != "observe":
return False, f"suggest(有不及格年) = {s}, want observe"
if "复核" not in r and "不合格" not in r:
return False, f"reason 应提示不合格/复核,实际 {r!r}"
return True, "observe + '不合格'"
def test_suggest_renew_upward() -> tuple[bool, str]:
"""全部 >= 80 且逐年上升 → renew"""
from service.task_generator import suggest
rows = [
_fake_perf(2023, 80, moral=85),
_fake_perf(2024, 85, moral=85),
_fake_perf(2025, 90, moral=85),
]
s, r = suggest(rows)
if s != "renew":
return False, f"suggest(全部 80+ 上升) = {s}, want renew"
return True, "renew + 上升趋势"
def test_suggest_observe_default() -> tuple[bool, str]:
"""达标但非逐年上升 → observe(常规评审)"""
from service.task_generator import suggest
rows = [
_fake_perf(2023, 85, moral=85),
_fake_perf(2024, 80, moral=85), # 回落
_fake_perf(2025, 85, moral=85),
]
s, r = suggest(rows)
if s != "observe":
return False, f"suggest(回落) = {s}, want observe"
return True, "observe(常规评审)"
# ---------- service.scanner 端到端 ----------
def test_scanner_no_alerts() -> tuple[bool, str]:
"""所有合同都在 90 天外 → 0 预警"""
from service import scanner
conn = _make_conn()
conn.executemany(
"""INSERT INTO contracts
(contract_id, employee_id, teacher_name, unit_id, unit_name,
position, contract_type, contract_start, contract_end,
trial_period_end, status, last_material_id, renewed_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, '可续聘', NULL, NULL)""",
[
("C-FAR-1", "T001", "张三", "U1", "数学系", "教授", "正式",
"2022-01-01", "2099-12-31", "2022-07-01"),
],
)
conn.commit()
result = scanner.scan(conn, today=date(2024, 1, 15))
if result["scanned_count"] != 1:
return False, f"scanned={result['scanned_count']}, want 1"
if result["yellow_count"] + result["orange_count"] + result["red_count"] != 0:
return False, f"应无预警,实测 yellow={result['yellow_count']} orange={result['orange_count']} red={result['red_count']}"
return True, "1 合同全在 90 天外 → 0 预警"
def test_scanner_three_levels() -> tuple[bool, str]:
"""3 份合同分别命中黄/橙/红"""
from service import scanner
conn = _make_conn()
conn.executemany(
"""INSERT INTO contracts
(contract_id, employee_id, teacher_name, unit_id, unit_name,
position, contract_type, contract_start, contract_end,
trial_period_end, status, last_material_id, renewed_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, '可续聘', NULL, NULL)""",
[
("C-YEL", "T001", "黄老师", "U1", "数学系", "教授", "正式",
"2022-01-01", "2024-05-15", "2022-07-01"), # 距 2024-02-15 = 90 天 → yellow
("C-ORA", "T002", "橙老师", "U1", "物理系", "讲师", "正式",
"2022-01-01", "2024-04-15", "2022-07-01"), # 距 2024-02-15 = 60 天 → orange
("C-RED", "T003", "红老师", "U1", "化学系", "副教授", "正式",
"2022-01-01", "2024-03-15", "2022-07-01"), # 距 2024-02-15 = 29 天 → red
],
)
conn.commit()
result = scanner.scan(conn, today=date(2024, 2, 15))
if result["scanned_count"] != 3:
return False, f"scanned={result['scanned_count']}, want 3"
if result["yellow_count"] != 1 or result["orange_count"] != 1 or result["red_count"] != 1:
return False, (f"分级数 yellow={result['yellow_count']} "
f"orange={result['orange_count']} red={result['red_count']}, "
f"want 1/1/1")
# 验证 alerts 表实际写入
alerts = conn.execute("SELECT contract_id, alert_level FROM alerts").fetchall()
by_cid = {a["contract_id"]: a["alert_level"] for a in alerts}
if by_cid.get("C-YEL") != "yellow":
return False, f"C-YEL 写入等级 {by_cid.get('C-YEL')}, want yellow"
if by_cid.get("C-ORA") != "orange":
return False, f"C-ORA 写入等级 {by_cid.get('C-ORA')}, want orange"
if by_cid.get("C-RED") != "red":
return False, f"C-RED 写入等级 {by_cid.get('C-RED')}, want red"
return True, "3 合同 → 1 黄 + 1 橙 + 1 红,alerts 表写入正确"
def test_scanner_skips_non_active_contracts() -> tuple[bool, str]:
"""status != '可续聘' 的合同不应被扫描"""
from service import scanner
conn = _make_conn()
conn.execute(
"""INSERT INTO contracts
(contract_id, employee_id, teacher_name, unit_id, unit_name,
position, contract_type, contract_start, contract_end,
trial_period_end, status, last_material_id, renewed_at)
VALUES ('C-DONE', 'T001', '已续签老师', 'U1', '数学系', '教授', '正式',
'2022-01-01', '2024-03-15', '2022-07-01', '已续签', NULL, NULL)"""
)
conn.commit()
result = scanner.scan(conn, today=date(2024, 2, 15))
if result["scanned_count"] != 0:
return False, f"已续签合同不应被扫描,scanned={result['scanned_count']}"
alerts = conn.execute("SELECT COUNT(*) AS n FROM alerts").fetchone()["n"]
if alerts != 0:
return False, f"应无 alerts 写入,实测 {alerts}"
return True, "已续签合同被跳过"
# ---------- service.html_renderer 降级路径 ----------
def test_html_render_basic() -> tuple[bool, str]:
"""HTML 渲染(无 jinja2 走 str.replace 降级路径)"""
from service import html_renderer
tmpdir = Path(tempfile.mkdtemp(prefix="alert-hr-test-"))
saved = os.environ.get("HTML_OUTPUT_DIR")
os.environ["HTML_OUTPUT_DIR"] = str(tmpdir)
try:
conn = _make_conn()
conn.execute(
"""INSERT INTO contracts
(contract_id, employee_id, teacher_name, unit_id, unit_name,
position, contract_type, contract_start, contract_end,
trial_period_end, status, last_material_id, renewed_at)
VALUES ('C-HTML', 'T100', '渲染测试', 'U9', '测试系', '教授', '正式',
'2022-01-01', '2024-06-30', '2022-07-01', '可续聘', NULL, NULL)"""
)
conn.execute(
"""INSERT INTO alerts
VALUES ('A-HTML', 'C-HTML', 'orange', 60, 'test',
'2024-04-01T00:00:00', 'open')"""
)
conn.execute(
"""INSERT INTO renewal_tasks
VALUES ('T-HTML', 'C-HTML', 'A-HTML', 'renew', '近 3 年考核良好',
'业绩良好', '["材料 A", "材料 B"]',
'2024-04-01T00:00:00', 'pending',
'rules', '')"""
)
conn.commit()
path = html_renderer.render_task_html(conn, "T-HTML")
if not path.exists():
return False, f"输出文件不存在 {path}"
text = path.read_text(encoding="utf-8")
if "渲染测试" not in text:
return False, "HTML 应含教师姓名"
if "续聘准备任务书" not in text:
return False, "HTML 应含 TITLE"
if "橙色" not in text:
return False, "HTML 应含「橙色」预警等级标签"
if "建议续聘" not in text:
return False, "HTML 应含「建议续聘」建议标签"
return True, f"HTML 渲染 OK → {path.name}"
finally:
if saved is None:
os.environ.pop("HTML_OUTPUT_DIR", None)
else:
os.environ["HTML_OUTPUT_DIR"] = saved
shutil.rmtree(tmpdir, ignore_errors=True)
def test_html_render_unknown_task_raises() -> tuple[bool, str]:
"""渲染不存在的 task_id 应抛 ValueError"""
from service import html_renderer
conn = _make_conn()
try:
html_renderer.render_task_html(conn, "T-NONEXISTENT")
except ValueError as e:
return True, f"ValueError OK: {e}"
return False, "应抛 ValueError,实际未抛"
# ---------- v1.1.6:service.platform_collector 纯函数 ----------
def test_gen_contract_id_emp_id() -> tuple[bool, str]:
"""_gen_contract_id 工号派生(v1.1.1 起用工号作主键)"""
from service import platform_collector as pc
cases = [
(("T20190033", "2026-04-15"), "RPA-T20190033-20260415"),
(("T20200055", "2026-06-30"), "RPA-T20200055-20260630"),
(("T20180021", "2026-08-31"), "RPA-T20180021-20260831"),
]
for (emp_id, end_date), expected in cases:
got = pc._gen_contract_id(emp_id, end_date)
if got != expected:
return False, f"({emp_id!r}, {end_date!r}) → {got!r},want {expected!r}"
return True, f"3 工号派生正确"
def test_gen_contract_id_special_chars() -> tuple[bool, str]:
"""_gen_contract_id 工号含特殊字符时正确清洗(保留前 10 字符)"""
from service import platform_collector as pc
# 工号含非 \w 字符 → 清洗;超过 10 字符截断
got = pc._gen_contract_id("T-2019.0033", "2026-04-15")
# 清洗后保留 \w,"T-2019.0033" → "T20190033"(8 字符,未截断)
if "RPA-" not in got or "20260415" not in got:
return False, f"清洗后格式错:{got!r}"
# 长工号 → 截断到 10 字符
got2 = pc._gen_contract_id("ABCDEFGHIJKLMNO", "2026-01-01")
# "ABCDEFGHIJ"(10 字符)
if not got2.startswith("RPA-ABCDEFGHIJ-"):
return False, f"长工号未截断到 10 字符:{got2!r}"
return True, "清洗 + 截断 OK"
def test_normalize_date_iso() -> tuple[bool, str]:
"""_normalize_date ISO 日期直通"""
from service import platform_collector as pc
cases = [
("2026-04-15", "2026-04-15"),
("2024-01-01", "2024-01-01"),
]
for inp, want in cases:
got = pc._normalize_date(inp)
if got != want:
return False, f"{inp!r} → {got!r},want {want!r}"
return True, "ISO 日期直通"
def test_normalize_date_fallback() -> tuple[bool, str]:
"""_normalize_date 非标准格式截前 10 字符"""
from service import platform_collector as pc
# 非标准但带 ISO 前缀
got = pc._normalize_date("2026-04-15T00:00:00")
if got != "2026-04-15":
return False, f"ISO 前缀应保留前 10 字符 → {got!r}"
# 完全非日期 → 截前 10 字符
got2 = pc._normalize_date("20260415X")
if got2 != "20260415X": # 10 字符
return False, f"非日期应截前 10 字符 → {got2!r}"
# 带空白
got3 = pc._normalize_date(" 2026-04-15 ")
if got3 != "2026-04-15":
return False, f"带空白应去除 → {got3!r}"
return True, "非标准格式 fallback OK"
def test_parse_days_positive() -> tuple[bool, str]:
"""_parse_days 正天数"""
from service import platform_collector as pc
cases = [
("30", 30),
("90", 90),
("还剩 45 天", 45),
("0", 0),
]
for inp, want in cases:
got = pc._parse_days(inp)
if got != want:
return False, f"{inp!r} → {got!r},want {want!r}"
return True, "正天数解析 OK"
def test_parse_days_negative() -> tuple[bool, str]:
"""_parse_days 负数 / 兜底"""
from service import platform_collector as pc
# 含负号
if pc._parse_days("-15") != -15:
return False, "负天数未解析"
# 已过期(带 "已过期" 文本)
if pc._parse_days("已过期 7 天") != 7:
return False, "已过期文本未解析数字部分"
# 无法解析
if pc._parse_days("") is not None:
return False, "空字符串应返回 None"
if pc._parse_days("无数字") is not None:
return False, "纯文本应返回 None"
return True, "负数 + 兜底 OK"
def test_parse_alert_red_orange_yellow_unknown() -> tuple[bool, str]:
"""_parse_alert 红/橙/黄/未知四级映射"""
from service import platform_collector as pc
cases = [
("红色预警", "red"),
("红牌", "red"),
("橙色预警", "orange"),
("黄灯", "yellow"),
("黄色预警", "yellow"),
("无", "unknown"),
("", "unknown"),
("blue", "unknown"),
]
for inp, want in cases:
got = pc._parse_alert(inp)
if got != want:
return False, f"{inp!r} → {got!r},want {want!r}"
return True, "四级映射 OK"
# ---------- 入口 ----------
def main() -> int:
parser = argparse.ArgumentParser(description="默认 release 测试套件")
parser.add_argument("-v", "--verbose", action="store_true",
help="详细输出每个用例结果")
args = parser.parse_args()
global _VERBOSE
_VERBOSE = args.verbose
print("=== alert-contract-expiring-hr release tests ===", flush=True)
_run("util/parse_iso_date", test_parse_iso_date)
_run("util/days_to", test_days_to)
_run("util/classify_alert", test_classify_alert)
_run("util/month_match", test_month_match)
_run("constants/skill_md_version_alignment", test_skill_md_version_alignment)
_run("constants/skill_md_slug_alignment", test_skill_md_slug_alignment)
_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)
_run("db/parse_sandbox_card_mixed", test_parse_sandbox_card_mixed_formats)
_run("db/upsert_performance_record", test_upsert_performance_record_grade_mapping)
_run("task_generator/suggest_empty", test_suggest_empty_records)
_run("task_generator/suggest_low_moral", test_suggest_low_moral)
_run("task_generator/suggest_fail_year", test_suggest_fail_year)
_run("task_generator/suggest_renew", test_suggest_renew_upward)
_run("task_generator/suggest_observe", test_suggest_observe_default)
_run("scanner/no_alerts", test_scanner_no_alerts)
_run("scanner/three_levels", test_scanner_three_levels)
_run("scanner/skips_non_active", test_scanner_skips_non_active_contracts)
_run("html_renderer/basic", test_html_render_basic)
_run("html_renderer/unknown_task_raises", test_html_render_unknown_task_raises)
_run("platform_collector/gen_contract_id_emp_id", test_gen_contract_id_emp_id)
_run("platform_collector/gen_contract_id_special", test_gen_contract_id_special_chars)
_run("platform_collector/normalize_date_iso", test_normalize_date_iso)
_run("platform_collector/normalize_date_fallback", test_normalize_date_fallback)
_run("platform_collector/parse_days_positive", test_parse_days_positive)
_run("platform_collector/parse_days_negative", test_parse_days_negative)
_run("platform_collector/parse_alert_levels", test_parse_alert_red_orange_yellow_unknown)
total = len(_RESULTS)
passed = sum(1 for _, ok, _ in _RESULTS if ok)
failed = total - passed
print()
print(f"=== Summary: {passed}/{total} passed, {failed} failed ===", flush=True)
if failed:
print("\nFailures:")
for name, ok, msg in _RESULTS:
if not ok:
print(f" - {name}: {msg}")
return 1
return 0
if __name__ == "__main__":
raise SystemExit(main())