求职者端 - 职位搜索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“学历要求”),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),):“”职位搜索思路1. bool.must/should关键词 multi_match有 keyword 才加2. bool.filter城市、学历、状态等精确条件不参与算分3. from/size分页4. sort有词按相关度或按 publish_time“”# 索引不存在时友好提示if not await es_client.indices.exists(indexBOSS_JOB_INDEX_NAME):return {“code”: 0,“message”: 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_clausesif filter_clauses:bool_query[“filter”] filter_clauses# 既无关键词也无筛选时匹配全部仍建议至少有默认 status filterquery: 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# 列表页只取卡片字段减小sourcesource_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,fromfrom,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,},},}Lobster AIpython运行整段代码就是按照文档字符串docstring中的四步来构建 ES 查询并解析结果返回。下面详细拆解开头先 indices.exists 看索引在不在不在就直接返回 code:0 友好提示先去建索引同步数据避免后面查询直接报错。这个兜底做得好。关键词处理只有当keyword非空时才向must子句添加一个multi_match查询。multi_match支持在多个字段中同时搜索job_name^3、job_desc^2中的^符号表示权重系数——职位名称字段的匹配权重是职位描述字段的 1.5 倍3:2这使得名称匹配的结果在排序中更具优势。best_fields类型表示取所有字段中得分最高的那个字段作为最终得分operator:and要求查询词条必须全部出现在文档中才能匹配。筛选条件所有筛选条件城市、学历、经验、行业、公司规模、融资阶段、状态都放入filter子句。filter与must的关键区别在于filter不参与相关性评分计算且结果可缓存对于纯过滤条件使用filter比must性能更优。每个条件都使用term查询进行精确匹配。查询组装根据条件动态构建bool查询如果存在must子句则添加must如果存在filter子句则添加filter。如果两者都不存在即用户未提供任何搜索关键词和筛选条件则使用match_all查询匹配所有文档例如用户直接浏览职位列表时。排序sortscore 且有关键词时先按相关度 _score、再按发布时间兜底其他情况没关键词、或者指定按时间一律按 publish_time 倒序missing:last 是说没有发布时间的排最后不会报错。分页from (page-1)*page_size 算偏移size 是每页条数ES 的分页就是这么用的。瘦身source_fields 只挑列表卡片要展示的字段ES 就只返回这些_source 截断数据量大的时候省不少网络带宽。结果解析ES 7 的 total 是个字典 {“value”: n, “relation”: “eq”}代码用 isinstance 判断了两种返回格式取 value 当 total_count再算总页数。每条命中把 _source 当一条记录顺手把 _score 也塞进去方便以后调相关度时排查。最后 lists page_info 一起返回前端分页直接拿着用。求职者端-文件上传基于附件简历的优化求职者这边要能传文件、看职位、投简历。下面四块上传、详情、投递串起来就是简历整理好存 OSS → 看职位 → 投的时候把默认简历拷一份给企业。fileupload_router.post(“/upload”, summary“上传文件”)async def file_upload(file: UploadFile File(…, description“上传文件”)):file_content await file.read()filename file.filenameoss AliyunOSSTool()is_success, success_res oss.upload_single_file(file_content,filename,oss_path“job_seeker_avatar/”)if is_success: logger.info(上传成功) snowflake SnowflakeSingleton.get_instance(worker_id1) file_id snowflake.get_id() # 优化动态替换域名避免硬编码 source_access_url success_res[access_url] #https://baimugudu.oss-cn-beijing.aliyuncs.com/job_seeker_avatar/5a17279e-赵丽丽_20250923_153314.docx #job_seeker_avatar/5a17279e-赵丽丽_20250923_153314.docx source_key source_access_url.replace(fhttps://{oss.bucket_name}.{oss.endpoint}/, ) logger.info(fsource_key:{source_key}) # 修复提取纯文件名避免路径重复 source_filename os.path.basename(source_key) # 只保留文件名如e3d73f50-入职通知书.docx 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}) logger.info(fcopy_data:{copy_data}) return { code: 1, message: 上传成功, data: success_res }Lobster AIpython运行这是附件简历上传的优化版。POST /upload 收一个文件先照旧传到 OSS 的 job_seeker_avatar/ 目录。上传成功后多了一步整理文件用雪花算法SnowflakeSingleton生成全局唯一的 file_id给文件一个不会撞车的名字。关键优化是动态拼路径、不写死域名从返回的 access_url 里把 https://bucket.endpoint/ 前缀抠掉得到纯 OSS keysource_key再用 os.path.basename 只取文件名避免把整段路径又拼进新路径导致重复。然后调 copy_single_file 在 OSS 服务端把文件从 job_seeker_avatar/ 拷到 job_seeker_avatar/copy/ 子目录新文件名是 {file_id}-{原文件名}overwriteFalse 表示同名不覆盖。求职者端-职业详情job_router.get(“/detail/{id}”, summary“职位详情”)async def get_job_detail(idPath(…, description“职位ID”)):resawait JobService.get_job_detail(id)return {“code”: 1,“message”: “查询成功”,“data”: res}staticmethod async def get_job_detail(id: int): jobawait Job.get_or_none(idid) enterprise_id job.enterprise_id enterprise await Enterprise.get_or_none(identerprise_id) info{ job_info:job, enterprise_info:enterprise } return infoLobster AIpython运行按 id 取职位再按职位上的 enterprise_id 把发布企业捞出来拼成 {job_info, enterprise_info} 返回。求职者点开职位就能同时看到职位内容和公司信息。求职者端-投递简历job_router.post(“/resume_submission”, summary“职位投递”)async def resume_submission(job_seeker_idDepends(get_job_seeker_info),job_id:intQuery(…, description“职位ID”)):await JobService.resume_submission(job_seeker_id,job_id)return {“code”: 1,“message”: “投递成功”}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) # 没有默认简历时给出明确提示不再抛出 NoneType 错误 if not attachmentResume: raise Exception(请先上传并设置默认简历后再投递) file_urlattachmentResume.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_idsnowflake.get_id() # 修复提取纯文件名避免路径重复 source_filenameos.path.basename(source_key)# 只保留文件名如e3d73f50-入职通知书.docx target_key fjob_seeker_resume/{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}) logger.info(fcopy_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 ) return 1Lobster AIpython运行求职者投一个职位时不是重新传简历而是把他的默认附件简历从 OSS 里拷贝一份到 job_seeker_resume/ 目录作为这次投递的快照——这样求职者以后改了原简历企业看到的还是投递时的那份不会串。流程先找默认简历is_defaultTrue没有就明确报错请先上传并设置默认简历后再投递这比之前直接炸 NoneType 友好多了拿到原文件 URL抠出 OSS key雪花算法 生成新名服务端拷到 job_seeker_resume/{file_id}_{原文件名}最后在 ResumeSubmissionRecords 里记一条投递记录存拷贝后的地址、投递时间、状态已投递(SUBMITTED)。企业端-简历中心enterprise_router.get(“/resume_center”, summary“简历中心”, description“简历中心”)async def resume_center(team_idDepends(get_job_info)):resawait EnterpriseService.resume_center(team_id)return {“code”: 1,“message”: “查询成功”,“data”: res}staticmethod async def resume_center(team_id:int): jobsawait Job.filter(recruit_team_idteam_id) data_dict_list[] for job in jobs: resume_submission_records await ResumeSubmissionRecords.filter(job_idjob.id) for resume_submission_record in resume_submission_records: job_seeker_idresume_submission_record.job_seeker_id job_seekerawait JobSeeker.get_or_none(idjob_seeker_id) jobIIntentionawait JobIntention.get_or_none(job_seeker_idjob_seeker_id) workExperienceawait WorkExperience.filter(job_seeker_idjob_seeker_id).order_by(-entry_time).first() info_dict{ job_seeker_id: job_seeker.id, job_seeker: job_seeker, job_intention: jobIIntention, work_experience: workExperience, resume_status: resume_submission_record.resume_status, submit_time: resume_submission_record.submit_time, } data_dict_list.append(info_dict) return data_dict_listLobster AIpython运行两层循环先把这个团队发的所有职位捞出来再对每个职位把投递记录ResumeSubmissionRecords查出来每条记录去拼求职者的基本信息、求职意向JobIntention、最近一段工作经历WorkExperience 按入职时间倒序取第一条连同投递状态和投递时间塞进一个字典最后整成一个列表返回。企业 HR 在简历中心看到的谁投了哪个岗位、啥时候投的、目前啥状态、最近干过啥就是这么来的。总结求职者端 —— 文件上传、职位详情与投递、企业端简历中心文件上传附件简历优化版OSS 服务端拷贝 雪花算法重命名动态拼路径避免硬编码域名。职位详情按 id 取职位并关联企业信息一同返回。投递简历将求职者默认附件简历 OSS 拷贝为投递快照并落投递记录无默认简历时给出明确报错。企业端简历中心双层循环拼出团队收到的全部投递及其求职者信息。求职者端 —— ES 职位搜索/search作为 Elasticsearch 章节收尾闭合「建索引→灌数据→检索」链路multi_match 多字段加权检索、filter 精确过滤不算分、from/size 分页、_source 字段瘦身返回列表卡片。
