|
1 | 1 | from __future__ import annotations |
2 | 2 |
|
| 3 | +import base64 |
3 | 4 | import json |
4 | 5 | import traceback |
5 | 6 | from datetime import UTC, datetime |
@@ -1235,6 +1236,67 @@ def drug_portfolio(self, *, query: dict[str, list[str]]) -> dict[str, Any]: |
1235 | 1236 | def promising_cures(self, *, query: dict[str, list[str]]) -> dict[str, Any]: |
1236 | 1237 | return self.drug_portfolio(query=query) |
1237 | 1238 |
|
| 1239 | + def structure_file(self, *, query: dict[str, list[str]]) -> dict[str, Any]: |
| 1240 | + raw_path = _query_optional_string(query, name="path") |
| 1241 | + if not raw_path: |
| 1242 | + raise BadRequestError("path query parameter is required") |
| 1243 | + |
| 1244 | + resolved = self._resolve_allowed_structure_path(raw_path) |
| 1245 | + if not resolved.exists() or not resolved.is_file(): |
| 1246 | + raise NotFoundError(f"Structure file not found: {resolved}") |
| 1247 | + |
| 1248 | + suffix = resolved.suffix.lower() |
| 1249 | + if suffix not in {".pdb", ".cif", ".mmcif", ".bcif"}: |
| 1250 | + raise BadRequestError( |
| 1251 | + "Unsupported structure format. Allowed: .pdb, .cif, .mmcif, .bcif" |
| 1252 | + ) |
| 1253 | + |
| 1254 | + max_bytes = 8 * 1024 * 1024 |
| 1255 | + file_size = resolved.stat().st_size |
| 1256 | + if file_size > max_bytes: |
| 1257 | + raise BadRequestError( |
| 1258 | + f"Structure file is too large ({file_size} bytes). Limit is {max_bytes} bytes." |
| 1259 | + ) |
| 1260 | + |
| 1261 | + if suffix == ".bcif": |
| 1262 | + content = base64.b64encode(resolved.read_bytes()).decode("ascii") |
| 1263 | + return { |
| 1264 | + "path": str(resolved), |
| 1265 | + "format": "bcif", |
| 1266 | + "encoding": "base64", |
| 1267 | + "content": content, |
| 1268 | + } |
| 1269 | + |
| 1270 | + text = resolved.read_text(encoding="utf-8") |
| 1271 | + format_name = "pdb" if suffix == ".pdb" else "mmcif" |
| 1272 | + return { |
| 1273 | + "path": str(resolved), |
| 1274 | + "format": format_name, |
| 1275 | + "encoding": "utf-8", |
| 1276 | + "content": text, |
| 1277 | + } |
| 1278 | + |
| 1279 | + def _resolve_allowed_structure_path(self, raw_path: str) -> Path: |
| 1280 | + text = raw_path.strip() |
| 1281 | + if not text: |
| 1282 | + raise BadRequestError("path must be a non-empty string") |
| 1283 | + |
| 1284 | + candidate = Path(text).expanduser() |
| 1285 | + if not candidate.is_absolute(): |
| 1286 | + candidate = (self.config.resolved_workspace_root / candidate).resolve() |
| 1287 | + else: |
| 1288 | + candidate = candidate.resolve() |
| 1289 | + |
| 1290 | + allowed_roots = ( |
| 1291 | + self.config.resolved_workspace_root.resolve(), |
| 1292 | + self.config.data_dir.resolve(), |
| 1293 | + ) |
| 1294 | + if not any(_is_path_within(candidate, root) for root in allowed_roots): |
| 1295 | + raise BadRequestError( |
| 1296 | + "path must be within the configured workspace root or studio data directory" |
| 1297 | + ) |
| 1298 | + return candidate |
| 1299 | + |
1238 | 1300 | def clawcures_handoff(self, payload: dict[str, Any]) -> dict[str, Any]: |
1239 | 1301 | objective = _optional_nonempty_string(payload.get("objective"), "objective") |
1240 | 1302 | system_prompt = _optional_nonempty_string( |
@@ -2102,6 +2164,14 @@ def _optional_string_list(value: Any, field_name: str) -> list[str] | None: |
2102 | 2164 | return normalized |
2103 | 2165 |
|
2104 | 2166 |
|
| 2167 | +def _is_path_within(path: Path, root: Path) -> bool: |
| 2168 | + try: |
| 2169 | + path.relative_to(root) |
| 2170 | + except ValueError: |
| 2171 | + return False |
| 2172 | + return True |
| 2173 | + |
| 2174 | + |
2105 | 2175 | def _to_metric_float(value: Any) -> float | None: |
2106 | 2176 | if value is None: |
2107 | 2177 | return None |
@@ -2386,6 +2456,12 @@ def do_GET(self) -> None: # noqa: N802 |
2386 | 2456 | self, HTTPStatus.OK, app.promising_cures(query=query) |
2387 | 2457 | ) |
2388 | 2458 | return |
| 2459 | + if path == "/api/structure-file": |
| 2460 | + query = parse_qs(parsed.query, keep_blank_values=False) |
| 2461 | + _json_response( |
| 2462 | + self, HTTPStatus.OK, app.structure_file(query=query) |
| 2463 | + ) |
| 2464 | + return |
2389 | 2465 | if path == "/api/clinical/trials": |
2390 | 2466 | _json_response(self, HTTPStatus.OK, app.clinical_trials()) |
2391 | 2467 | return |
|
0 commit comments