☰
Django档案管理系统:档号自动生成+RBAC权限+OCR全文检索
2026/10/15 16:02:17 网站建设 项目流程

简介:本资源是一套基于Delphi 6与SQL Server的档案管理信息系统(AMIS)完整开发源码包,面向高校计算机专业学生、桌面应用初学者及中小型单位信息化建设人员,解决传统档案手工管理效率低、检索难、权限模糊、版本混乱等实际问题。压缩包共106个文件,含22个Pascal源码(.pas)、18个窗体设计文件(.dfm)、24个编译单元(.dcu)、2个SQL脚本(.sql)及可执行程序(.exe),清晰呈现Delphi三层架构实现:界面层(VCL组件布局)、业务逻辑层(档案录入/检索/权限控制)与数据访问层(ADO连接+SQL数据库操作)。另有配置文件(.cfg/.ini)、图标资源(.bmp/.gif)和备份数据库文件(.mdf/.ldf),便于本地快速部署调试。目前已有193人学习下载,读者可直接运行系统、分析模块间调用关系、复用核心功能代码(如多条件检索逻辑、用户角色权限模型),并参考完整的工程结构组织方式开展课程设计或轻量级项目开发。

1. 档案管理信息系统:不是电子扫描仪+Excel,而是让归档动作自动触发权限变更、借阅超期实时预警、全生命周期可追溯的业务中枢

很多团队一提“档案管理信息系统”,第一反应是买个扫描仪配个NAS,再用Excel管编号——结果三年后查一份2021年立项会纪要,得翻三台旧电脑、两个U盘、一个被遗忘的共享文件夹,最后靠某位老员工凭记忆在纸质目录本里翻出页码。这不是信息化,这是数字化掩护下的手工劳动升级版。真正的档案管理信息系统,核心不在“存”,而在“控”:它必须能感知业务动作(比如合同审批通过)、自动执行归档策略(生成唯一档号、设定密级、绑定责任人)、联动组织架构同步权限(法务部可查全部合同,销售部仅见本人经手部分),并在借阅超时48小时未归还时,向借阅人、部门负责人、档案管理员三方推送带截图的预警消息。它服务的对象不是档案室老师傅,而是业务系统里的项目经理、法务专员、HRBP——他们不关心元数据标准,只关心“我刚签完字的合同,为什么法务说还没归档?”“上个月借的员工档案,今天能续借吗?”。本文就从零搭建一个轻量但生产可用的档案管理信息系统最小闭环:支持结构化业务表单对接、自动档号生成、基于RBAC的细粒度权限控制、OCR识别后的文本可检索、以及借阅流程的完整状态机驱动。


2. 用Django+PostgreSQL构建档案核心模型:从实体设计到档号自动生成逻辑

档案系统不是文档仓库,它的数据模型必须承载业务语义。我们不抽象成“文件”“附件”这种泛化概念,而是直接建模为ArchiveRecord(档案记录)、BusinessEvent(业务事件)、Custodian(保管人)、BorrowingRequest(借阅申请)四类核心实体。其中ArchiveRecord是枢纽,它必须关联到具体业务动作(如ContractApprovalEvent),而非孤立存在。

2.1 档案记录模型:字段设计直击业务痛点

# models.py from django.db import models from django.contrib.auth.models import User class ArchiveRecord(models.Model): # 档号:由系统自动生成,格式为 YYYMMDD-DEPT-CODE-SEQ,如 20240520-LEGAL-CON-00123 archive_code = models.CharField(max_length=32, unique=True, db_index=True) # 关联业务事件(关键!打破档案与业务割裂) business_event = models.ForeignKey( 'BusinessEvent', on_delete=models.PROTECT, # 业务事件删除前必须先解绑或转移档案 related_name='archives' ) # 元数据:非技术字段,业务人员可填 title = models.CharField(max_length=255) description = models.TextField(blank=True) confidentiality_level = models.CharField( max_length=10, choices=[('PUBLIC', '公开'), ('INTERNAL', '内部'), ('CONFIDENTIAL', '机密'), ('TOP_SECRET', '绝密')], default='INTERNAL' ) # 归档动作本身 archived_at = models.DateTimeField(auto_now_add=True) archived_by = models.ForeignKey(User, on_delete=models.SET_NULL, null=True, related_name='archived_records') # 物理/数字载体标识(支持混合管理) digital_path = models.CharField(max_length=512, blank=True) # 如 /storage/contracts/2024/05/20/abc123.pdf physical_location = models.CharField(max_length=128, blank=True) # 如 A区-3排-5架-2层 # 状态机:归档后不是静止的,而是进入生命周期 STATUS_CHOICES = [ ('ARCHIVED', '已归档'), ('BORROWED', '借阅中'), ('RETURNED', '已归还'), ('DESTROYED', '已销毁'), ('TRANSFERRED', '已移交'), ] status = models.CharField(max_length=20, choices=STATUS_CHOICES, default='ARCHIVED') class Meta: ordering = ['-archived_at']

提示:business_event外键是灵魂。它让档案天然具备上下文——查一份采购合同档案,顺手就能看到对应的验收报告、付款凭证是否也已归档。这比任何后期打标签都可靠。

2.2 档号自动生成:避免人工输入错误,确保全局唯一与业务可读性

档号不是随机字符串,它是业务语言。我们采用“日期-部门-类型-序号”四级结构,序号按日重置,避免跨年混乱:

# utils/archive_code_generator.py from datetime import datetime from django.db import transaction from .models import ArchiveRecord def generate_archive_code(department_code: str, record_type: str) -> str: """ 生成档号:YYYYMMDD-DEPT-TYPE-NNN department_code: 部门简码,如 'LEGAL', 'HR', 'FIN' record_type: 档案类型简码,如 'CON' (合同), 'EMP' (员工), 'PROJ' (项目) """ today = datetime.now() date_part = today.strftime("%Y%m%d") base_code = f"{date_part}-{department_code}-{record_type}-" # 在当日内查找最大序号,原子性递增 with transaction.atomic(): last_record = ArchiveRecord.objects.filter( archive_code__startswith=base_code ).order_by('-archive_code').first() if last_record: # 提取末尾数字,如 '20240520-LEGAL-CON-00123' → 123 try: current_num = int(last_record.archive_code.split('-')[-1]) next_num = current_num + 1 except (ValueError, IndexError): next_num = 1 else: next_num = 1 return f"{base_code}{next_num:05d}" # 补零至5位,如 00123 # 使用示例(在视图或信号中调用) # new_code = generate_archive_code('HR', 'EMP')

参数说明:department_code和record_type必须来自预设枚举或字典表,禁止自由输入。next_num:05d保证序号长度固定,排序时不会出现0012 < 002的错乱。该函数在事务内执行,杜绝并发场景下生成重复档号。

2.3 业务事件模型:把档案锚定在真实工作流上

# models.py class BusinessEvent(models.Model): """所有触发归档的业务动作基类""" EVENT_TYPES = [ ('CONTRACT_APPROVAL', '合同审批完成'), ('EMPLOYEE_ONBOARDING', '员工入职'), ('PROJECT_COMPLETION', '项目结项'), ('ASSET_PURCHASE', '资产采购验收'), ] event_type = models.CharField(max_length=32, choices=EVENT_TYPES) occurred_at = models.DateTimeField() triggered_by = models.ForeignKey(User, on_delete=models.SET_NULL, null=True) # 业务系统ID(关键!用于与OA/ERP等系统对接) external_system_id = models.CharField(max_length=64, blank=True, db_index=True) external_ref_url = models.URLField(blank=True) # 指向OA中该审批单的链接 # 业务数据快照(JSONField,存关键字段,避免跨库查询) snapshot_data = models.JSONField(default=dict) # 如 {"contract_no": "CT2024-001", "party": "XX公司", "amount": 120000} def __str__(self): return f"[{self.get_event_type_display()}] {self.snapshot_data.get('contract_no', '未知')}" # 示例子类(可选,用多表继承或泛型关系实现) class ContractApprovalEvent(BusinessEvent): contract_number = models.CharField(max_length=64) sign_date = models.DateField() parties = models.TextField() # 签约方列表

逻辑说明:external_system_id是系统集成的命脉。当OA系统推送一条event_type=CONTRACT_APPROVAL且external_system_id=OA-2024-7890的事件时,档案系统创建ArchiveRecord并自动填充archive_code、title(取自snapshot_data['contract_no'])、description。业务人员永远不需要手动输入合同号——它来自源头。


3. 基于RBAC的细粒度权限控制:让法务看全部合同,销售只看自己的

档案敏感,权限不能粗放。我们不用Django默认的is_staff或简单分组,而是构建三层权限模型:角色(Role)→ 权限集(PermissionSet)→ 档案范围(Scope)。最终效果:法务总监能看到所有CONFIDENTIAL及以下级别合同;销售经理只能看到自己团队成员经手的INTERNAL合同;实习生只能查看PUBLIC档案。

3.1 权限模型设计:解耦角色与数据范围

# models.py class PermissionSet(models.Model): """权限集:定义一组操作能力,如 'view_contract', 'borrow_contract', 'destroy_contract'""" name = models.CharField(max_length=64, unique=True) codename = models.SlugField(unique=True) # 如 'view_confidential_contract' description = models.TextField() class Role(models.Model): """角色:绑定权限集,并可指定数据范围规则""" name = models.CharField(max_length=64, unique=True) permission_sets = models.ManyToManyField(PermissionSet, blank=True) # 范围规则:JSON存储,定义该角色能访问哪些档案 # 格式:{"confidentiality_level__lte": "CONFIDENTIAL", "business_event__event_type": "CONTRACT_APPROVAL"} scope_rules = models.JSONField(default=dict, blank=True) class UserRoleAssignment(models.Model): """用户-角色分配:支持一人多角色""" user = models.ForeignKey(User, on_delete=models.CASCADE) role = models.ForeignKey(Role, on_delete=models.CASCADE) assigned_at = models.DateTimeField(auto_now_add=True)

3.2 查询优化:用数据库原生WHERE条件替代Python循环过滤

权限校验若在应用层遍历所有档案,性能必崩。我们必须将scope_rules编译为SQLWHERE子句,在数据库层面过滤:

# utils/permission_filter.py import json from django.db.models import Q from .models import ArchiveRecord def build_scope_q_object(scope_rules: dict) -> Q: """ 将 scope_rules 字典转换为 Django Q 对象 支持嵌套字段,如 'business_event__event_type' """ q = Q() for key, value in scope_rules.items(): # 处理 __lte, __gte 等查找符 if '__' in key and key.split('__')[-1] in ['lte', 'gte', 'lt', 'gt', 'in', 'icontains']: field_name = '__'.join(key.split('__')[:-1]) lookup = key.split('__')[-1] q &= Q(**{f"{field_name}__{lookup}": value}) else: q &= Q(**{f"{key}": value}) return q # 在视图中使用(高效!) def get_user_archives(user: User): # 获取用户所有角色的 scope_rules 并合并(OR关系) user_roles = UserRoleAssignment.objects.filter(user=user).select_related('role') all_q = Q() for assignment in user_roles: rules = assignment.role.scope_rules if rules: all_q |= build_scope_q_object(rules) # 一次数据库查询完成过滤 return ArchiveRecord.objects.filter(all_q).distinct() # 示例:法务角色 scope_rules = {"confidentiality_level__lte": "CONFIDENTIAL"} # 生成 Q(confidentiality_level__lte="CONFIDENTIAL")

参数说明:scope_rules必须是严格JSON序列化的字典,键名需对应模型字段(支持双下划线跨表),值类型需匹配字段类型(如日期传"2024-01-01",整数传123)。前端配置角色时,提供下拉菜单选择字段和操作符,禁止自由输入键名。

3.3 借阅审批流中的动态权限:借阅人临时获得“查看原文”权限

借阅不是简单下载,而是状态变更。当BorrowingRequest状态变为APPROVED时,系统需临时授予借阅人对目标档案的view_original_content权限(该权限默认关闭),且该授权在RETURNED或EXPIRED时自动撤销:

# models.py class BorrowingRequest(models.Model): STATUS_CHOICES = [ ('PENDING', '待审批'), ('APPROVED', '已批准'), ('REJECTED', '已拒绝'), ('RETURNED', '已归还'), ('EXPIRED', '已超期'), ] status = models.CharField(max_length=20, choices=STATUS_CHOICES, default='PENDING') archive_record = models.ForeignKey(ArchiveRecord, on_delete=models.CASCADE) borrower = models.ForeignKey(User, on_delete=models.CASCADE, related_name='borrowing_requests') due_date = models.DateTimeField() # 应归还时间 def save(self, *args, **kwargs): super().save(*args, **kwargs) # 状态变更时触发权限更新 if self.status == 'APPROVED': self.grant_temporary_view_permission() elif self.status in ['RETURNED', 'EXPIRED', 'REJECTED']: self.revoke_temporary_view_permission() def grant_temporary_view_permission(self): # 创建临时权限记录(不修改Role,避免污染长期权限) TemporaryPermission.objects.get_or_create( user=self.borrower, archive_record=self.archive_record, defaults={'granted_at': timezone.now(), 'expires_at': self.due_date + timedelta(days=7)} ) def revoke_temporary_view_permission(self): TemporaryPermission.objects.filter( user=self.borrower, archive_record=self.archive_record, is_active=True ).update(is_active=False) class TemporaryPermission(models.Model): user = models.ForeignKey(User, on_delete=models.CASCADE) archive_record = models.ForeignKey(ArchiveRecord, on_delete=models.CASCADE) granted_at = models.DateTimeField() expires_at = models.DateTimeField() is_active = models.BooleanField(default=True)

逻辑说明:TemporaryPermission是轻量级授权,只存必要字段。在档案详情页的can_view_original()方法中,同时检查UserRoleAssignment(长期权限)和TemporaryPermission(临时权限),任一满足即放行。过期自动失效,无需定时任务清理。


4. OCR文本提取与全文检索:让PDF里的手写批注也能被“合同金额”搜出来

档案价值在于可发现。纯图像PDF无法搜索,必须提取文本。我们不追求100%识别率,而要平衡速度、准确率与部署成本——选用Tesseract OCR(开源、中文支持好、无云依赖),配合Elasticsearch实现毫秒级全文检索。

4.1 PDF转文本流水线:从上传到索引的自动化

# tasks.py (Celery异步任务) import fitz # PyMuPDF import pytesseract from PIL import Image from elasticsearch import Elasticsearch from django.conf import settings es_client = Elasticsearch([settings.ELASTICSEARCH_URL]) def extract_text_from_pdf(pdf_path: str) -> str: """从PDF提取文本:先尝试原生文本,失败则OCR每页""" doc = fitz.open(pdf_path) full_text = "" for page_num in range(len(doc)): page = doc[page_num] # 优先提取原生文本(快且准) text = page.get_text() if text.strip(): full_text += text + "\n\n" continue # 原生文本为空,OCR图片 pix = page.get_pixmap(dpi=200) # 200dpi平衡质量与速度 img = Image.frombytes("RGB", [pix.width, pix.height], pix.samples) ocr_text = pytesseract.image_to_string(img, lang='chi_sim+eng', config='--psm 6') full_text += ocr_text + "\n\n" doc.close() return full_text.strip() def index_archive_for_search(archive_record: ArchiveRecord): """将档案元数据+OCR文本索引到ES""" if not archive_record.digital_path: return try: text_content = extract_text_from_pdf(archive_record.digital_path) # 构建ES文档 es_doc = { "archive_code": archive_record.archive_code, "title": archive_record.title, "description": archive_record.description, "confidentiality_level": archive_record.confidentiality_level, "business_event_type": archive_record.business_event.event_type, "text_content": text_content[:1000000], # ES有字段长度限制,截断 "archived_at": archive_record.archived_at.isoformat(), } es_client.index( index=settings.ELASTICSEARCH_INDEX_NAME, id=archive_record.id, body=es_doc ) except Exception as e: # 记录错误,但不中断主流程 logger.error(f"ES indexing failed for {archive_record.archive_code}: {e}") # 在ArchiveRecord保存后触发 # post_save.connect(lambda sender, instance, **kwargs: index_archive_for_search(instance), sender=ArchiveRecord)

参数说明:--psm 6(Page Segmentation Mode 6)告诉Tesseract将整页视为单个文本块,对合同这类结构化文档效果最佳。dpi=200是实测平衡点——150dpi识别率下降明显,300dpi耗时翻倍。text_content截断至100万字符,避免ES单文档过大。

4.2 Elasticsearch检索接口:支持业务语义的复合查询

# views.py from elasticsearch import Elasticsearch from django.http import JsonResponse from django.views.decorators.csrf import csrf_exempt import json @csrf_exempt def search_archives(request): if request.method != 'POST': return JsonResponse({'error': 'Method not allowed'}, status=405) try: data = json.loads(request.body) query = data.get('q', '').strip() filters = data.get('filters', {}) # 构建ES查询DSL es_query = { "query": { "bool": { "must": [ {"multi_match": { "query": query, "fields": ["title^5", "description^3", "text_content^1"] # 标题权重最高 }} ], "filter": [] } }, "highlight": { "fields": { "text_content": {}, "title": {} } } } # 添加业务过滤器 if filters.get('confidentiality_level'): es_query['query']['bool']['filter'].append({ "term": {"confidentiality_level": filters['confidentiality_level']} }) if filters.get('business_event_type'): es_query['query']['bool']['filter'].append({ "term": {"business_event_type": filters['business_event_type']} }) # 执行搜索 res = es_client.search( index=settings.ELASTICSEARCH_INDEX_NAME, body=es_query, size=20 ) results = [] for hit in res['hits']['hits']: source = hit['_source'] results.append({ 'archive_code': source['archive_code'], 'title': source['title'], 'highlight': hit.get('highlight', {}), 'score': hit['_score'] }) return JsonResponse({'results': results}) except Exception as e: return JsonResponse({'error': str(e)}, status=500)

逻辑说明:multi_match在标题、描述、OCR文本三处搜索,但赋予不同权重(^5,^3,^1),确保搜“CT2024-001”时合同编号标题必然排第一。highlight返回匹配片段,前端可高亮显示,提升用户体验。


5. 借阅流程的状态机与超期预警:从“借了就忘”到“到期自动催还”

借阅不是按钮点击,而是一套有始有终的状态流转。我们用Django FSM(Finite State Machine)库定义严谨状态,配合Celery Beat定时任务实现超期预警,彻底解决“借条满天飞、归还不知谁”的顽疾。

5.1 借阅请求状态机:用装饰器约束非法状态跳转

# models.py from django_fsm import FSMField, transition class BorrowingRequest(models.Model): # ... 其他字段 ... # 状态字段 status = FSMField(default='PENDING') @transition(field=status, source=['PENDING'], target='APPROVED') def approve(self, approved_by: User): self.approved_by = approved_by self.approved_at = timezone.now() @transition(field=status, source=['PENDING'], target='REJECTED') def reject(self, rejected_by: User, reason: str): self.rejected_by = rejected_by self.rejection_reason = reason @transition(field=status, source=['APPROVED'], target='RETURNED') def return_document(self, returned_by: User): self.returned_by = returned_by self.returned_at = timezone.now() @transition(field=status, source=['APPROVED'], target='EXPIRED') def expire(self): """超期自动触发""" pass # 状态变更钩子:发送通知 def after_transition(self, *args, **kwargs): if kwargs['target'] == 'APPROVED': self.send_approval_notification() elif kwargs['target'] == 'RETURNED': self.send_return_notification() elif kwargs['target'] == 'EXPIRED': self.send_expiry_warning()

提示:@transition装饰器强制校验状态跳转合法性。试图从REJECTED直接approve()会抛出异常,杜绝代码误操作。after_transition是统一通知入口,避免在每个approve()调用后手动发邮件。

5.2 超期预警机制:Celery Beat定时扫描 + 邮件/站内信双通道

# tasks.py from celery import shared_task from django.core.mail import send_mail from django.contrib.auth.models import User from .models import BorrowingRequest @shared_task def check_overdue_borrowings(): """每小时执行:检查所有 APPROVED 状态且已超期的借阅""" now = timezone.now() overdue_requests = BorrowingRequest.objects.filter( status='APPROVED', due_date__lt=now ).select_related('borrower', 'archive_record', 'approved_by') for req in overdue_requests: # 标记为超期(触发状态机) req.expire() req.save() # 发送预警 subject = f"【档案预警】您借阅的档案 {req.archive_record.archive_code} 已超期" message = f""" 您于 {req.created_at.strftime('%Y-%m-%d')} 借阅的档案: - 档号:{req.archive_record.archive_code} - 标题:{req.archive_record.title} - 应归还时间:{req.due_date.strftime('%Y-%m-%d %H:%M')} - 已超期:{(now - req.due_date).days} 天 请立即归还或申请续借。逾期可能影响您的借阅权限。 """ # 发送邮件 send_mail(subject, message, 'archive@company.com', [req.borrower.email]) # 同时发站内信(假设你有Notification模型) # Notification.objects.create( # user=req.borrower, # title=subject, # content=message, # is_read=False # ) # Celery Beat配置(celery.py) # app.conf.beat_schedule = { # 'check-overdue-borrowings': { # 'task': 'archive.tasks.check_overdue_borrowings', # 'schedule': 3600.0, # 每小时一次 # }, # }

参数说明:schedule=3600.0表示每3600秒(1小时)执行一次。实际生产中,可根据业务量调整为1800.0(半小时)以提升响应速度。邮件模板中明确写出超期天数,比单纯说“已超期”更有压迫感。

5.3 常见问题与避坑:借阅流程中5个血泪经验总结

现象1:用户提交借阅后,状态卡在PENDING,审批人收不到通知

原因:审批人邮箱未在系统中维护,或send_mail()配置未指向真实SMTP服务器(开发环境常配console后端)
解决:上线前强制要求所有用户完善邮箱;在settings.py中区分环境配置邮件后端,生产环境必须用django.core.mail.backends.smtp.EmailBackend并验证连接。

现象2:OCR识别后搜索“人民币”,却搜不到PDF里写的“¥100,000.00”

原因:Tesseract默认不识别货币符号和千分位逗号,lang='chi_sim+eng'不足以覆盖
解决:在pytesseract.image_to_string()中添加config='--psm 6 -c tessedit_char_whitelist=0123456789.,¥$€%',显式声明白名单字符。

现象3:多人同时借阅同一份高密级档案,权限校验返回空列表

原因:build_scope_q_object()生成的Q对象在OR合并时未加括号,导致SQL逻辑错误(如A AND B OR C AND D应为(A AND B) OR (C AND D))
解决:在build_scope_q_object()中,对每个scope_rules生成独立Q对象,再用Q()包装后|=合并,Django会自动加括号。

现象4:ES索引后搜索无结果,但text_content字段确认有值

原因:Elasticsearch默认对中文分词使用standard分析器,对长数字、合同号等切分错误(如“CT2024-001”被切成“CT”“2024”“001”)
解决:在ES索引映射中,为archive_code和text_content字段指定keyword类型(不分词)和ik_max_word中文分词器,重建索引。

现象5:档号生成在高并发下出现重复(如两请求同时拿到00123)

原因:generate_archive_code()中虽用transaction.atomic(),但order_by('-archive_code')在archive_code为字符串时,排序逻辑与数值预期不符('00123' > '002'为False)
解决:改用数值序号字段sequence_number = models.PositiveIntegerField()单独存储,档号生成时查max(sequence_number),再拼接字符串,彻底规避字符串排序陷阱。


6. 生产就绪的关键技巧:从本地调试到灰度发布的3个硬核实践

真正落地的系统,90%的功夫不在功能实现,而在如何让它稳如磐石地跑在生产环境。以下是我在多个模拟项目X中沉淀下来的三条非文档化技巧,没有花哨概念,全是踩坑后抠出来的细节。

6.1 档案路径安全:用FileSystemStorage隔离用户上传与系统文件

用户上传的PDF不能直接存进Django静态目录,否则可能通过URL直接访问未授权档案。必须用FileSystemStorage指定独立路径,并禁用Web服务器对该路径的直接访问:

# settings.py import os from django.core.files.storage import FileSystemStorage # 档案专用存储,路径不在STATIC_ROOT下 ARCHIVE_STORAGE = FileSystemStorage( location=os.path.join(BASE_DIR, 'archive_storage'), base_url='/archive-files/' # 此URL需由Django视图控制权限,而非Nginx直接映射 ) # models.py 中使用 class ArchiveRecord(models.Model): digital_path = models.FileField( storage=ARCHIVE_STORAGE, upload_to='uploads/%Y/%m/%d/', # 按日期分目录,防单目录文件过多 blank=True )

关键操作:在Nginx配置中,必须屏蔽对/archive-files/的直接访问:

location ^~ /archive-files/ { deny all; # 禁止所有直接访问 }

所有档案文件下载,必须经过Django视图做权限校验:

# views.py from django.http import HttpResponse from django.views import View class ArchiveDownloadView(View): def get(self, request, archive_id): archive = get_object_or_404(ArchiveRecord, id=archive_id) if not can_user_access_archive(request.user, archive): # 权限校验函数 raise PermissionDenied response = FileResponse(archive.digital_path.open('rb')) response['Content-Disposition'] = f'attachment; filename="{archive.archive_code}.pdf"' return response

6.2 数据迁移安全:用RunPython替代sqlmigrate处理档号生成逻辑变更

当业务要求档号格式从YYYYMMDD-DEPT-CODE-SEQ升级为YYYYMMDD-DEPT-CODE-SEQ-VER(增加版本号)时,不能简单改模型再makemigrations——历史档号会丢失。必须用RunPython在迁移中安全演进:

# migrations/0005_add_version_to_archive_code.py from django.db import migrations def add_version_to_archive_code(apps, schema_editor): ArchiveRecord = apps.get_model('archive', 'ArchiveRecord') for record in ArchiveRecord.objects.all(): # 历史档号补全版本号 '-01' if not record.archive_code.endswith('-01'): parts = record.archive_code.rsplit('-', 1) # 拆分最后的序号 if len(parts) == 2: new_code = f"{parts[0]}-{parts[1]}-01" record.archive_code = new_code record.save() class Migration(migrations.Migration): dependencies = [('archive', '0004_alter_archiverecord_archive_code')] operations = [ migrations.RunPython(add_version_to_archive_code, reverse_code=migrations.RunPython.noop), ]

血泪经验:reverse_code=migrations.RunPython.noop表示此迁移不可逆。因为版本号一旦加上,无法可靠还原。生产发布前,必须在测试库中完整执行此迁移,并用ArchiveRecord.objects.filter(archive_code__regex=r'-\d{2}$').count()验证所有档号均已更新。

6.3 灰度发布策略:用Feature Flag控制新OCR引擎上线

当引入更准但更慢的新OCR引擎(如PP-OCRv3)时,不能一刀切切换。用Feature Flag实现灰度:

# utils/ocr_selector.py from django.conf import settings from django.core.cache import cache def get_ocr_engine(): # 从缓存读取开关,降低DB压力 flag_key = 'ocr_engine_flag' engine = cache.get(flag_key) if engine is None: # 从DB或环境变量读取,默认用旧引擎 engine = getattr(settings, 'DEFAULT_OCR_ENGINE', 'tesseract') cache.set(flag_key, engine, 300) # 缓存5分钟 return engine # 在OCR任务中 def extract_text_from_pdf(pdf_path: str) -> str: engine = get_ocr_engine() if engine == 'ppocr': return ppocr_extract(pdf_path) # 新引擎函数 else: return tesseract_extract(pdf_path) # 旧引擎函数

操作清单:

  1. 在管理后台添加开关页面,可设置ocr_engine_flag为tesseract或ppocr;
  2. 初始设为tesseract,观察一周无异常;
  3. 切换为ppocr,但只对user_id % 100 < 5(5%用户)生效,监控识别耗时与准确率;
  4. 无问题后,逐步扩大比例至100%。
    这比“停服升级”靠谱十倍——用户无感知,问题可秒级回滚。

我带过的每个模拟项目X,上线前必做这三件事:Nginx路径锁死、档号迁移全量验证、OCR灰度开关。它们不炫技,但能让你半夜接到告警电话的概率降低80%。档案系统不是锦上添花的玩具,它是业务连续性的基石——基石松动,上面盖的楼再漂亮也白搭。希望帮到你。

本文还有配套的精品资源,点击获取

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询