文章目录一、ES搜索二、职位详情和投递简历三、简历中心一、ES搜索后端代码es_data_router.get(/search, summary职位搜索, description职位搜索) async def search_job(# ---------- 关键词 ---------- keyword: Optional[str] Query(None, description搜索关键词如 Java、前端), # ---------- 筛选 ---------- work_location: Optional[str] Query(None, description工作地点精确匹配), edu_require: Optional[str] Query(None, description学历要求), exp_require: Optional[str] Query(None, description经验要求), industry_id: Optional[int] Query(None, description行业ID), company_scale: Optional[int] Query(None, description公司规模枚举值), financing_stage: Optional[int] Query(None, description融资阶段枚举值), status: Optional[int] Query(1, description职位状态默认1招聘中), # ---------- 分页 / 排序 ---------- page: int Query(1, ge1, description页码), page_size: int Query(10, ge1, le50, description每页条数), sort: str Query(time, description排序score相关度time发布时间), es_client: AsyncElasticsearch Depends(es_client_depend), ): # 索引不存在时友好提示 if not await es_client.indices.exists(indexBOSS_JOB_INDEX_NAME): return { code: 400, msg: f索引 {BOSS_JOB_INDEX_NAME} 不存在请先创建并同步数据, } must_clauses: list[dict[str, Any]] [] filter_clauses: list[dict[str, Any]] [] # ----- 1关键词打在多个 text 字段职位名称加权 ----- if keyword and keyword.strip(): must_clauses.append( { multi_match: { query: keyword.strip(), fields: [ job_name^3, # 名称命中权重更高 job_desc^2, duty_require, enterprise_name, industry_name, ], type: best_fields, operator: and, # 可按需改成 or召回更宽 } } ) # ----- 2硬筛选全部放 filter ----- if status is not None: filter_clauses.append({term: {status: status}}) if work_location: filter_clauses.append({term: {work_location: work_location}}) if edu_require: filter_clauses.append({term: {edu_require: edu_require}}) if exp_require: filter_clauses.append({term: {exp_require: exp_require}}) if industry_id is not None: filter_clauses.append({term: {industry_id: industry_id}}) if company_scale is not None: filter_clauses.append({term: {enterpriseInfo_company_scale: company_scale}}) if financing_stage is not None: filter_clauses.append({term: {enterpriseInfo_financing_stage: financing_stage}}) # ----- 3组装 bool ----- bool_query: dict[str, Any] {} if must_clauses: bool_query[must] must_clauses if filter_clauses: bool_query[filter] filter_clauses # 既无关键词也无筛选时匹配全部仍建议至少有默认 status filter query: dict[str, Any] {bool: bool_query} if bool_query else {match_all: {}} # ----- 4排序 ----- # 有关键词且 sortscore → 按相关度否则按发布时间倒序 if sort score and keyword: sort_clause: Any [_score, {publish_time: {order: desc, missing: _last}}] else: sort_clause [{publish_time: {order: desc, missing: _last}}] # ----- 5分页ES 用 from / size ----- from_ (page - 1) * page_size # 列表页只取卡片字段减小 _source source_fields [ job_id, job_name, work_location, min_salary, max_salary, salary_times, edu_require, exp_require, job_tags, status, publish_time, enterprise_id, enterprise_name, enterprise_city_name, enterpriseInfo_company_scale, enterpriseInfo_financing_stage, industry_id, industry_name, ] resp await es_client.search( indexBOSS_JOB_INDEX_NAME, queryquery, sortsort_clause, from_from_, sizepage_size, sourcesource_fields, track_total_hitsTrue, # 拿到准确 total ) hits resp.get(hits, {}) total hits.get(total, {}) # ES 7 total 一般是 {value: n, relation: eq} total_count total.get(value, 0) if isinstance(total, dict) else int(total or 0) lists [] for hit in hits.get(hits, []): item hit.get(_source) or {} item[_score] hit.get(_score) # 可选方便调相关度 lists.append(item) return { code: 1, message: success, data: { lists: lists, page_info: { page: page, page_size: page_size, total_count: total_count, total_page: (total_count page_size - 1) // page_size if page_size else 0, }, }, }二、职位详情和投递简历后端代码staticmethod async def get_job_by_id(job_id: int): res await Job.get_or_none(idjob_id) enterprise_id res.enterprise_id enterprise await Enterprise.get_or_none(identerprise_id) info { job_info: res, enterprise_info: enterprise } return info staticmethod async def resume_submission(job_seeker_id: int,job_id: int): oss AliyunOSSTool() attachmentResume await AttachmentResume.get_or_none( job_seeker_idjob_seeker_id,is_defaultTrue) file_url attachmentResume.file_url source_key file_url.replace(fhttps://{oss.bucket_name}.{oss.endpoint}/, ) logger.info(fsource_key:{source_key}) snowflake SnowflakeSingleton.get_instance(worker_id1) file_id snowflake.get_id() source_filename os.path.basename(source_key) target_key fjob_seeker_avatar/copy/{file_id}-{source_filename} copy_ok,copy_data oss.copy_single_file( source_oss_keysource_key, target_oss_keytarget_key, overwriteFalse) logger.info(fcopy_ok:{copy_ok},copy_data:{copy_data}) await ResumeSubmissionRecords.create( job_idjob_id, job_seeker_idjob_seeker_id, copy_resume_urlcopy_data[target_url], submit_timenow(), resume_statusResumeStatus.SUBMITTED ) job_router.get(/job/{job_id},summary职位详情) async def get_job_by_id(job_id:int,): res await JobService.get_job_by_id(job_id) return {code: 200, data: res, msg: 查询成功} job_router.post(/resume_submission,summary职位投递) async def resume_submission( job_id:intQuery(...,title职位ID,description职位ID), job_seeker_id Depends(get_job_info), ): res await JobService.resume_submission(job_seeker_id,job_id) return {code: 200, data: res, msg: 简历投递成功}三、简历中心staticmethod async def resume_center(team_id:int): jobs await Job.filter(recruit_team_idteam_id) info_list [] for job in jobs: job_id job.id resume_submission_records await ResumeSubmissionRecords.filter(job_idjob_id) for i in resume_submission_records: job_seeker_id i.job_seeker_id job_seeker await JobSeeker.get_or_none(idjob_seeker_id) jobIntention await JobIntention.get_or_none(job_seeker_idjob_seeker_id) workExperience (await WorkExperience.filter(job_seeker_idjob_seeker_id). order_by(-entry_time).first()) info_dict { job_seeker_id: job_seeker_id, job_seeker: job_seeker, jobIntention: jobIntention, workExperience: workExperience, resume_status: i.resume_status, submit_time: i.submit_time, } info_list.append(info_dict) return info_list enterprise_router.get(/resume_center, summary简历中心, description简历中心) async def resume_center(team_idDepends(get_job_info)): res await EnterpriseService.resume_center(team_id) return {code: 200, data: res, msg: 登出成功}