Shared

shared —— 基础模块

本目录是全工具包依赖的 Python 库。不是入口，不被任何 AI 智能体当作 Skill 直接加载。 业务脚本（boss_jd.py / recommend_list.py 等）import 这些模块即可。

模块清单

cdp_preflight.py —— CDP 连接 + 登录态探测

设计动机

本工具包不依赖任何第三方 CLI（包括早期项目曾用的 boss_agent_cli 的 boss login / boss me / boss status 等登录检测命令）。本模块用 patchright 直连 Edge CDP，自己判断：

• Edge 是否在指定端口（如 http://localhost:9222）跑着
• BOSS 招聘者 session 是否有效（zp_at + wt2 + bst 三 cookie 都在且非空）
• 当前页面是 recommend / chat / job_edit / login / unknown

所有 BOSS HTTP 调用（recruiter_job_catalog 的 list_jobs 等）都走浏览器内 fetch（page.evaluate('fetch(...)')）， 复用浏览器真实 TLS 指纹 + 自动带 cookie，不需要单独同步 __zp_stoken__。

接口

from cdp_preflight import (
    connect_cdp,         # 函数
    check_login,         # 函数
    get_cookies,         # 函数
    CDPSession,          # dataclass
    RECRUITER_REQUIRED_COOKIES,  # tuple: ('zp_at', 'wt2', 'bst')
)

# 1) 连 Edge
session = connect_cdp()                    # 默认 http://localhost:9222
session = connect_cdp("http://1.2.3.4:9222", timeout_ms=15000)

# 2) 检查登录
state = check_login(session)
# state = {
#   'logged_in': True/False,
#   'cookies': {
#     'zp_at': True/False,
#     'wt2': True/False,
#     'bst': True/False,
#     'present': [...],   # 所有 cookie 名
#     'total': int,
#   },
#   'current_url': 'https://www.zhipin.com/web/chat/recommend',
#   'page_kind': 'recommend' | 'chat' | 'job_edit' | 'login' | 'unknown',
# }

# 3) 取 cookie（HTTP 库复用）
cookies = get_cookies(session)  # {'zp_at': '...', 'wt2': '...', ...}

# 4) 关闭（不关 Edge，只关 playwright 进程）
session.disconnect()

异常

失败判定（业务侧怎么用）

state = check_login(session)
if not state['logged_in']:
    missing = [k for k, v in state['cookies'].items() if not v and k in RECRUITER_REQUIRED_COOKIES]
    raise RuntimeError(f"BOSS 未登录，缺 cookie: {missing}；请在 9222 Edge 窗口扫码")

if state['page_kind'] == 'login':
    raise RuntimeError("BOSS 已踢回登录页，请重新扫码")

if state['page_kind'] == 'unknown':
    raise RuntimeError("当前页面不是 BOSS 域，请把 Edge 切到 BOSS 后再试")

recruiter_job_catalog.py —— 招聘者岗位目录

设计动机

本工具包不依赖任何第三方 CLI（包括早期项目曾用的 boss_agent_cli 的 boss --role recruiter hr jobs list 命令）。本模块通过 patchright 连 CDP， 在已登录的浏览器 context 里调 fetch 打 https://www.zhipin.com/wapi/zpjob/job/chatted/jobList（GET）， 复用浏览器 TLS 指纹 + 自动带 cookie，不依赖 stoken/手写 session.enc 同步。

BOSS 后端 API（来自 boss_agent_cli/api/recruiter.yaml）

BOSS 响应 schema（已验证）：

{"code": 0, "message": "Success", "zpData": [
  {"encryptJobId": "9a7759...", "jobId": 559622717, "jobName": "...", "description": "...",
   "salaryDesc": "15-20K", "address": "宁波", "jobOnlineStatus": 1, ...},
  ...
]}

zpData 直接是 list（不是 {"list": [...]} 那种二次包装）。

接口

from recruiter_job_catalog import (
    list_jobs,              # 函数：拉所有岗位
    resolve_recruiter_job,  # 函数：按 query 定位单个岗位
    fetch_job_detail,       # 函数：拿单个岗位 JSON 详情（结构化字段）
    JOB_LIST_URL,           # 常量
    JOB_EDIT_URL,           # 常量
    BASE_URL,               # 常量 'https://www.zhipin.com'
)

# 1) 列岗位
result = list_jobs()
# result = {
#   'ok': True/False,
#   'command': 'recruiter-jobs-list',
#   'schema_version': '1.0',
#   'data': [{...}, ...],      # 或 None（失败时）
#   'pagination': None,
#   'error': {'code': str, 'message': str} | None,
#   'hints': None,
# }
if not result['ok']:
    raise RuntimeError(f"list_jobs 失败: {result['error']}")
for job in result['data']:
    print(job['encryptJobId'], job['jobName'], job['salaryDesc'])

# 2) 解析 query
job = resolve_recruiter_job('9a7759badfd95d350nFz3d-_F1NX')   # eid 精确
job = resolve_recruiter_job('559622717')                        # jobId 精确
job = resolve_recruiter_job('线控底盘制动、转向工程师')         # jobName 精确
job = resolve_recruiter_job('工程师')                          # 模糊（含子串）
job = resolve_recruiter_job('不存在的岗位xxx')                  # → None
# job = {'encryptJobId', 'jobId', 'jobName', 'address', 'salaryDesc', ...}

# 3) JSON 详情
detail = fetch_job_detail('9a7759badfd95d350nFz3d-_F1NX')
# detail = {'ok': True, 'command': 'recruiter-jobs-detail', 'data': {...BOSS 原始 JSON...}}

query 匹配规则

resolve_recruiter_job(query) 按以下顺序找：

1. encryptJobId 精确（最权威，找到就立刻返回）
2. jobId 精确（str/int 兼容）
3. jobName 精确
4. jobName 含 query 子串（模糊，第一个匹配）

失败码（error.code）

注意

• fetch_job_detail 是结构化 JSON 详情，跟 boss_jd.py 抓 iframe DOM（含富文本描述）是两套。
  • 业务侧一般用 boss_jd.py 拿完整 JD
  • 本函数用于「快速校验岗位存在」或「辅助对比」
• 真实 Edge TLS 指纹 + 自动 cookie，不需要管 __zp_stoken__
• 每次调用 connect_cdp() 是有成本的（patchwright 实例化约 1-2s）。 在一个业务脚本里尽量复用同一个 session。

输出路径管理（output_manager.py）

from output_manager import JobOutputManager

out = JobOutputManager(
    job_name='线控底盘制动、转向工程师',
    encrypt_job_id='9a7759badfd95d350nFz3d-_F1NX',
    run_id='2026-07-31_134548',   # 必填（新任务由 RunOrchestrator 创建）
)

# 路径属性
out.jd_path              # job_detail.json
out.new_resumes_path     # new_resumes.json
out.recommend_geek_ids_path  # recommend_geek_ids.json
out.screening_results_path   # screening_results.json
out.report_path          # HTML 报告
out.get_process_path('foo.json')  # 通用 process/ 下文件

run_id 生命周期（run_orchestrator.py）

from run_orchestrator import RunOrchestrator

orch = RunOrchestrator(
    job_name='线控底盘制动、转向工程师',
    encrypt_job_id='9a7759badfd95d350nFz3d-_F1NX',
)

# 新任务（boss_jd.py 入口）
run_id = orch.create_new_run()      # 自动生成 YYYY-MM-DD_HHMMSS

# 继续旧任务（其它脚本入口）
run_id = orch.bind_existing_run('2026-07-31_134548')

# 标记步骤
orch.mark_done('jd', run_id=run_id)
orch.init_run_state(run_id)        # 翻 confirmed=false，等用户确认
orch.finish(run_id)                # 整轮跑完（默认 auto_greet 成功时自动调）

详细铁律见 boss-hr-auto/SKILL.md 的 ## 🚨 run_id 铁律 章节。

Windows 环境编码

fix_encoding.py 在所有 CLI 脚本入口都 import，强制 stdout 为 UTF-8。 仍然推荐：跑脚本时显式用 python -X utf8 ... 或设 PYTHONIOENCODING=utf-8 环境变量。

PowerShell / cmd 中文参数解析不稳定：所有 CLI 调用应通过 cli_runner.py（参数数组，不拼字符串）。

cli_runner 工具白名单

白名单外 tool 一律拒绝（ValueError）；脚本路径逃逸（../）也拒绝。