day10

ES 职位搜索

搜索接口入口

app/apis/es_data_api.py：

@es_data_router.get("/search", summary="职位搜索")
async def search_jobs(
keyword: Optional[str] = Query(None, description="搜索关键词，如 Java、前端"),
work_location: Optional[str] = Query(None, description="工作地点"),
edu_require: Optional[str] = Query(None, description="学历要求"),
status: Optional[int] = Query(1, description="职位状态，默认1=招聘中"),
page: int = Query(1, ge=1),
page_size: int = Query(10, ge=1, le=50),
sort: str = Query("time", description="排序：score=相关度，time=发布时间"),
es_client: AsyncElasticsearch = Depends(es_client_depend),
):
# ===== 先尝试 ES 全文检索 =====
try:
es_result = await _search_from_es(
es_client, keyword, work_location, edu_require,
exp_require, industry_id, company_scale, financing_stage,
status, page, page_size, sort,
)
if es_result is not None:
return es_result
except Exception as e:
logger.warning(f"ES 搜索异常，降级到 MySQL: {e}")

# ===== ES 不行就降级查 MySQL =====
logger.info("使用 MySQL 降级搜索")
return await _search_from_mysql(
keyword, work_location, edu_require, exp_require,
industry_id, company_scale, financing_stage, status,
page, page_size, sort,
)

核心思路：ES 返回 None（索引不存在 / 没数据）或抛异常，都走 MySQL。

ES 检索函数

async def _search_from_es(es_client, keyword, ...):
# 索引不存在 → 直接返回 None，交给 MySQL
if not await es_client.indices.exists(index=BOSS_JOB_INDEX_NAME):
logger.info("ES 索引不存在，降级到 MySQL")
return None

# … 组装 bool 查询（multi_match 做关键词，term 做筛选）…
resp = await es_client.search(index=BOSS_JOB_INDEX_NAME, query=query, ...)
total_count = resp["hits"]["total"]["value"]
lists = [hit["_source"] for hit in resp["hits"]["hits"]]

# 索引在但没数据 → 也降级
if total_count == 0 and not lists:
return None

return {"code": 1, "message": "success (ES)", "data": {"lists": lists, ...}}

MySQL 降级查询

这就是真正从无到有写出来的函数，用普通 ORM 查询拼出和 ES 一模一样的返回结构：

async def _search_from_mysql(keyword, work_location, edu_require, ...):
from tortoise.expressions import Q

q = Q()
if status is not None:
q &= Q(status=status)
if work_location:
q &= Q(work_location__icontains=work_location)
if keyword and keyword.strip():
kw = keyword.strip()
q &= Q(job_name__icontains=kw) | Q(job_desc__icontains=kw)

total_count = await Job.filter(q).count()
jobs = await Job.filter(q).order_by("-publish_time").offset(offset).limit(page_size)

# 批量查企业名，避免 N+1
enterprise_ids = list({j.enterprise_id for j in jobs if j.enterprise_id})
enterprise_map = {e["id"]: e["enterprise_name"]
for e in await Enterprise.filter(id__in=enterprise_ids).values("id", "enterprise_name")}

lists = [{
"job_id": job.id,
"job_name": job.job_name,
"work_location": job.work_location,
"min_salary": job.min_salary,
"max_salary": job.max_salary,
"edu_require": job.edu_require,
"exp_require": job.exp_require,
"job_tags": job.job_tags or [],
"status": job.status.value if hasattr(job.status, "value") else job.status,
"publish_time": job.publish_time.isoformat() if job.publish_time else None,
"enterprise_id": job.enterprise_id,
"enterprise_name": enterprise_map.get(job.enterprise_id, ""),
} for job in jobs]

return {"code": 1, "message": "success (MySQL)", "data": {"lists": lists, ...}}

为什么值得写：降级的返回结构和 ES 版完全一致，前端一套渲染逻辑通吃两种情况。