# BOSS直聘企业管理系统后端实现详解：从注册到审核的完整流程

一、前言在BOSS直聘这类招聘平台中企业管理是核心功能之一。企业用户从注册到审核通过需要经过一系列完整的流程。本文将详细介绍如何使用FastAPI Tortoise ORM Redis技术栈实现企业管理系统的后端核心功能包括企业注册与认证企业列表查询与分页企业详情查看企业审核通过/拒绝企业登录验证码方式数据扁平化处理技巧二、数据库模型设计2.1 核心数据表结构企业管理涉及多张关联表主要包括表名说明t_enterprise企业主表存储企业基本状态t_enterprise_info企业详细信息表t_enterprise_qualification企业资质信息表t_enterprise_review企业审核记录表2.2 关键枚举类型class AccountStatus(str, Enum): NORMAL normal # 正常状态 PENDING_AUDIT pending # 待审核 FROZEN frozen # 已冻结 class BlackListStatus(int, Enum): NOT_BANNED 0 # 未被拉黑 BANNED 1 # 已被拉黑 class AuditType(str, Enum): NEW_ENTERPRISE_AUTH new # 新企业认证 ANNUAL_REVIEW annual # 年度审核 class AuthType(str, Enum): ENTERPRISE_LICENSE_AUTH license # 营业执照认证三、企业注册与认证流程3.1 企业注册接口企业注册是整个流程的起点用户提交企业信息和相关资质文件后系统会自动创建审核记录。接口定义enterprise_router.post(/save, summary保存企业信息) async def saveEnterpriseInfo( form: EnterpriseCreateRequest Depends(get_enterprise_form), business_license_file: UploadFile File(None), legal_id_front_file: UploadFile File(None), legal_id_back_file: UploadFile File(None), ): await EnterpriseService.saveEnterpriseInfo( form, business_license_file, legal_id_front_file, legal_id_back_file ) return {code: 1, message: 保存成功}3.2 核心注册逻辑staticmethod async def saveEnterpriseInfo( form: EnterpriseCreateRequest, business_license_file: UploadFile, legal_id_front_file: UploadFile, legal_id_back_file: UploadFile, ): async with in_transaction() as conn: # 1. 创建企业主记录 enterprise await Enterprise.create( enterprise_nameform.enterprise_name, enterprise_codestr(uuid.uuid4()), account_statusAccountStatus.PENDING_AUDIT, blacklist_statusBlackListStatus.NOT_BANNED, audit_typeAuditType.NEW_ENTERPRISE_AUTH, submit_timenow(), auth_typeAuthType.ENTERPRISE_LICENSE_AUTH ) # 2. 创建企业详细信息 await EnterpriseInfo.create( enterprise_nameform.enterprise_name, unified_social_credit_codeform.unified_social_credit_code, legal_representativeform.legal_representative, enterprise_identerprise.id, # ... 其他字段 ) # 3. 上传并保存资质文件 success_res await AliyunOSSTool.upload_file(business_license_file) success_res2 await AliyunOSSTool.upload_file(legal_id_front_file) success_res3 await AliyunOSSTool.upload_file(legal_id_back_file) # 4. 创建企业资质记录 await EnterpriseQualification.create( enterprise_identerprise.id, business_license_urlsuccess_res[access_url], legal_id_front_urlsuccess_res2[access_url], legal_id_back_urlsuccess_res3[access_url], # ... 其他字段 ) # 5. 自动创建审核记录 ⭐ 关键点 await EnterpriseReview.create( enterprise_identerprise.id, review_resultNone, remark企业注册提交认证申请等待审核 )3.3 流程说明用户提交注册 → 开启数据库事务 ├── 创建企业主记录状态待审核 ├── 创建企业详细信息 ├── 上传资质文件到OSS ├── 创建企业资质记录 └── 创建审核记录review_result None → 提交事务四、企业列表查询与分页4.1 列表接口实现enterprise_router.get(/list, summary查询企业列表) async def select_enterprise_list( page: int Query(1, ge1), page_size: int Query(10, ge10, le50), enterprise_name: str Query(None), submit_time_start: str Query(None), submit_time_end: str Query(None), ): res await EnterpriseService.select_enterprise_list( page, page_size, enterprise_name, submit_time_start, submit_time_end ) return {code: 1, message: 查询成功, data: res}4.2 列表查询核心逻辑关键点数据扁平化处理staticmethod async def select_enterprise_list(page, page_size, enterprise_name, submit_time_start, submit_time_end): # 1. 构建查询条件 query_enterprise Enterprise.all() if enterprise_name: query_enterprise query_enterprise.filter( enterprise_name__icontainsenterprise_name ) if submit_time_start: query_enterprise query_enterprise.filter( submit_time__gtesubmit_time_start ) if submit_time_end: query_enterprise query_enterprise.filter( submit_time__ltesubmit_time_end ) # 2. 分页计算 total_count await query_enterprise.count() total_page math.ceil(total_count / page_size) offset (page - 1) * page_size # 3. 执行分页查询 query_enterprise await query_enterprise.offset(offset).limit(page_size) # 4. 数据扁平化处理 ⭐ 关键技巧 enterprise_list [] for enterprise in query_enterprise: enterprise_id enterprise.id # 查询关联数据 enterpriseinfo await EnterpriseInfo.get_or_none( enterprise_identerprise_id ).prefetch_related(industry) enterprise_qualification await EnterpriseQualification.get_or_none( enterprise_identerprise_id ) enterprise_review await EnterpriseReview.get_or_none( enterprise_identerprise_id ) # 处理审核状态 audit_status 0 reject_reason None if enterprise_review: audit_status enterprise_review.review_result if enterprise_review.review_result else 0 reject_reason enterprise_review.review_reason # 获取行业名称 industry_name if enterpriseinfo and enterpriseinfo.industry: industry_name enterpriseinfo.industry.name # 构建扁平数据结构 item { id: enterprise.id, enterprise_name: enterprise.enterprise_name, legal_representative: enterpriseinfo.legal_representative if enterpriseinfo else , contact_name: enterprise_qualification.contact_name if enterprise_qualification else , contact_phone: enterprise_qualification.contact_phone if enterprise_qualification else , industry: industry_name, submit_time: enterprise.submit_time.isoformat() if enterprise.submit_time else , audit_status: audit_status, reject_reason: reject_reason, business_license_url: enterprise_qualification.business_license_url if enterprise_qualification else , } enterprise_list.append(item) return { list: enterprise_list, total: total_count, total_page: total_page, page: page, page_size: page_size }4.3 数据扁平化技巧为什么需要扁平化原始数据结构是嵌套对象{ enterprise: {...}, enterprise_info: {...}, enterprise_qualification: {...}, enterprise_review: {...} }前端表格组件el-table需要扁平结构{ id: 1, enterprise_name: 示例科技, legal_representative: 张三, audit_status: 0 }扁平化的好处简化前端模板代码减少前端渲染计算便于数据导出五、企业详情查看5.1 详情接口enterprise_router.get(/detail/{enterprise_id}, summary查询企业详情) async def select_enterprise_by_id(enterprise_id: int): res await EnterpriseService.select_enterprise_by_id(enterprise_id) return {code: 1, message: 查询成功, data: res}5.2 详情查询实现staticmethod async def select_enterprise_by_id(enterprise_id: int): # 查询企业主表 enterprise await Enterprise.get_or_none(identerprise_id) # 查询详细信息预加载行业关联 enterpriseinfo await EnterpriseInfo.get_or_none( enterprise_identerprise_id ).prefetch_related(industry) # 查询资质信息 enterprise_qualification await EnterpriseQualification.get_or_none( enterprise_identerprise_id ) # 查询审核记录 enterprise_review await EnterpriseReview.get_or_none( enterprise_identerprise_id ) # 组装扁平数据 return { id: enterprise.id if enterprise else None, enterprise_name: enterprise.enterprise_name if enterprise else , unified_social_credit_code: enterpriseinfo.unified_social_credit_code if enterpriseinfo else , legal_representative: enterpriseinfo.legal_representative if enterpriseinfo else , registered_capital: enterpriseinfo.registered_capital if enterpriseinfo else , establish_date: str(enterpriseinfo.establish_date) if enterpriseinfo else , industry: enterpriseinfo.industry.name if enterpriseinfo and enterpriseinfo.industry else , contact_name: enterprise_qualification.contact_name if enterprise_qualification else , contact_phone: enterprise_qualification.contact_phone if enterprise_qualification else , audit_status: enterprise_review.review_result if enterprise_review and enterprise_review.review_result else 0, reject_reason: enterprise_review.review_reason if enterprise_review else None, business_license_url: enterprise_qualification.business_license_url if enterprise_qualification else , }六、企业审核流程6.1 审核接口enterprise_router.post(/review, summary企业审核) async def enterprise_review(enterpriseReviewCreateRequest: EnterpriseReviewCreateRequest): await EnterpriseService.enterprise_review(enterpriseReviewCreateRequest) return {code: 1, message: 审核完成}6.2 审核逻辑实现staticmethod async def enterprise_review(enterpriseReviewCreateRequest: EnterpriseReviewCreateRequest): enterprise_id enterpriseReviewCreateRequest.enterprise_id review_result enterpriseReviewCreateRequest.review_result review_reason enterpriseReviewCreateRequest.review_reason # 1. 查询或创建审核记录 enterprise_review await EnterpriseReview.get_or_none( enterprise_identerprise_id ) if enterprise_review is None: # 新建审核记录 await EnterpriseReview.create( enterprise_identerprise_id, review_resultreview_result, review_reasonreview_reason, review_timenow() ) else: # 更新已有审核记录 enterprise_review.review_result review_result enterprise_review.review_reason review_reason enterprise_review.review_time now() await enterprise_review.save() # 2. 审核通过时更新企业状态 if review_result 1: # 通过 enterprise await Enterprise.get_or_none(identerprise_id) enterprise.account_status AccountStatus.NORMAL await enterprise.save() elif review_result 2: # 拒绝 # 可选添加拒绝处理逻辑 pass # TODO: 发送短信/邮件通知企业用户6.3 审核状态流转企业注册 → 待审核 (review_result None) │ ├── 审核通过 → review_result 1 → account_status NORMAL │ └── 审核拒绝 → review_result 2 → account_status 保持 PENDING 记录拒绝原因 review_reason七、企业登录验证码方式7.1 验证码发送enterprise_router.post(/send_sms_code/{mobile}, summary发送验证码) async def send_sms_code(mobile: str): res await EnterpriseService.send_sms_code(mobile) return {code: 1, message: 发送成功, data: res}7.2 验证码登录enterprise_router.post(/login, summary企业登录) async def login(loginMobileRequest: LoginMobileRequest): res await EnterpriseService.login(loginMobileRequest) return {code: 1, message: 登录成功, data: res}7.3 登录逻辑实现staticmethod async def login(loginMobileRequest: LoginMobileRequest): mobile loginMobileRequest.mobile code loginMobileRequest.code # 1. 通过手机号查找企业 enterprise_qualifications await EnterpriseQualification.filter( contact_phonemobile ) if not enterprise_qualifications: raise Exception(手机号不存在) for qualification in enterprise_qualifications: enterprise_id qualification.enterprise_id # 2. 检查审核记录 enterprise_review await EnterpriseReview.get_or_none( enterprise_identerprise_id ) # 处理没有审核记录的情况兼容老数据 if enterprise_review is None: await EnterpriseReview.create( enterprise_identerprise_id, review_resultNone, remark自动创建审核记录 ) raise Exception(账号未审核通过请等待审核) # 审核拒绝 if enterprise_review.review_result 2: raise Exception(f账号审核未通过: {enterprise_review.review_reason}) # 审核通过验证验证码 if enterprise_review.review_result