66 lines
2.1 KiB
Python
66 lines
2.1 KiB
Python
|
"""
|
||
|
|
检查知识库的元数据字段定义
|
||
|
|
"""
|
||
|
|
|
||
|
|
import asyncio
|
||
|
|
import sys
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
sys.path.insert(0, str(Path(__file__).parent.parent))
|
||
|
|
|
||
|
|
from sqlalchemy import select
|
||
|
|
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
|
||
|
|
from sqlalchemy.orm import sessionmaker
|
||
|
|
from app.core.config import get_settings
|
||
|
|
from app.models.entities import (
|
||
|
|
MetadataFieldDefinition,
|
||
|
|
MetadataFieldStatus,
|
||
|
|
FieldRole,
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
async def check_metadata_fields():
|
||
|
|
"""检查元数据字段定义"""
|
||
|
|
settings = get_settings()
|
||
|
|
|
||
|
|
engine = create_async_engine(settings.database_url)
|
||
|
|
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
|
||
|
|
|
||
|
|
async with async_session() as session:
|
||
|
|
tenant_id = "szmp@ash@2026"
|
||
|
|
|
||
|
|
print(f"\n{'='*80}")
|
||
|
|
print(f"检查租户 {tenant_id} 的元数据字段定义")
|
||
|
|
print(f"{'='*80}")
|
||
|
|
|
||
|
|
stmt = select(MetadataFieldDefinition).where(
|
||
|
|
MetadataFieldDefinition.tenant_id == tenant_id,
|
||
|
|
MetadataFieldDefinition.status == MetadataFieldStatus.ACTIVE,
|
||
|
|
)
|
||
|
|
result = await session.execute(stmt)
|
||
|
|
fields = result.scalars().all()
|
||
|
|
|
||
|
|
print(f"\n找到 {len(fields)} 个活跃字段定义:")
|
||
|
|
|
||
|
|
for f in fields:
|
||
|
|
print(f"\n 字段: {f.field_key}")
|
||
|
|
print(f" label: {f.label}")
|
||
|
|
print(f" type: {f.type}")
|
||
|
|
print(f" required: {f.required}")
|
||
|
|
print(f" field_roles: {f.field_roles}")
|
||
|
|
print(f" options: {f.options}")
|
||
|
|
print(f" default_value: {f.default_value}")
|
||
|
|
|
||
|
|
filterable_fields = [
|
||
|
|
f for f in fields
|
||
|
|
if f.field_roles and FieldRole.RESOURCE_FILTER.value in f.field_roles
|
||
|
|
]
|
||
|
|
print(f"\n{'='*80}")
|
||
|
|
print(f"可过滤字段 (field_roles 包含 resource_filter): {len(filterable_fields)} 个")
|
||
|
|
for f in filterable_fields:
|
||
|
|
print(f" - {f.field_key} (label: {f.label}, required: {f.required})")
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
asyncio.run(check_metadata_fields())
|