""" 字段元数据自动生成工具 从 SQLAlchemy Model 或 Pydantic Schema 自动生成 FIELD_METADATA """ from typing import Dict, Any, Type, get_origin, get_args, Union from sqlalchemy import inspect from sqlalchemy.orm import DeclarativeMeta from pydantic import BaseModel import inspect as py_inspect def generate_field_metadata( model: Type[DeclarativeMeta], sensitive_fields: list = None, maskable_fields: list = None, hidden_fields: list = None, field_labels: Dict[str, str] = None ) -> Dict[str, Dict[str, Any]]: """ 从 SQLAlchemy Model 自动生成字段元数据 Args: model: SQLAlchemy Model 类 sensitive_fields: 敏感字段列表，如 ['mobile', 'email'] maskable_fields: 可脱敏字段列表，如 ['mobile', 'email', 'id_card'] hidden_fields: 默认隐藏字段列表，如 ['password'] field_labels: 字段中文名称映射，如 {'name': '姓名', 'mobile': '手机号'} Returns: 字段元数据字典 Example: >>> FIELD_METADATA = generate_field_metadata( ... User, ... sensitive_fields=['mobile', 'email', 'password'], ... maskable_fields=['mobile', 'email'], ... hidden_fields=['password'], ... field_labels={'name': '姓名', 'mobile': '手机号'} ... ) """ sensitive_fields = sensitive_fields or [] maskable_fields = maskable_fields or [] hidden_fields = hidden_fields or [] field_labels = field_labels or {} metadata = {} # 获取 Model 的所有列 mapper = inspect(model) for column in mapper.columns: field_name = column.name # 跳过内部字段 if field_name.startswith('_'): continue # 确定字段类型 field_type = _get_field_type(column.type) # 确定默认权限 default_permission = "hidden" if field_name in hidden_fields else "read" # 生成字段标签 label = field_labels.get(field_name) or _generate_label(field_name, column.comment) metadata[field_name] = { "label": label, "field_type": field_type, "sensitive": field_name in sensitive_fields, "maskable": field_name in maskable_fields, "default_permission": default_permission } return metadata def _get_field_type(column_type) -> str: """ 根据 SQLAlchemy 列类型确定字段类型 Args: column_type: SQLAlchemy 列类型 Returns: 字段类型字符串: string/integer/boolean/datetime/float """ type_name = column_type.__class__.__name__.lower() if 'int' in type_name or 'serial' in type_name: return "integer" elif 'bool' in type_name: return "boolean" elif 'date' in type_name or 'time' in type_name: return "datetime" elif 'float' in type_name or 'numeric' in type_name or 'decimal' in type_name: return "float" else: return "string" def _generate_label(field_name: str, comment: str = None) -> str: """ 生成字段标签 优先使用数据库注释，如果没有则根据字段名生成 Args: field_name: 字段名 comment: 数据库注释 Returns: 字段标签 """ if comment: return comment # 常见字段名映射 common_labels = { 'id': 'ID', 'name': '名称', 'code': '编码', 'title': '标题', 'description': '描述', 'remark': '备注', 'status': '状态', 'sort': '排序', 'create_time': '创建时间', 'update_time': '更新时间', 'created_at': '创建时间', 'updated_at': '更新时间', 'is_deleted': '是否删除', 'is_active': '是否激活', 'username': '用户名', 'password': '密码', 'email': '邮箱', 'mobile': '手机号', 'phone': '电话', 'address': '地址', 'avatar': '头像', 'gender': '性别', 'age': '年龄', 'dept_id': '部门ID', 'user_id': '用户ID', 'role_id': '角色ID', } # 如果在常见映射中，直接返回 if field_name in common_labels: return common_labels[field_name] # 处理带前缀的字段 if field_name.startswith('sys_'): base_name = field_name[4:] if base_name in common_labels: return f"系统{common_labels[base_name]}" # 处理下划线分隔的字段名 if '_' in field_name: parts = field_name.split('_') # 尝试翻译每个部分 translated_parts = [common_labels.get(part, part.title()) for part in parts] return ''.join(translated_parts) # 默认返回首字母大写的字段名 return field_name.replace('_', ' ').title() def generate_field_metadata_from_schema( schema: Type[BaseModel], model: Type[DeclarativeMeta] = None ) -> Dict[str, Dict[str, Any]]: """ 从 Pydantic Response Schema 生成字段元数据 优势： 1. 根据 Schema 的 Optional 类型判断字段是否必填 2. 必填字段标记为 required=True，前端禁止隐藏 3. 可选字段可以被隐藏 Args: schema: Pydantic Response Schema 类 model: SQLAlchemy Model 类（可选，用于获取数据库注释） Returns: 字段元数据字典 """ metadata = {} # 获取 Schema 的所有字段 schema_fields = schema.model_fields # 如果提供了 model，获取数据库注释 db_comments = {} if model: mapper = inspect(model) for column in mapper.columns: if column.comment: db_comments[column.name] = column.comment # 自动识别敏感/可脱敏字段的关键词 sensitive_keywords = ['password', 'passwd', 'pwd', 'secret', 'token', 'key'] maskable_keywords = ['mobile', 'phone', 'tel', 'email', 'mail', 'id_card', 'idcard', 'name'] hidden_keywords = ['password', 'passwd', 'pwd', 'secret', 'token', 'key'] for field_name, field_info in schema_fields.items(): # 判断字段是否必填（非 Optional） is_required = field_info.is_required() # 获取字段类型 field_type = _get_pydantic_field_type(field_info.annotation) # 生成标签 label = db_comments.get(field_name) or _generate_label(field_name) # 判断是否敏感/可脱敏 field_name_lower = field_name.lower() is_sensitive = any(keyword in field_name_lower for keyword in sensitive_keywords) is_maskable = any(keyword in field_name_lower for keyword in maskable_keywords) is_hidden = any(keyword in field_name_lower for keyword in hidden_keywords) # 默认权限 default_permission = "hidden" if is_hidden else "read" metadata[field_name] = { "label": label, "field_type": field_type, "required": is_required, # 必填字段，前端禁止隐藏 "sensitive": is_sensitive, "maskable": is_maskable, "default_permission": default_permission } return metadata def _get_pydantic_field_type(annotation) -> str: """ 从 Pydantic 字段类型获取字段类型字符串 Args: annotation: Pydantic 字段类型注解 Returns: 字段类型字符串 """ # 处理 Optional 类型 origin = get_origin(annotation) if origin is Union: args = get_args(annotation) # Optional[X] 实际是 Union[X, None]，取第一个非 None 类型 annotation = next((arg for arg in args if arg is not type(None)), str) # 获取类型名称 if hasattr(annotation, '__name__'): type_name = annotation.__name__.lower() else: type_name = str(annotation).lower() if 'int' in type_name: return "integer" elif 'bool' in type_name: return "boolean" elif 'datetime' in type_name or 'date' in type_name: return "datetime" elif 'float' in type_name or 'decimal' in type_name: return "float" else: return "string" def auto_generate_field_metadata(model: Type[DeclarativeMeta]) -> Dict[str, Dict[str, Any]]: """ 完全自动生成字段元数据（使用默认规则） 自动识别敏感字段： - password, passwd, pwd: 密码相关 - mobile, phone, tel: 手机号相关 - email, mail: 邮箱相关 - id_card, idcard, identity: 身份证相关 - bank_card, bankcard: 银行卡相关 - secret, token, key: 密钥相关 Args: model: SQLAlchemy Model 类 Returns: 字段元数据字典 """ # 自动识别敏感字段 sensitive_keywords = [ 'password', 'passwd', 'pwd', 'mobile', 'phone', 'tel', 'email', 'mail', 'id_card', 'idcard', 'identity', 'bank_card', 'bankcard', 'secret', 'token', 'key' ] # 自动识别可脱敏字段 maskable_keywords = [ 'mobile', 'phone', 'tel', 'email', 'mail', 'id_card', 'idcard', 'identity', 'name', # 姓名可脱敏 ] # 自动识别隐藏字段 hidden_keywords = [ 'password', 'passwd', 'pwd', 'secret', 'token', 'key' ] mapper = inspect(model) sensitive_fields = [] maskable_fields = [] hidden_fields = [] for column in mapper.columns: field_name = column.name.lower() # 检查是否为敏感字段 if any(keyword in field_name for keyword in sensitive_keywords): sensitive_fields.append(column.name) # 检查是否可脱敏 if any(keyword in field_name for keyword in maskable_keywords): maskable_fields.append(column.name) # 检查是否默认隐藏 if any(keyword in field_name for keyword in hidden_keywords): hidden_fields.append(column.name) return generate_field_metadata( model, sensitive_fields=sensitive_fields, maskable_fields=maskable_fields, hidden_fields=hidden_fields )