一、前言在BOSS直聘這類招聘平臺(tái)中企業(yè)管理是核心功能之一。企業(yè)用戶從注冊(cè)到審核通過需要經(jīng)過一系列完整的流程。本文將詳細(xì)介紹如何使用FastAPI Tortoise ORM Redis技術(shù)棧實(shí)現(xiàn)企業(yè)管理系統(tǒng)的后端核心功能包括企業(yè)注冊(cè)與認(rèn)證企業(yè)列表查詢與分頁(yè)企業(yè)詳情查看企業(yè)審核通過/拒絕企業(yè)登錄驗(yàn)證碼方式數(shù)據(jù)扁平化處理技巧二、數(shù)據(jù)庫(kù)模型設(shè)計(jì)2.1 核心數(shù)據(jù)表結(jié)構(gòu)企業(yè)管理涉及多張關(guān)聯(lián)表主要包括表名說明t_enterprise企業(yè)主表存儲(chǔ)企業(yè)基本狀態(tài)t_enterprise_info企業(yè)詳細(xì)信息表t_enterprise_qualification企業(yè)資質(zhì)信息表t_enterprise_review企業(yè)審核記錄表2.2 關(guān)鍵枚舉類型class AccountStatus(str, Enum): NORMAL normal # 正常狀態(tài) PENDING_AUDIT pending # 待審核 FROZEN frozen # 已凍結(jié) class BlackListStatus(int, Enum): NOT_BANNED 0 # 未被拉黑 BANNED 1 # 已被拉黑 class AuditType(str, Enum): NEW_ENTERPRISE_AUTH new # 新企業(yè)認(rèn)證 ANNUAL_REVIEW annual # 年度審核 class AuthType(str, Enum): ENTERPRISE_LICENSE_AUTH license # 營(yíng)業(yè)執(zhí)照認(rèn)證三、企業(yè)注冊(cè)與認(rèn)證流程3.1 企業(yè)注冊(cè)接口企業(yè)注冊(cè)是整個(gè)流程的起點(diǎn)用戶提交企業(yè)信息和相關(guān)資質(zhì)文件后系統(tǒng)會(huì)自動(dòng)創(chuàng)建審核記錄。接口定義enterprise_router.post(/save, summary保存企業(yè)信息) 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 核心注冊(cè)邏輯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. 創(chuàng)建企業(yè)主記錄 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. 創(chuàng)建企業(yè)詳細(xì)信息 await EnterpriseInfo.create( enterprise_nameform.enterprise_name, unified_social_credit_codeform.unified_social_credit_code, legal_representativeform.legal_representative, enterprise_identerprise.id, # ... 其他字段 ) # 3. 上傳并保存資質(zhì)文件 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. 創(chuàng)建企業(yè)資質(zhì)記錄 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. 自動(dòng)創(chuàng)建審核記錄 ? 關(guān)鍵點(diǎn) await EnterpriseReview.create( enterprise_identerprise.id, review_resultNone, remark企業(yè)注冊(cè)提交認(rèn)證申請(qǐng)等待審核 )3.3 流程說明用戶提交注冊(cè) → 開啟數(shù)據(jù)庫(kù)事務(wù) ├── 創(chuàng)建企業(yè)主記錄狀態(tài)待審核 ├── 創(chuàng)建企業(yè)詳細(xì)信息 ├── 上傳資質(zhì)文件到OSS ├── 創(chuàng)建企業(yè)資質(zhì)記錄 └── 創(chuàng)建審核記錄review_result None → 提交事務(wù)四、企業(yè)列表查詢與分頁(yè)4.1 列表接口實(shí)現(xiàn)enterprise_router.get(/list, summary查詢企業(yè)列表) 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 列表查詢核心邏輯關(guān)鍵點(diǎn)數(shù)據(jù)扁平化處理staticmethod async def select_enterprise_list(page, page_size, enterprise_name, submit_time_start, submit_time_end): # 1. 構(gòu)建查詢條件 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. 分頁(yè)計(jì)算 total_count await query_enterprise.count() total_page math.ceil(total_count / page_size) offset (page - 1) * page_size # 3. 執(zhí)行分頁(yè)查詢 query_enterprise await query_enterprise.offset(offset).limit(page_size) # 4. 數(shù)據(jù)扁平化處理 ? 關(guān)鍵技巧 enterprise_list [] for enterprise in query_enterprise: enterprise_id enterprise.id # 查詢關(guān)聯(lián)數(shù)據(jù) 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 ) # 處理審核狀態(tài) 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 # 獲取行業(yè)名稱 industry_name if enterpriseinfo and enterpriseinfo.industry: industry_name enterpriseinfo.industry.name # 構(gòu)建扁平數(shù)據(jù)結(jié)構(gòu) 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 數(shù)據(jù)扁平化技巧為什么需要扁平化原始數(shù)據(jù)結(jié)構(gòu)是嵌套對(duì)象{ enterprise: {...}, enterprise_info: {...}, enterprise_qualification: {...}, enterprise_review: {...} }前端表格組件el-table需要扁平結(jié)構(gòu){ id: 1, enterprise_name: 示例科技, legal_representative: 張三, audit_status: 0 }扁平化的好處簡(jiǎn)化前端模板代碼減少前端渲染計(jì)算便于數(shù)據(jù)導(dǎo)出五、企業(yè)詳情查看5.1 詳情接口enterprise_router.get(/detail/{enterprise_id}, summary查詢企業(yè)詳情) 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 詳情查詢實(shí)現(xiàn)staticmethod async def select_enterprise_by_id(enterprise_id: int): # 查詢企業(yè)主表 enterprise await Enterprise.get_or_none(identerprise_id) # 查詢?cè)敿?xì)信息預(yù)加載行業(yè)關(guān)聯(lián) enterpriseinfo await EnterpriseInfo.get_or_none( enterprise_identerprise_id ).prefetch_related(industry) # 查詢資質(zhì)信息 enterprise_qualification await EnterpriseQualification.get_or_none( enterprise_identerprise_id ) # 查詢審核記錄 enterprise_review await EnterpriseReview.get_or_none( enterprise_identerprise_id ) # 組裝扁平數(shù)據(jù) 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 , }六、企業(yè)審核流程6.1 審核接口enterprise_router.post(/review, summary企業(yè)審核) async def enterprise_review(enterpriseReviewCreateRequest: EnterpriseReviewCreateRequest): await EnterpriseService.enterprise_review(enterpriseReviewCreateRequest) return {code: 1, message: 審核完成}6.2 審核邏輯實(shí)現(xiàn)staticmethod async def enterprise_review(enterpriseReviewCreateRequest: EnterpriseReviewCreateRequest): enterprise_id enterpriseReviewCreateRequest.enterprise_id review_result enterpriseReviewCreateRequest.review_result review_reason enterpriseReviewCreateRequest.review_reason # 1. 查詢或創(chuàng)建審核記錄 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. 審核通過時(shí)更新企業(yè)狀態(tài) 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: 發(fā)送短信/郵件通知企業(yè)用戶6.3 審核狀態(tài)流轉(zhuǎn)企業(yè)注冊(cè) → 待審核 (review_result None) │ ├── 審核通過 → review_result 1 → account_status NORMAL │ └── 審核拒絕 → review_result 2 → account_status 保持 PENDING 記錄拒絕原因 review_reason七、企業(yè)登錄驗(yàn)證碼方式7.1 驗(yàn)證碼發(fā)送enterprise_router.post(/send_sms_code/{mobile}, summary發(fā)送驗(yàn)證碼) async def send_sms_code(mobile: str): res await EnterpriseService.send_sms_code(mobile) return {code: 1, message: 發(fā)送成功, data: res}7.2 驗(yàn)證碼登錄enterprise_router.post(/login, summary企業(yè)登錄) async def login(loginMobileRequest: LoginMobileRequest): res await EnterpriseService.login(loginMobileRequest) return {code: 1, message: 登錄成功, data: res}7.3 登錄邏輯實(shí)現(xiàn)staticmethod async def login(loginMobileRequest: LoginMobileRequest): mobile loginMobileRequest.mobile code loginMobileRequest.code # 1. 通過手機(jī)號(hào)查找企業(yè) enterprise_qualifications await EnterpriseQualification.filter( contact_phonemobile ) if not enterprise_qualifications: raise Exception(手機(jī)號(hào)不存在) for qualification in enterprise_qualifications: enterprise_id qualification.enterprise_id # 2. 檢查審核記錄 enterprise_review await EnterpriseReview.get_or_none( enterprise_identerprise_id ) # 處理沒有審核記錄的情況兼容老數(shù)據(jù) if enterprise_review is None: await EnterpriseReview.create( enterprise_identerprise_id, review_resultNone, remark自動(dòng)創(chuàng)建審核記錄 ) raise Exception(賬號(hào)未審核通過請(qǐng)等待審核) # 審核拒絕 if enterprise_review.review_result 2: raise Exception(f賬號(hào)審核未通過: {enterprise_review.review_reason}) # 審核通過驗(yàn)證驗(yàn)證碼 if enterprise_review.review_result