企业端-招聘团队企业过审之后得有人来发职位。下面这几块串起来就是审核通过时自动建招聘团队 → 团队成员凭手机号验证码登录 → 发职位 → 看自己的职位列表。staticmethod async def enterprise_review(enterpriseReviewCreateRequest: EnterpriseReviewCreateRequest): enterprise_identerpriseReviewCreateRequest.enterprise_id enterprise_reviewawait EnterpriseReview.get_or_none(enterprise_identerprise_id) if enterprise_review is None: await EnterpriseReview.create( enterprise_identerprise_id, review_resultenterpriseReviewCreateRequest.review_result, review_reasonenterpriseReviewCreateRequest.review_reason, remarkenterpriseReviewCreateRequest.remark, audit_timenow() ) else: enterprise_review.review_resultenterpriseReviewCreateRequest.review_result if enterpriseReviewCreateRequest.review_result else enterprise_review.review_result enterprise_review.review_reasonenterpriseReviewCreateRequest.review_reason if enterpriseReviewCreateRequest.review_reason else enterprise_review.review_reason enterprise_review.remarkenterpriseReviewCreateRequest.remark if enterpriseReviewCreateRequest.remark else enterprise_review.remark enterprise_review.audit_timenow() await enterprise_review.save() if enterpriseReviewCreateRequest.review_result1: enterpriseawait Enterprise.get_or_none(identerprise_id) enterprise.account_statusAccountStatus.NORMAL await enterprise.save() enterpriseQualificationawait EnterpriseQualification.get_or_none(enterprise_identerprise_id) await RecruitTeam.create( mobileenterpriseQualification.contact_phone, nameenterpriseQualification.contact_name, emailenterpriseQualification.contact_email, enterprise_identerprise_id, statusTeamMemberStatus.NORMAL ) status 审核通过 if enterpriseReviewCreateRequest.review_result 1 else 审核驳回 await EnterpriseReviewHistory.create( enterprise_identerprise_id, operationstatus, statusstatus, operator系统自动审核, remarkenterpriseReviewCreateRequest.remark, created_atnow() )这是前面管理者端审核的升级版在建/改审核记录 通过时放开账号的基础上多做了两件事审核一通过review_result 1除了把企业account_status置为NORMAL还会从企业资质里拽出联系人的手机、姓名、邮箱自动建一条RecruitTeam招聘团队记录状态NORMAL。等于企业过审的同一刻那个联系人就自动成了招聘团队第一个成员——后面团队登录就是对着这张表验的。不管通过还是驳回末尾都往EnterpriseReviewHistory写一条操作记录操作人写死系统自动审核相当于留个审计痕以后查谁在啥时候改了审核状态有处可查。字段名从之前的review_time换成了audit_time其余逻辑一致。企业端-登录async def team_login(loginMobileRequest: LoginMobileRequest): recruit_teamawait RecruitTeam.get_or_none(mobileloginMobileRequest.mobile, statusTeamMemberStatus.NORMAL, is_deletedDeleteStatus.NOT_DELETED ) if recruit_team is None: raise Exception(手机号不存在或账号状态异常) key fboss-api:enterprise-login:sms:{loginMobileRequest.mobile} redis_code redis_client.get(key) if redis_code is None: raise Exception(验证码已过期) if redis_code ! loginMobileRequest.code: raise Exception(验证码错误) access_token, refresh_token create_tokens(str(recruit_team.id), loginMobileRequest.mobile) redis_client.delete(key) return { enterprise_access_token: access_token, enterprise_refresh_token: refresh_token, } enterprise_router.post(/login, summary企业登录, description企业登录) async def login(loginMobileRequest: LoginMobileRequest): resawait EnterpriseService.team_login(loginMobileRequest) return { code: 1, message: 登录成功, data: res }团队登录和企业登录第四节那个套路一样——都是手机号 短信验证码区别在对着哪张表验、验的是谁用手机号去RecruitTeam查而且顺手卡了statusNORMAL和is_deletedNOT_DELETED——被禁用或已删除的成员直接登不进来。查不到抛手机号不存在或账号状态异常这里用的是get_or_noneis None判断是对的不像之前企业登录那个filter永远非 None 的坑。Redis 取验证码过期或不对分别报错都过了就用recruit_team.id当 subject 发 token发完把验证码删掉。企业端-发布职位job_router.post(/save, summary保存职位信息, description保存职位信息) async def saveJob(jobCreateRequest: JobCreateRequest,team_idDepends(get_job_info)): await JobService.saveJob(jobCreateRequest,team_id) return { code: 1, message: 保存成功 } class JobCreateRequest(BaseModel): job_name: str Field(..., description职位名称) # 使用枚举替换原来int类型dept_id department_id: int Field(..., description所属部门) work_location: str Field(..., description工作地点) min_salary: str Field(..., description最低薪资支持「面议」) max_salary: str | None Field(None, description最高薪资) salary_times: str | None Field(None, description几薪) exp_require: str Field(..., description经验要求) edu_require: str Field(..., description学历要求) gender_require: str Field(..., description性别要求) recruit_num: str Field(..., description招聘人数) job_tags: list[str] | None Field(None, description职位标签示例[五险一金,年终奖]) job_desc: str Field(..., description职位描述) duty_require: str Field(..., description任职要求) status: int Field(..., description0:草稿,1:招聘中,2:暂停招聘,3:已关闭) staticmethod async def saveJob(jobCreateRequest: JobCreateRequest,team_id:int): recruitteam await RecruitTeam.get_or_none(idteam_id) await Job.create(**jobCreateRequest.dict(),publish_timenow(), enterprise_idrecruitteam.enterprise_id,recruit_team_idteam_id)api层POST /save发职位。jobCreateRequest装着职位信息team_id不从前端点而是Depends(get_job_info)从登录态里解出来——保证谁发的职位就挂在谁的团队下前端伪造不了归属。schema层职位的数据结构。几个值得说的点薪资是str而不是数字因为要兼容面议这种非数字值max_salary、salary_times、job_tags都可空job_tags是字符串数组存标签列表status用 int 表示职位状态——0 草稿、1 招聘中、2 暂停招聘、3 已关闭。注释里那句用枚举替换原来 int 类型 dept_id是个优化方向部门 id 直接写 int 容易传错值后面用枚举收口会更稳。service层先按team_id把招聘团队捞出来——目的是拿到它归属的enterprise_id。然后一条Job.create搞定用**jobCreateRequest.dict()把 schema 里的字段全展开进职位表publish_time填当前时间企业 id 从团队带过来团队 id 也记上。这么一来职位天然就和企业、团队绑死了发布人没法把职位挂到别人的企业底下。企业端-职位列表job_router.get(/list, summary职位列表) async def selectJobList( page: int Query(1, description页码), page_size: int Query(10, description每页数量), keywords: str | None Query(None, description搜索关键词), status: str | None Query(None, description职位状态), cityname: str | None Query(None, description城市名称), team_idDepends(get_job_info) ): resawait JobService.selectJobList(page,page_size,keywords,status,cityname,team_id) return { code: 1, message: 查询成功, data: res } staticmethod async def selectJobList( page: int, page_size: int, keywords: str | None, status: str | None, cityname: str | None, team_id: int ): queryJob.filter(recruit_team_idteam_id) if keywords: queryquery.filter(job_name__icontainskeywords) if status: queryquery.filter(statusstatus) if cityname: queryquery.filter(citynamecityname) total_countawait query.count() total_pagemath.ceil(total_count/page_size) job_query_setawait query.offset((page-1)*page_size).limit(page_size).all() return { total_count: total_count, total_page: total_page, job_list: job_query_set, page: page, page_size: page_size }先filter(recruit_team_idteam_id)把范围锁死——一个团队只能看自己发的职位数据隔离在这一层就做了。然后条件叠加keywords走job_name__icontains忽略大小写模糊匹配status、cityname是精确匹配。最后跟前面列表接口一样的套路算总数、算总页数、offsetlimit 分页把列表和分页信息一起返回。ES职位搜索使用ES原因职位多了之后光靠数据库 LIKE 查中文又慢又不智能所以接了 ES 做全文检索。下面三块客户端怎么建、索引怎么定义、数据怎么灌进去。ES_HOST os.getenv(ES_HOST, http://localhost:9200) es_client: AsyncElasticsearch | None None async def get_es_client() - AsyncElasticsearch: 获取 ES 异步客户端单例 global es_client if es_client is None: es_client AsyncElasticsearch(ES_HOST) logger.info(ES 客户端初始化成功) return es_client async def close_es_client(): 关闭 ES 客户端连接 global es_client if es_client is not None: await es_client.close() es_client Noneasync def es_client_depend() - AsyncElasticsearch: return await get_es_client()ES 客户端做成单例。get_es_client第一次调才真正连之后全局复用同一个AsyncElasticsearch实例——避免每次请求都重连省资源。close_es_client在应用关闭时调关连接、置空。es_client_depend专门给 FastAPI 的Depends用路由里写Depends(es_client_depend)就能拿到那个单例不用自己管连接生命周期。BOSS_JOB_INDEX_NAME boss_job_index es_data_router.post(/create-index, summary创建索引) async def create_index(es_client: AsyncElasticsearch Depends(es_client_depend)): mappings{ properties:{ job_id: { type: long }, job_name: { type: text, analyzer: ik_max_word }, department_id: { type: long }, work_location: { type: keyword }, min_salary: { type: keyword }, max_salary: { type: keyword }, salary_times: { type: keyword }, edu_require: { type: keyword, }, exp_require: { type: keyword }, gender_require: { type: keyword }, recruit_num: { type: long }, job_tags: { type: keyword }, job_desc: { type: text, analyzer: ik_max_word }, duty_require: { type: text, analyzer: ik_max_word }, status: { type: integer }, publish_time: { type: date, # format: yyyy-MM-dd HH:mm:ss }, enterprise_id: { type: long }, recruit_team_id: { type: long }, enterprise_name: { type: text, analyzer: ik_max_word }, enterprise_code: { type: keyword }, enterprise_city_id: { type: long }, enterprise_city_name: { type: keyword }, enterprise_account_status: { type: integer }, enterprise_create_time: { type: date, # format: yyyy-MM-dd HH:mm:ss }, enterprise_auth_time: { type: date, # format: yyyy-MM-dd HH:mm:ss }, enterprise_auth_type: { type: integer }, enterprise_risk_level: { type: integer }, enterprise_blacklist_status: { type: integer }, enterprise_complaint_count: { type: integer }, enterprise_company_website: { type: keyword }, enterprise_email: { type: keyword }, enterprise_audit_type: { type: integer }, enterprise_submit_time: { type: date, # format: yyyy-MM-dd HH:mm:ss }, enterpriseInfo_unified_social_credit_code: { type: keyword }, enterpriseInfo_legal_representative: { type: keyword }, enterpriseInfo_registered_capital: { type: keyword}, enterpriseInfo_establish_date: { type: date, # format: yyyy-MM-dd HH:mm:ss }, enterpriseInfo_register_status: { type: integer }, enterpriseInfo_company_scale: { type: keyword }, enterpriseInfo_financing_stage: { type: keyword }, enterpriseInfo_headquarters_address: { type: keyword }, industry_id: { type: long }, industry_name: { type: text, analyzer: ik_max_word } } } if await es_client.indices.exists(indexBOSS_JOB_INDEX_NAME): return 索引已存在 await es_client.indices.create(indexBOSS_JOB_INDEX_NAME, mappingsmappings) return 索引创建成功 es_data_router.post(/insert-data, summary同步数据) async def insert_data(es_client: AsyncElasticsearch Depends(es_client_depend)): jobsawait Job.all() for job in jobs: enterprise_id job.enterprise_id enterprise await Enterprise.get_or_none(identerprise_id).prefetch_related(city) enterpriseInfoawait EnterpriseInfo.get_or_none(enterprise_identerprise_id).prefetch_related(industry) job_info{ job_id: job.id, job_name: job.job_name, department_id: job.department_id, work_location: job.work_location, min_salary: job.min_salary, max_salary: job.max_salary, salary_times: job.salary_times, edu_require: job.edu_require, exp_require: job.exp_require, gender_require: job.gender_require, recruit_num: job.recruit_num, job_tags: job.job_tags, job_desc: job.job_desc, duty_require: job.duty_require, status: job.status, publish_time: job.publish_time, enterprise_id: job.enterprise_id, recruit_team_id: job.recruit_team_id, enterprise_name: enterprise.enterprise_name, enterprise_code: enterprise.enterprise_code, # enterprise_city_id: enterprise.city.id, # enterprise_city_name: job.work_location, enterprise_account_status: enterprise.account_status, enterprise_create_time: enterprise.create_time, enterprise_auth_time: enterprise.auth_time, enterprise_auth_type: enterprise.auth_type, enterprise_risk_level: enterprise.risk_level, enterprise_blacklist_status: enterprise.blacklist_status, enterprise_complaint_count: enterprise.complaint_count, enterprise_company_website: enterprise.company_website, enterprise_email: enterprise.email, enterprise_audit_type: enterprise.audit_type, enterprise_submit_time: enterprise.submit_time, enterpriseInfo_unified_social_credit_code: enterpriseInfo.unified_social_credit_code, enterpriseInfo_legal_representative: enterpriseInfo.legal_representative, enterpriseInfo_registered_capital: enterpriseInfo.registered_capital, enterpriseInfo_establish_date: enterpriseInfo.establish_date, enterpriseInfo_register_status: enterpriseInfo.register_status, enterpriseInfo_company_scale: enterpriseInfo.company_scale, enterpriseInfo_financing_stage: enterpriseInfo.financing_stage, enterpriseInfo_headquarters_address: enterpriseInfo.headquarters_address, enterpriseInfo_business_scope: enterpriseInfo.business_scope, industry_id: enterpriseInfo.industry.id, industry_name: enterpriseInfo.industry.name, } await es_client.index(indexBOSS_JOB_INDEX_NAME, documentjob_info)创建索引定义职位索引的字段结构mapping。类型分两类别混要全文搜的中文文本——job_name、job_desc、duty_require、enterprise_name、industry_name——用textik_max_word分词器。ik_max_word是专门切中文的能把人力资源切成人力/资源/人力资源那样搜索才准ES 默认的英文分词器对中文是一字一切基本没法用。其余像薪资、状态、各种 code 用keyword/long/integer/date这些是精确匹配或排序、聚合用的不分词。同步数据把数据库里的职位全量同步进 ES相当于索引建好之后灌数据。逻辑是遍历所有Job每条都去把对应的Enterprise和EnterpriseInfo含行业查出来拼成一个扁平的job_info字典——把职位字段和企业字段揉一块儿存。这种宽表做法让搜索时不用再 JOIN 数据库直接就能按企业名、行业名筛职位。
