Commit 09cee42c authored by 李文光's avatar 李文光

feat: 招聘规则库+agent只读(Phase2) 与 ADR0003 skill 契约校正

- 规则库: GET/PUT /api/recruiting-rules(登录读/admin写, data/recruiting-rules.json, 不进业务表)
- agent 只读: GET /api/agent/recruiting-rules + GET /api/agent/jobs(jobId/title, 含对内字段, X-Ingest-Token), security 放行
- llm.py: internalNote 纳入 _INTERNAL_ONLY_JOB_KEYS, 对内笔记 MD 不出对外/LLM
- skill 契约按 battle 定稿校正: 入口2=产出定稿包 -> agent 代建 OA 单(不用"同步到招聘系统"话术); OA bill 双轨(对外MD+执行字段进OA, 对内笔记 internalOnly 载荷按 oaBillNo 认领时合并归档 internalNote); jd-sync 改为内部/QA 导入通道
- 后端 96 passed
parent 84e9cff1
......@@ -11,12 +11,14 @@ from backend.app.config import STATIC_ROOT, get_settings
from backend.app.db import SessionLocal, create_all
from backend.app.models import User, utc_now_iso
from backend.app.routers import (
agent,
ai,
auth,
external_claims,
health,
ingest,
pool,
recruiting_rules,
reports,
resume,
state,
......@@ -77,6 +79,8 @@ def create_app() -> FastAPI:
app.include_router(ingest.router, prefix="/api")
app.include_router(pool.router, prefix="/api")
app.include_router(external_claims.router, prefix="/api")
app.include_router(recruiting_rules.router, prefix="/api")
app.include_router(agent.router, prefix="/api")
app.include_router(reports.router, prefix="/api")
@app.get("/", include_in_schema=False)
......
"""agent 只读查询仓储(ADR 0003 / Phase 2)。
供 QwenPaw 等受信外部智能体在会话前读回「在招岗位 / 旧 JD / 对内档案」,
支撑 recruit-grill 的老岗位迭代与入口分流。含对内字段,仅授权调用方可用。
"""
from typing import Any
from sqlalchemy import or_, select
from sqlalchemy.orm import Session
from backend.app.models import Job
from backend.app.repositories.json_utils import parse_json
_MAX_AGENT_JOBS = 20
def search_jobs_for_agent(session: Session, job_id: str = "", title: str = "") -> list[dict[str, Any]]:
"""按 jobId(精确)或 title(包含匹配,跨全部工作区)读岗位 data JSON。"""
job_id = (job_id or "").strip()
title = (title or "").strip()
if not job_id and not title:
return []
query = select(Job)
if job_id:
query = query.where(Job.id == job_id)
if title:
like = f"%{title}%"
query = query.where(or_(Job.title.like(like), Job.department.like(like)))
query = query.order_by(Job.updated_at.desc()).limit(_MAX_AGENT_JOBS)
jobs = []
for row in session.scalars(query):
data = parse_json(row.data, {})
if not data.get("createdAt"):
data = {**data, "createdAt": data.get("created_at") or row.created_at}
jobs.append(data)
return jobs
"""QwenPaw / 企微智能体只读查询(ADR 0003 / Phase 2)。
- GET /api/agent/recruiting-rules 读取公司招聘规则库
- GET /api/agent/jobs 按 jobId / title 读岗位(含对内字段,仅授权调用方)
鉴权:与 /api/ingest、/api/jd/sync-external 相同的 X-Ingest-Token(或 ?token=)。
未配置 INGEST_TOKEN 时退化为登录态/开放模式,适合内网联调。
"""
from typing import Any
from fastapi import APIRouter, Depends, Request
from sqlalchemy.orm import Session
from backend.app.db import get_db
from backend.app.repositories.agent_repository import search_jobs_for_agent
from backend.app.security import verify_ingest_token
from backend.app.services.recruiting_rules import read_rules
router = APIRouter()
@router.get("/agent/recruiting-rules")
def agent_get_recruiting_rules(request: Request) -> dict[str, Any]:
token_error = verify_ingest_token(request)
if token_error:
return token_error
return {"ok": True, "rules": read_rules()}
@router.get("/agent/jobs")
def agent_get_jobs(
request: Request,
db: Session = Depends(get_db),
) -> dict[str, Any]:
token_error = verify_ingest_token(request)
if token_error:
return token_error
job_id = str(request.query_params.get("jobId") or "").strip()
title = str(request.query_params.get("title") or "").strip()
jobs = search_jobs_for_agent(db, job_id=job_id, title=title)
return {"ok": True, "jobs": jobs, "total": len(jobs)}
"""招聘规则库路由(ADR 0003 / Phase 2)。
- GET /api/recruiting-rules 读规则库(任何已登录用户)
- PUT /api/recruiting-rules 整档覆盖(仅 admin)
QwenPaw 智能体走 /api/agent/recruiting-rules(X-Ingest-Token,见 routers/agent.py)。
"""
from typing import Any
from fastapi import APIRouter, Depends, HTTPException
from backend.app.dependencies import CurrentUser, get_current_user, require_admin
from backend.app.services.recruiting_rules import read_rules, write_rules
router = APIRouter()
@router.get("/recruiting-rules")
def get_recruiting_rules(_: CurrentUser = Depends(get_current_user)) -> dict[str, Any]:
return {"ok": True, "rules": read_rules()}
@router.put("/recruiting-rules")
def put_recruiting_rules(
payload: dict[str, Any],
admin: CurrentUser = Depends(require_admin),
) -> dict[str, Any]:
try:
updated = write_rules(payload, admin.username)
except (ValueError, OSError) as exc:
raise HTTPException(status_code=400, detail=f"规则库保存失败: {exc}") from exc
return {"ok": True, "rules": updated}
......@@ -29,7 +29,7 @@ class DeploymentAuthMiddleware(BaseHTTPMiddleware):
if path in _EXEMPT_API_PATHS:
return await call_next(request)
# 配置了 INGEST_TOKEN 时,采集接口交给路由内令牌校验
if path in ("/api/ingest", "/api/jd/sync-external") and settings.ingest_token:
if path in ("/api/ingest", "/api/jd/sync-external", "/api/agent/jobs", "/api/agent/recruiting-rules") and settings.ingest_token:
return await call_next(request)
# 分析报告公网只读(ADR 0002):匿名放行 GET,路由内部校验 report_id + token。
if path.startswith("/api/share/") and request.method == "GET":
......
......@@ -447,7 +447,17 @@ jdVersion, jdStatus, approvalStatus, jd, responsibilities, requirements, mustHav
# 对内字段:用于内部访谈/寻源,严禁随岗位信息发送到对外 LLM 或写进对外 JD。
_INTERNAL_ONLY_JOB_KEYS = ("sensitive", "genderRestriction", "ageRestriction", "internalFeasibility", "verification", "probationPeriod", "probationCriteria", "nonCompete")
_INTERNAL_ONLY_JOB_KEYS = (
"sensitive",
"genderRestriction",
"ageRestriction",
"internalFeasibility",
"verification",
"probationPeriod",
"probationCriteria",
"nonCompete",
"internalNote",
)
def _sanitize_job_for_external(job: dict[str, Any]) -> dict[str, Any]:
......
"""招聘规则库(ADR 0003 / Phase 2):公司级共享的招聘硬规则/术语/待定决策。
首版形态 = 招聘系统 data/recruiting-rules.json 单文件 + 读写 API(轻量,不进业务表、
不进整包 state)。与仓库根 CONTEXT.md(领域词表)语义分离;规则内容属公司内部知识,
禁止进入对外 JD 或 LLM prompt。后期若公司级规则要驱动 AI 初筛,再升级为业务表。
"""
import json
from pathlib import Path
from typing import Any
from backend.app.config import get_settings
RULES_FILENAME = "recruiting-rules.json"
DEFAULT_RULES: dict[str, Any] = {
"hardRules": [],
"glossary": [],
"pendingDecisions": [],
"notes": "",
"updatedAt": "",
"updatedBy": "",
}
def rules_path() -> Path:
settings = get_settings()
settings.data_dir.mkdir(parents=True, exist_ok=True)
return settings.data_dir / RULES_FILENAME
def read_rules() -> dict[str, Any]:
"""读取规则库;文件缺失/损坏时返回默认空档。"""
path = rules_path()
if not path.exists():
return dict(DEFAULT_RULES)
try:
data = json.loads(path.read_text(encoding="utf-8"))
except (json.JSONDecodeError, OSError):
return dict(DEFAULT_RULES)
if not isinstance(data, dict):
return dict(DEFAULT_RULES)
return {**DEFAULT_RULES, **data}
def write_rules(document: dict[str, Any], updated_by: str = "") -> dict[str, Any]:
"""整档覆盖规则库,记录更新人与时间。不可序列化内容抛 ValueError。"""
from backend.app.models import utc_now_iso
merged = {**DEFAULT_RULES, **document, "updatedAt": utc_now_iso(), "updatedBy": updated_by or ""}
# 提前序列化校验,避免把不可序列化内容写进磁盘
json.dumps(merged, ensure_ascii=False)
rules_path().write_text(json.dumps(merged, ensure_ascii=False, indent=2), encoding="utf-8")
return merged
"""招聘规则库 + agent 只读查询(ADR 0003 / Phase 2)测试。
- 规则库:任何登录用户可读;仅 admin 可整档写;数据落 DATA_DIR/recruiting-rules.json。
- agent 只读:X-Ingest-Token 读规则库、按 jobId/title 读岗位(含对内字段,仅授权调用方)。
"""
import pytest
from fastapi.testclient import TestClient
def _make_client(tmp_path, monkeypatch, ingest_token=""):
monkeypatch.setenv("RECRUITMENT_SKIP_ENV_FILES", "1")
monkeypatch.setenv("DATABASE_URL", f"sqlite:///{(tmp_path / 'test-rules.sqlite').as_posix()}")
monkeypatch.setenv("DATA_DIR", str(tmp_path / "data"))
monkeypatch.setenv("FILES_DIR", str(tmp_path / "uploads"))
monkeypatch.setenv("ADMIN_PASSWORD", "admin-secret")
monkeypatch.setenv("ADMIN_USERNAME", "admin")
monkeypatch.setenv("SESSION_SECRET", "test-session-secret")
monkeypatch.setenv("INGEST_TOKEN", ingest_token)
monkeypatch.setenv("LLM_API_KEY", "")
monkeypatch.setenv("DEEPSEEK_API_KEY", "")
monkeypatch.setenv("OPENAI_API_KEY", "")
import backend.app.config as config
import backend.app.db as db
config.get_settings.cache_clear()
db.engine.dispose()
db.settings = config.get_settings()
db.engine = db.create_engine(
db.normalize_database_url(db.settings.database_url),
connect_args={"check_same_thread": False},
future=True,
)
db.SessionLocal.configure(bind=db.engine)
from backend.app.main import create_app
app = create_app()
with TestClient(app) as test_client:
yield test_client
db.engine.dispose()
config.get_settings.cache_clear()
@pytest.fixture()
def auth_client(tmp_path, monkeypatch):
yield from _make_client(tmp_path, monkeypatch, ingest_token="")
@pytest.fixture()
def agent_client(tmp_path, monkeypatch):
yield from _make_client(tmp_path, monkeypatch, ingest_token="agent-secret-token")
def _login(client, username, password):
return client.post("/api/auth/login", json={"username": username, "password": password})
def _headers(token):
return {"Authorization": f"Bearer {token}"}
# ---- 规则库:开放模式默认 + admin 整档写 ----
def test_rules_default_and_write_open_mode(client):
body = client.get("/api/recruiting-rules").json()
assert body["ok"] is True
assert body["rules"]["hardRules"] == []
payload = {
"hardRules": [{"job": "销售岗", "rule": "自带华东客户资源, 不接受无资源转型"}],
"glossary": [{"term": "开荒", "meaning": "从 0 到 1 搭新业务线"}],
"pendingDecisions": ["销售岗签约额指标待定"],
"notes": "内部使用, 不对外",
}
resp = client.put("/api/recruiting-rules", json=payload)
assert resp.status_code == 200
rules = resp.json()["rules"]
assert rules["hardRules"][0]["job"] == "销售岗"
assert rules["updatedBy"] == "local"
assert rules["updatedAt"]
again = client.get("/api/recruiting-rules").json()["rules"]
assert again["glossary"][0]["term"] == "开荒"
assert again["pendingDecisions"][0].startswith("销售岗")
# ---- 规则库:关闭模式 user 只读、admin 可写 ----
def test_rules_admin_write_user_read(auth_client):
admin_token = _login(auth_client, "admin", "admin-secret").json()["token"]
admin_headers = _headers(admin_token)
resp = auth_client.post(
"/api/auth/users",
json={"username": "alice", "name": "Alice", "role": "user", "password": "alice12345"},
headers=admin_headers,
)
assert resp.status_code == 200
alice_headers = _headers(_login(auth_client, "alice", "alice12345").json()["token"])
# user 可读不可写
assert auth_client.get("/api/recruiting-rules", headers=alice_headers).json()["ok"] is True
forbidden = auth_client.put(
"/api/recruiting-rules", json={"hardRules": [{"job": "x", "rule": "y"}]}, headers=alice_headers
)
assert forbidden.status_code == 403
# admin 可写
ok = auth_client.put(
"/api/recruiting-rules", json={"hardRules": [{"job": "技术岗", "rule": "本科及以上"}]}, headers=admin_headers
)
assert ok.status_code == 200
assert ok.json()["rules"]["updatedBy"] == "admin"
# ---- agent 只读:token 读规则库 / 读岗位(含对内字段) ----
def test_agent_rules_and_jobs_readonly(agent_client):
# 未带 token -> 401
assert agent_client.get("/api/agent/recruiting-rules").status_code == 401
assert agent_client.get("/api/agent/jobs?jobId=x").status_code == 401
agent_headers = {"X-Ingest-Token": "agent-secret-token"}
# 规则库可读
rules = agent_client.get("/api/agent/recruiting-rules", headers=agent_headers).json()
assert rules["ok"] is True
assert rules["rules"]["hardRules"] == []
# 建岗位(含对内字段)到 admin 工作区
admin_token = _login(agent_client, "admin", "admin-secret").json()["token"]
admin_headers = _headers(admin_token)
job = {
"id": "job-agent-1",
"title": "高级前端工程师",
"department": "技术部",
"jd": "负责前端架构与落地",
"mustHave": "Vue3",
"sensitive": "年龄 30 以下(内部)",
"genderRestriction": "无",
"jdStatus": "已生效",
}
state = {"jobs": [job], "candidates": [], "offers": [], "tasks": [], "eventLog": []}
assert agent_client.put("/api/state", json=state, headers=admin_headers).status_code == 200
# 按 jobId 读回(含对内字段)
found = agent_client.get("/api/agent/jobs?jobId=job-agent-1", headers=agent_headers).json()
assert found["total"] == 1
assert found["jobs"][0]["title"] == "高级前端工程师"
assert found["jobs"][0]["sensitive"] == "年龄 30 以下(内部)"
# 按 title 模糊读
by_title = agent_client.get("/api/agent/jobs?title=前端", headers=agent_headers).json()
assert by_title["total"] >= 1
# 不存在 -> 空
missing = agent_client.get("/api/agent/jobs?jobId=job-nope", headers=agent_headers).json()
assert missing["total"] == 0
assert agent_client.get("/api/agent/jobs", headers=agent_headers).json()["total"] == 0
......@@ -12,15 +12,15 @@ recruit-grill skill 的产出是四件套(对外 JD / 对内笔记 / CONTEXT
## 决策 / Decision
1. **落位分层**:skill 本体(recruit-grill / jd-sync)以招聘系统仓库为**唯一维护源**`docs/recruit-grill`),qwenpaw 各工作区只放运行副本;**长期产出不进 qwenpaw 目录**。岗位级产出(对外字段 + 对内字段 + 访谈笔记归档)落招聘系统 `job` 数据;公司级知识(招聘硬规则/术语表/待定决策)落招聘系统侧**轻量规则库**(运行时文件 + 读写 API,admin 可写、登录用户与授权 agent 只读),首版**不建业务表**;关键词迭代表落岗位 `matchKeywords`;在招岗位表由岗位状态派生,不落 markdown。根 `CONTEXT.md` 保持领域词表单义,招聘规则库与其剥离。
2. **入口2 形态**企微那台把 JD **直接推 OA 审批(不经招聘系统先行占位)**;审批通过后,招聘系统新增**「岗位认领」菜单**,招聘专员把审批通过的岗位**认领进本人岗位管理**(生成 jobId 后才成为正式岗位)。
2. **入口2 形态**用人部门在企微聊完(recruit-grill)产出**定稿包**(对外 JD MD + 结构化 JSON + 对内笔记 MD,internalOnly 不进 OA)后,由 QwenPaw **代建 OA 审批单**(对外 MD + 结构化 JSON;不经招聘系统先行占位、不向用人部门提供"同步到招聘系统"话术);审批通过后,招聘系统新增**「岗位认领」菜单**,招聘专员把审批通过的岗位**认领进本人岗位管理**(生成 jobId 后才成为正式岗位)。
3. **待认领源做成可切换插槽**:「OA 审批直读(后期同库,hget_oa 只读)」与「`sync-external` 直接落待认领区(过渡期/本地测试)」统一走同一个**待认领池**入口;外部 JD 不再直接建进岗位管理(旧直建行为降级为本地导入兜底)。
4. **认领语义复刻 ADR 0001 行级模式**:OA 审批表只读;招聘系统本地维护 bill↔job↔owner 映射(不进整包 state,类比 ADR 0002);认领 = 行级迁移 + 生成 jobId + 占位(他人见"已认领")+ 清待认领快照 + 事件/待办联动。列表全 HR 可见,认领归属本人。
5. **认领落点状态**:认领后 `jdStatus=待确认``approvalStatus=审批通过(来源 OA)`;招聘专员核对 JD 内容后置 `已生效` 再开招,不自动生效。
6. **OA bill 数据契约**:入口2 审批单必须携带**岗位全量 JSON(对外字段 + 对内字段/笔记分列标记)**,认领时按 normalize 规则解析入库;对内字段继续受 `_INTERNAL_ONLY_JOB_KEYS` 脱敏保护,不落入对外 JD / LLM prompt。
6. **OA bill 数据契约(双轨)**:审批单携带**对外 MD + 结构化 JSON(岗位执行字段,含 age/gender/nonCompete 等对内执行字段)****对内笔记 MD(寻源策略/命脉推导等)不进 OA**,由 agent 走内部通道按 `oaBillNo` 关联暂存,HR 认领 OA 通过岗位时合并归档进岗位 `internalNote`;对内内容继续受 `_INTERNAL_ONLY_JOB_KEYS`(含 `internalNote`脱敏保护,不落入对外 JD / LLM prompt。
7. **入口1 审批源 seam**:保留系统内简化审批(草稿→待确认→审批中→已生效)为**过渡态**`approvalStatus` 设计为可切换源(内部简化 ↔ OA 直读),同库接通后映射 OA 表状态(待审/通过/驳回)到系统枚举;OA 是审批生命周期 owner,招聘系统只读消费。
## 影响 / Consequences
- 正向:企微产出与 HR 流程解耦又收口(统一经"待认领池 → 认领 → 岗位管理");认领复用人才池已验证的行级模式,不破坏整包 state / owner 隔离;对内字段脱敏纪律不变;skill 与规则库单一维护源,企微那台与本地不再漂移。
- 代价/注意:待认领清单与 bill 映射为整包 state 之外的独立存储 → 整包导出备份不含待认领区,需在认领/迁移逻辑中显式处理;OA 未接通前认领菜单只有过渡期数据源(sync-external 落点);入口2 若 OA 侧未按契约携带全量 JSON,认领会缺对内档案(契约需在企微那台 skill 分发时一并固化)。
- 代价/注意:待认领清单与 bill 映射为整包 state 之外的独立存储 → 整包导出备份不含待认领区,需在认领/迁移逻辑中显式处理;OA 未接通前认领菜单只有过渡期数据源(sync-external 落点,仅内部/QA 用);双轨要求 OA bill 与 internalOnly 载荷以 `oaBillNo` 对齐,否则认领会缺对内档案(契约需在企微那台 skill 分发时一并固化)。
- 后续可做:OA 同库直读落地(hget_oa,审批表结构确认后实现 Reader 映射);规则库从文件升级为业务表(当公司级规则需要驱动 AI 初筛时);认领前简历去重 / 批量认领。
\ No newline at end of file
# 外部岗位认领 + OA 审批直读 —— 设计定稿与落地清单
> 关联:ADR 0003(决策正文与理由)、根 CONTEXT.md(术语:外部岗位/待认领岗位/岗位认领/岗位对内档案/招聘规则库/OA 审批直读)
> 状态:**Phase 1 已落地(2026-09-03)**:后端待认领表 + sync-external 落待认领区 + 认领接口(89 tests passed);前端「岗位认领」菜单/页面 + 外部已审批岗位「核对后确认生效」流转(npm run build 通过)。
> 未做:Phase 2(规则库读写 API + agent 只读查询)、Phase 3(OA 同库直读 Reader + 审批源切换)。
> 状态:**Phase 1 + Phase 2 已落地(2026-09-03)**(93 tests passed)。
> - Phase 1:后端待认领表 + sync-external 落待认领区 + 认领接口;前端「岗位认领」菜单/页面 + 外部已审批岗位「核对后确认生效」流转(npm run build 通过)。
> - Phase 2:招聘规则库读写 API(GET/PUT /api/recruiting-rules,登录读 / admin 写,文件 data/recruiting-rules.json);agent 只读查询(GET /api/agent/recruiting-rules、GET /api/agent/jobs?jobId=&title=,X-Ingest-Token)。
> Phase 2.5(skill 契约):仓库维护源已同步 ADR 0003——recruit-grill 增加「两种部署形态」(形态 B 读系统规则库/岗位 API);jd-sync 收编至 docs/jd-sync。
> **2026-09-03 battle 校正**:入口2 收尾 = 产出定稿包 → agent 代建 **OA 审批单**(非"同步到招聘系统"话术);OA bill 走双轨(对外 MD + 执行字段进 OA,对内笔记 MD 经 internalOnly 载荷按 oaBillNo 关联,认领时合并归档 internalNote);jd-sync 仅内部/QA 导入。
> 待执行:把仓库 skill 副本部署到企微那台/本地 qwenpaw 工作区并配置环境变量。
> 未做:Phase 3(OA 同库直读 Reader + 审批源切换)。
## 一、定稿流程
```
入口1:HR 在 recruit-sys 建岗(网页/系统内 grill) ──提交 OA(bill↔jobId)──▶ OA 审批
入口2:用人部门在企微 QwenPaw(recruit-grill) 产出 JD ──JD 直推 OA(bill 携带岗位全量 JSON)──▶ OA 审批
入口2:用人部门在企微 QwenPaw(recruit-grill) 聊完 → 产出定稿包
(对外 JD MD + 结构化 JSON + 对内笔记 MD[internalOnly])
→ agent 代建 OA 审批单(对外 MD + 结构化 JSON)──▶ OA 审批
│ 审批通过
recruit-sys「岗位认领」菜单(待认领源可切换:OA 同库直读 / 过渡期 sync-external)
│ 招聘专员认领 → 生成 jobId → 进本人岗位管理
recruit-sys「岗位认领」菜单(待认领源可切换:OA 同库直读 / 过渡期 sync-external[内部QA])
│ 认领时按 oaBillNo 合并 internalOnly 载荷
│ → 岗位归档 internalNote → 生成 jobId → 进本人岗位管理
岗位 待确认(approvalStatus=审批通过·来源OA)→ HR 核对 → 已生效 → 发布 → 招聘流程
读回:下次访谈 GET /api/agent/jobs + /api/agent/recruiting-rules
```
## 二、现状已具备(无需改)
......@@ -43,8 +52,8 @@
11. `utils/normalize.js` / `utils/constants.js`:新增字段(含对内笔记归档)与后端 data JSON 对齐
### skill / 分发
12. `docs/recruit-grill/` 提升为唯一维护源并提交(SKILL.md 注明产出路径约定 = 岗位全量 JSON 走 bill,不落 qwenpaw 目录
13. `jd-sync` / `recruit-grill` 产出契约与 bill 全量字段一致;SKILL.md「改版纪律」四件套载体同步检查
12. `docs/recruit-grill/` 提升为唯一维护源并提交(SKILL.md 注明:形态 B 产出定稿包走 OA 审批,不落 qwenpaw 目录、不给用人部门"同步到招聘系统"话术
13. `jd-sync`(内部/QA)产出契约与 OA bill 双轨契约一致(对外+执行字段进 OA,internalNote 经 internalOnly 载荷关联 oaBillNo);SKILL.md「改版纪律」四件套载体同步检查
14. 分发:本地 qwenpaw workspace 与 ~/.agents/skills 副本以仓库为准;企微那台部署按同一份拷贝 + OA 侧契约
### 测试与规范
......
---
name: jd-sync
description: 【内部/QA 专用】把已经梳理/确认好的招聘 JD 导入招聘系统「待认领区」(ADR 0003 过渡期通道;生产链路是 OA 审批 → HR 岗位认领)。仅 HR/开发者联调用,不面向用人部门——用人部门收尾动作是产出定稿后提交 OA 审批,不是"同步到招聘系统"。
---
# jd-sync —— 把 JD 同步到招聘系统
## 用途
**定位:内部/QA 过渡导入通道,不面向用人部门。**
生产链路(ADR 0003):用人部门在企微聊完 → agent 代建 OA 审批单 → 审批通过 → HR 在系统「岗位认领」认领。
本 skill 只在 OA 未接通或本地联调时,把一份已定稿的岗位 JSON 导入招聘系统公司级「待认领区」(**不再直接创建岗位**),HR 认领后成为正式岗位。
## 触发(仅内部)
- HR / 开发者联调:把一份已定稿的岗位 JSON 导入待认领区做端到端验证。
- 一轮 recruit-grill 访谈收尾、已产出可用 JD 正文与结构化字段之后(内部转 QA 时)。
- **不要**把它做成用人部门的话术;用人部门侧对应动作是「提交 OA 审批」(OA 建单已由 agent 实现)。
## 步骤
1. 汇总岗位全量信息,组织 payload:
- 幂等键(任一即可,重复推送更新同一认领单):`external_id` / `externalJobId` / `oaBillNo`
- `job` 对外字段:title / department / jd / responsibilities / requirements / mustHave / niceToHave / knockout / competency / matchKeywords / headcount …
- `job` 内可选**对内字段分列**(仅内部):sensitive / genderRestriction / ageRestriction / internalFeasibility / verification / probationPeriod / probationCriteria / nonCompete 等——系统认领后存入岗位**对内档案**,绝不进入对外 JD / LLM prompt
2. 运行同步脚本(脚本路径相对本 skill 目录):
```bash
cd {this_skill_dir} && python scripts/sync_jd.py --payload-file /tmp/jd.json
```
或直接把 JSON 通过 `--payload '{"job":{...},"external_id":"..."}'` 传入。
3. 脚本把 JSON POST 到招聘系统 `POST /api/jd/sync-external`
4. 读取脚本输出并告知用户:
- `ok=true``operation=created/updated` → 已落入待认领区:**"已同步,等待 HR 在系统「岗位认领」认领"**(可附 claim id);
- `operation=skipped``reason=already_claimed`)→ 该岗位已被 HR 认领,提示用户去系统查看已认领岗位;
- 失败 → 把失败原因反馈给用户。
## 注意
- 环境变量读取:`RECRUITMENT_API_BASE`(默认 `http://127.0.0.1:4177`)、`RECRUITMENT_INGEST_TOKEN`(对应后端 `INGEST_TOKEN`,可选)。
- 若脚本失败,请检查招聘系统是否启动、地址/令牌是否匹配。
#!/usr/bin/env python3
"""把 JD 同步到招聘系统 /api/jd/sync-external(ADR 0003:落「待认领区」)。
用法:
python scripts/sync_jd.py --payload '{"job":{...},"external_id":"..."}'
python scripts/sync_jd.py --payload-file /tmp/jd.json
环境变量:
RECRUITMENT_API_BASE 招聘系统地址(默认 http://127.0.0.1:4177)
RECRUITMENT_INGEST_TOKEN 与后端 INGEST_TOKEN 一致的令牌(可选)
"""
from __future__ import annotations
import argparse
import json
import os
import sys
import urllib.error
import urllib.request
def build_payload(args: argparse.Namespace) -> dict:
if args.payload:
return json.loads(args.payload)
if args.payload_file:
with open(args.payload_file, encoding="utf-8") as f:
return json.load(f)
return {}
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(description="sync JD to recruitment system")
parser.add_argument("--payload", help="JSON string payload")
parser.add_argument("--payload-file", help="path to JSON payload file")
args = parser.parse_args(argv)
payload = build_payload(args)
if not payload:
print("ERROR: no payload provided. Pass --payload or --payload-file.", file=sys.stderr)
return 2
base = os.environ.get("RECRUITMENT_API_BASE", "http://127.0.0.1:4177").rstrip("/")
url = f"{base}/api/jd/sync-external"
token = os.environ.get("RECRUITMENT_INGEST_TOKEN", "").strip()
headers = {"Content-Type": "application/json"}
if token:
headers["X-Ingest-Token"] = token
data = json.dumps(payload, ensure_ascii=False).encode("utf-8")
req = urllib.request.Request(url, data=data, headers=headers, method="POST")
try:
with urllib.request.urlopen(req, timeout=30) as resp:
body = resp.read().decode("utf-8", "replace")
print(body)
try:
result = json.loads(body)
except json.JSONDecodeError:
result = {}
if result.get("ok"):
op = result.get("operation", "")
claim = result.get("claim") or {}
if op == "skipped":
print(f'OK: already_claimed -> claim {claim.get("id", "")} / {claim.get("title", "")}')
else:
print(f'OK: JD {op} -> 待认领区 claim {claim.get("id", "")} / {claim.get("title", "")}')
return 0
print(f"FAIL: {body[:500]}", file=sys.stderr)
return 1
except urllib.error.HTTPError as e:
print(f"HTTP {e.code}: {e.read().decode('utf-8', 'replace')[:500]}", file=sys.stderr)
return 1
except Exception as e:
print(f"ERROR: {e}", file=sys.stderr)
return 1
if __name__ == "__main__":
raise SystemExit(main())
......@@ -11,6 +11,27 @@ description: 从用人部门的一句话招聘需求出发,用逼问式访谈
用人部门发起需求时通常说不清"要什么样的人"——不是他们不专业,是**需求本身没被逼问过**。这个 skill 的价值就是把模糊的"我要招个人"逼成可判断的硬门槛、判定标准、排除信号和搜索词。
## 两种部署形态(ADR 0003,重要)
本 skill 有两种跑法,决定「读什么 / 写到哪里」:
**形态 A · 独立本地(无招聘系统 / HR 自用)**:维持本文件「产出」节的**文件四件套**——`01-jd/<role>.md``01-jd/_internal/<role>.md`、CONTEXT.md 硬规则更新、关键词迭代表;工作区文件即事实源,现状不变。
**形态 B · 接入招聘系统(recruit-sys,企微/本地 QwenPaw 部署,ADR 0003)**
- **读(会话开始前拉上下文,不再以工作区 CONTEXT.md / 01-jd 为事实源,只留当次会话草稿)**
- `GET {RECRUITMENT_API_BASE}/api/agent/recruiting-rules`(头 `X-Ingest-Token`)→ 公司级硬规则 / 术语表 / 待定决策;
- `GET {RECRUITMENT_API_BASE}/api/agent/jobs?title=<岗位名>`(或 `?jobId=`)→ 老岗位对外 JD + **对内档案**,供「旧岗位迭代 / 入口分流 / 同步前查重」。
- **写(访谈收尾 = 产出「定稿包」,不再以 markdown 为长期事实源;不向用人部门提供"同步到招聘系统"话术)**
- 对外 JD MD(给人看 / 给 OA 审批);
- 结构化 JSON(岗位执行字段,含 age/gender/nonCompete 等对内执行字段);
- 对内笔记 MD(internalOnly:寻源策略/命脉推导,**不进 OA**)。
- 用户确认后由 agent **代建 OA 审批单**(对外 MD + 结构化 JSON;OA 建单能力在企微那台已实现)。审批通过后 HR 在招聘系统「岗位认领」认领,认领时按 `oaBillNo` 合并 internalOnly 对内载荷归档进岗位 `internalNote`
- **不重复维护**:在招岗位表 = 系统岗位状态派生;关键词迭代表 R1 → 岗位 `matchKeywords`;性别/年龄等敏感要求只放 JSON 内字段,绝不进对外 JD / LLM prompt。
- **环境变量**`RECRUITMENT_API_BASE``RECRUITMENT_INGEST_TOKEN`(与系统 `INGEST_TOKEN` 一致;配置后才有读取与同步权限)。
## 访谈纪律(最重要)
- **一轮可问 2~3 个相关问题**,组内合并问——全流程 **8 问 / 6 轮闭环**(顺序见 `references/question-bank.md`)。**绝不增加提问轮次**:用人部门反馈提问过多,收敛轮次是本次优化的核心诉求。
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment