"""读取前端组件目录""" from pathlib import Path from app.schemas.component_resource import ( ComponentCatalog, ComponentQuery, ComponentResult, ) class ComponentProvider: def __init__(self, catalog_path: str | Path | None = None): self.catalog_path = catalog_path def query(self, query: ComponentQuery) -> ComponentResult: # TODO 以后迁移数据库,保持组件查询接口不变 path = ( Path(self.catalog_path) if self.catalog_path is not None else Path(__file__).resolve().parent / "catalog" / "components.json" ) if not path.is_absolute(): path = Path(__file__).resolve().parents[2] / path try: # 每次读取当前文件,避免更新目录后仍使用旧缓存 content = path.read_text(encoding="utf-8-sig") except FileNotFoundError: return ComponentResult( status="unsupported", message="尚未配置前端组件目录,请先导出真实组件定义", ) # 文件损坏或定义不合法直接报错,不能伪装成查无结果 catalog = ComponentCatalog.model_validate_json(content) keyword = ( query.keyword.strip().casefold() if query.keyword is not None else None ) matches = [] for component in catalog.components: if query.key is not None and component.key != query.key: continue if query.type is not None and component.type != query.type: continue searchable = " ".join( [ component.key, component.type, component.name, component.description, *component.tags, ] ).casefold() if keyword is not None and keyword not in searchable: continue matches.append(component) return ComponentResult( status="resolved" if matches else "not_found", components=matches, catalog_version=catalog.version, source=catalog.source, message="已找到组件定义" if matches else "已查询组件目录,未找到匹配的组件", ) if __name__ == "__main__": provider = ComponentProvider() result = provider.query(ComponentQuery()) print(result.model_dump_json(indent=2))