Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
37 changes: 37 additions & 0 deletions boss_cli/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -314,6 +314,43 @@ def search_jobs(
params["jobType"] = job_type
return self._get(JOB_SEARCH_URL, params=params, action="搜索职位")

def search_jobs_by_url(
self,
url: str,
page: int = 1,
page_size: int = 15,
) -> dict[str, Any]:
"""Search jobs using a BOSS web URL.

Parses query parameters from a zhipin.com search URL and passes
them directly to the API, preserving all filter selections
(including multi-select) exactly as configured on the website.

Example URL:
https://www.zhipin.com/web/geek/jobs?city=101210100&experience=102,108&query=前端
"""
from urllib.parse import urlparse, parse_qs

parsed = urlparse(url)
qs = parse_qs(parsed.query, keep_blank_values=False)

# Extract known parameters (use first value from parse_qs lists)
params: dict[str, Any] = {
"query": qs.get("query", [""])[0],
"city": qs.get("city", ["101010100"])[0],
"page": page,
"pageSize": page_size,
}

# Forward all filter params directly (supports multi-select via comma)
for key in ("experience", "degree", "salary", "jobType",
"industry", "scale", "stage"):
val = qs.get(key, [None])[0]
if val:
params[key] = val

return self._get(JOB_SEARCH_URL, params=params, action="搜索职位(URL)")

def get_recommend_jobs(self, page: int = 1) -> dict[str, Any]:
"""Get personalized job recommendations.

Expand Down
203 changes: 133 additions & 70 deletions boss_cli/commands/search.py
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,28 @@

# ── Helper: render job table ────────────────────────────────────────

def _resolve_multi(value: str | None, code_map: dict[str, str]) -> str | None:
"""Resolve a possibly comma-separated filter value to API codes.

Supports both single values (e.g., '应届生') and multi-select
(e.g., '应届生,在校生' -> '102,108').
"""
if not value:
return None
parts = [v.strip() for v in value.split(",") if v.strip()]
codes = []
for part in parts:
code = code_map.get(part)
if code:
codes.append(code)
else:
available = ", ".join(code_map.keys())
raise click.BadParameter(
f"未知选项 '{part}'。可选值: {available}"
)
return ",".join(codes) if codes else None


def _render_job_table(
job_list: list[dict], title: str, page: int = 1, hint_next: str = "",
) -> None:
Expand Down Expand Up @@ -77,61 +99,88 @@ def _render_job_table(
# ── search ──────────────────────────────────────────────────────────

@click.command()
@click.argument("keyword")
@click.argument("keyword", required=False, default=None)
@click.option("--url", "search_url", default=None, help="直接使用 BOSS 网页 URL 搜索 (所有 filter 选项从 URL 解析)")
@click.option("-c", "--city", default="全国", help="城市名称或代码 (默认: 全国)")
@click.option("-p", "--page", default=1, type=int, help="页码 (默认: 1)")
@click.option("--salary", type=click.Choice(list(SALARY_CODES.keys())), help="薪资筛选")
@click.option("--exp", type=click.Choice(list(EXP_CODES.keys())), help="工作经验筛选")
@click.option("--degree", type=click.Choice(list(DEGREE_CODES.keys())), help="学历筛选")
@click.option("--industry", type=click.Choice(list(INDUSTRY_CODES.keys())), help="行业筛选 (如: 互联网, 金融)")
@click.option("--scale", type=click.Choice(list(SCALE_CODES.keys())), help="公司规模筛选 (如: 1000-9999人)")
@click.option("--stage", type=click.Choice(list(STAGE_CODES.keys())), help="融资阶段筛选 (如: A轮, 已上市)")
@click.option("--job-type", type=click.Choice(list(JOB_TYPE_CODES.keys())), help="职位类型 (全职/兼职/实习)")
@click.option("--salary", default=None, help="薪资筛选 (支持逗号分隔多选, 如: 5-10K,10-15K)")
@click.option("--exp", default=None, help="工作经验筛选 (支持逗号分隔多选, 如: 应届生,在校生)")
@click.option("--degree", default=None, help="学历筛选 (支持逗号分隔多选, 如: 本科,硕士)")
@click.option("--industry", default=None, help="行业筛选 (支持逗号分隔多选, 如: 互联网,人工智能)")
@click.option("--scale", default=None, help="公司规模筛选 (支持逗号分隔多选)")
@click.option("--stage", default=None, help="融资阶段筛选 (支持逗号分隔多选)")
@click.option("--job-type", default=None, help="职位类型 (支持逗号分隔多选, 如: 全职,实习)")
@structured_output_options
def search(
keyword: str, city: str, page: int,
keyword: str | None, search_url: str | None, city: str, page: int,
salary: str | None, exp: str | None, degree: str | None,
industry: str | None, scale: str | None, stage: str | None, job_type: str | None,
as_json: bool, as_yaml: bool,
) -> None:
"""搜索职位 (例: boss search Python --city 北京 --industry 互联网)"""
cred = require_auth()

city_code = resolve_city(city)
salary_code = SALARY_CODES.get(salary) if salary else None
exp_code = EXP_CODES.get(exp) if exp else None
degree_code = DEGREE_CODES.get(degree) if degree else None
industry_code = INDUSTRY_CODES.get(industry) if industry else None
scale_code = SCALE_CODES.get(scale) if scale else None
stage_code = STAGE_CODES.get(stage) if stage else None
job_type_code = JOB_TYPE_CODES.get(job_type) if job_type else None

def _action(c: BossClient) -> dict:
return c.search_jobs(
query=keyword, city=city_code, page=page,
experience=exp_code, degree=degree_code, salary=salary_code,
industry=industry_code, scale=scale_code, stage=stage_code,
job_type=job_type_code,
)
"""搜索职位

def _render(data: dict) -> None:
job_list = data.get("jobList", [])
# Always save index cache for `boss show` navigation
if job_list:
save_index(job_list, source=f"search:{keyword}")
支持两种方式:
boss search Python --city 北京 --exp 应届生
boss search --url "https://www.zhipin.com/web/geek/jobs?city=..."
"""
if not keyword and not search_url:
raise click.UsageError("必须提供 KEYWORD 或 --url 参数")

filters = [city]
for f in (salary, exp, degree, industry, scale, stage, job_type):
if f:
filters.append(f)
filter_str = " · ".join(filters)
cred = require_auth()

_render_job_table(
job_list,
title=f"🔍 搜索: {keyword} ({filter_str})",
page=page,
hint_next=f"更多结果: boss search \"{keyword}\" --city {city} -p {page + 1}" if data.get("hasMore") else "",
)
if search_url:
# URL 模式: 直接从 URL 解析参数
from urllib.parse import urlparse, parse_qs, unquote
qs = parse_qs(urlparse(search_url).query)
display_query = unquote(qs.get("query", [""])[0]) or "(URL搜索)"

def _action(c: BossClient) -> dict:
return c.search_jobs_by_url(search_url, page=page)

def _render(data: dict) -> None:
job_list = data.get("jobList", [])
if job_list:
save_index(job_list, source=f"search:{display_query}")
_render_job_table(
job_list,
title=f"🔍 搜索(URL): {display_query}",
page=page,
hint_next=f"更多结果: boss search --url \"{search_url}\" -p {page + 1}" if data.get("hasMore") else "",
)
else:
# 参数模式: 传统方式
city_code = resolve_city(city)
salary_code = _resolve_multi(salary, SALARY_CODES)
exp_code = _resolve_multi(exp, EXP_CODES)
degree_code = _resolve_multi(degree, DEGREE_CODES)
industry_code = _resolve_multi(industry, INDUSTRY_CODES)
scale_code = _resolve_multi(scale, SCALE_CODES)
stage_code = _resolve_multi(stage, STAGE_CODES)
job_type_code = _resolve_multi(job_type, JOB_TYPE_CODES)

def _action(c: BossClient) -> dict:
return c.search_jobs(
query=keyword, city=city_code, page=page,
experience=exp_code, degree=degree_code, salary=salary_code,
industry=industry_code, scale=scale_code, stage=stage_code,
job_type=job_type_code,
)

def _render(data: dict) -> None:
job_list = data.get("jobList", [])
if job_list:
save_index(job_list, source=f"search:{keyword}")
filters = [city]
for f in (salary, exp, degree, industry, scale, stage, job_type):
if f:
filters.append(f)
filter_str = " · ".join(filters)
_render_job_table(
job_list,
title=f"🔍 搜索: {keyword} ({filter_str})",
page=page,
hint_next=f"更多结果: boss search \"{keyword}\" --city {city} -p {page + 1}" if data.get("hasMore") else "",
)

handle_command(cred, action=_action, render=_render, as_json=as_json, as_yaml=as_yaml)

Expand Down Expand Up @@ -266,38 +315,46 @@ def _render_detail(data: dict) -> None:
# ── export ──────────────────────────────────────────────────────────

@click.command()
@click.argument("keyword")
@click.argument("keyword", required=False, default=None)
@click.option("--url", "search_url", default=None, help="直接使用 BOSS 网页 URL 导出 (所有 filter 从 URL 解析)")
@click.option("-c", "--city", default="全国", help="城市名称或代码")
@click.option("-n", "--count", default=30, type=int, help="导出数量 (默认: 30)")
@click.option("--salary", type=click.Choice(list(SALARY_CODES.keys())), help="薪资筛选")
@click.option("--exp", type=click.Choice(list(EXP_CODES.keys())), help="工作经验筛选")
@click.option("--degree", type=click.Choice(list(DEGREE_CODES.keys())), help="学历筛选")
@click.option("--industry", type=click.Choice(list(INDUSTRY_CODES.keys())), help="行业筛选")
@click.option("--scale", type=click.Choice(list(SCALE_CODES.keys())), help="公司规模筛选")
@click.option("--stage", type=click.Choice(list(STAGE_CODES.keys())), help="融资阶段筛选")
@click.option("--job-type", type=click.Choice(list(JOB_TYPE_CODES.keys())), help="职位类型")
@click.option("--salary", default=None, help="薪资筛选 (支持逗号分隔多选)")
@click.option("--exp", default=None, help="工作经验筛选 (支持逗号分隔多选)")
@click.option("--degree", default=None, help="学历筛选 (支持逗号分隔多选)")
@click.option("--industry", default=None, help="行业筛选 (支持逗号分隔多选)")
@click.option("--scale", default=None, help="公司规模筛选 (支持逗号分隔多选)")
@click.option("--stage", default=None, help="融资阶段筛选 (支持逗号分隔多选)")
@click.option("--job-type", default=None, help="职位类型 (支持逗号分隔多选)")
@click.option("-o", "--output", "output_file", default=None, help="输出文件路径 (默认: stdout)")
@click.option("--format", "fmt", type=click.Choice(["csv", "json"]), default="csv", help="输出格式")
def export(
keyword: str, city: str, count: int,
keyword: str | None, search_url: str | None, city: str, count: int,
salary: str | None, exp: str | None, degree: str | None,
industry: str | None, scale: str | None, stage: str | None, job_type: str | None,
output_file: str | None, fmt: str,
) -> None:
"""导出搜索结果为 CSV 或 JSON

例: boss export "golang" --city 杭州 -n 50 -o jobs.csv
支持两种方式:
boss export "golang" --city 杭州 -n 50 -o jobs.csv
boss export --url "https://www.zhipin.com/web/geek/jobs?city=..." -n 50 --format json
"""
cred = require_auth()
if not keyword and not search_url:
raise click.UsageError("必须提供 KEYWORD 或 --url 参数")

city_code = resolve_city(city)
salary_code = SALARY_CODES.get(salary) if salary else None
exp_code = EXP_CODES.get(exp) if exp else None
degree_code = DEGREE_CODES.get(degree) if degree else None
industry_code = INDUSTRY_CODES.get(industry) if industry else None
scale_code = SCALE_CODES.get(scale) if scale else None
stage_code = STAGE_CODES.get(stage) if stage else None
job_type_code = JOB_TYPE_CODES.get(job_type) if job_type else None
cred = require_auth()
use_url = search_url is not None

if not use_url:
city_code = resolve_city(city)
salary_code = _resolve_multi(salary, SALARY_CODES)
exp_code = _resolve_multi(exp, EXP_CODES)
degree_code = _resolve_multi(degree, DEGREE_CODES)
industry_code = _resolve_multi(industry, INDUSTRY_CODES)
scale_code = _resolve_multi(scale, SCALE_CODES)
stage_code = _resolve_multi(stage, STAGE_CODES)
job_type_code = _resolve_multi(job_type, JOB_TYPE_CODES)

all_jobs: list[dict] = []
pages_needed = (count + 14) // 15 # 15 per page
Expand All @@ -306,12 +363,15 @@ def export(
def _collect(c: BossClient) -> list[dict]:
nonlocal all_jobs
for pg in range(1, pages_needed + 1):
data = c.search_jobs(
query=keyword, city=city_code, page=pg,
experience=exp_code, degree=degree_code, salary=salary_code,
industry=industry_code, scale=scale_code, stage=stage_code,
job_type=job_type_code,
)
if use_url:
data = c.search_jobs_by_url(search_url, page=pg)
else:
data = c.search_jobs(
query=keyword, city=city_code, page=pg,
experience=exp_code, degree=degree_code, salary=salary_code,
industry=industry_code, scale=scale_code, stage=stage_code,
job_type=job_type_code,
)
job_list = data.get("jobList", [])
all_jobs.extend(job_list)
console.print(f" [dim]📦 第 {pg} 页: {len(job_list)} 个职位 (累计: {len(all_jobs)})[/dim]")
Expand All @@ -327,10 +387,12 @@ def _collect(c: BossClient) -> list[dict]:
else:
# CSV
buf = io.StringIO()
fieldnames = ["职位", "公司", "薪资", "经验", "学历", "城市", "地区", "技能", "securityId"]
fieldnames = ["职位", "公司", "薪资", "经验", "学历", "城市", "地区", "技能", "URL", "securityId"]
writer = csv.DictWriter(buf, fieldnames=fieldnames, extrasaction="ignore")
writer.writeheader()
for job in all_jobs:
eid = job.get("encryptJobId", "")
url = f"https://www.zhipin.com/job_detail/{eid}.html" if eid else ""
writer.writerow({
"职位": job.get("jobName", ""),
"公司": job.get("brandName", ""),
Expand All @@ -340,6 +402,7 @@ def _collect(c: BossClient) -> list[dict]:
"城市": job.get("cityName", ""),
"地区": job.get("areaDistrict", ""),
"技能": ", ".join(job.get("skills", [])),
"URL": url,
"securityId": job.get("securityId", ""),
})
output_text = buf.getvalue()
Expand Down
1 change: 1 addition & 0 deletions boss_cli/index_cache.py
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@ def save_index(jobs: list[dict[str, Any]], source: str = "search") -> None:
for job in jobs:
entry = {
"securityId": job.get("securityId", ""),
"encryptJobId": job.get("encryptJobId", ""),
"jobName": job.get("jobName", ""),
"brandName": job.get("brandName", ""),
"salaryDesc": job.get("salaryDesc", ""),
Expand Down