| 113 | |
| 114 | |
| 115 | class SourceNodes(BaseModel): |
| 116 | id: str |
| 117 | metadata: Dict[str, Any] |
| 118 | score: Optional[float] |
| 119 | text: str |
| 120 | url: Optional[str] |
| 121 | |
| 122 | @classmethod |
| 123 | def from_source_node(cls, source_node: NodeWithScore): |
| 124 | metadata = source_node.node.metadata |
| 125 | url = metadata.get("URL") |
| 126 | |
| 127 | if not url: |
| 128 | file_name = metadata.get("file_name") |
| 129 | url_prefix = os.getenv("FILESERVER_URL_PREFIX") |
| 130 | if not url_prefix: |
| 131 | logger.warning( |
| 132 | "Warning: FILESERVER_URL_PREFIX not set in environment variables" |
| 133 | ) |
| 134 | if file_name and url_prefix: |
| 135 | url = f"{url_prefix}/data/{file_name}" |
| 136 | |
| 137 | return cls( |
| 138 | id=source_node.node.node_id, |
| 139 | metadata=metadata, |
| 140 | score=source_node.score, |
| 141 | text=source_node.node.text, # type: ignore |
| 142 | url=url, |
| 143 | ) |
| 144 | |
| 145 | @classmethod |
| 146 | def from_source_nodes(cls, source_nodes: List[NodeWithScore]): |
| 147 | return [cls.from_source_node(node) for node in source_nodes] |
| 148 | |
| 149 | |
| 150 | class Result(BaseModel): |
nothing calls this directly
no outgoing calls
no test coverage detected