| """
|
| Stability Data Extractor - LLM-Powered Format-Agnostic Extraction
|
|
|
| This module uses LLM to understand and extract stability data from ANY file format.
|
| No hardcoded patterns - the LLM interprets the data structure dynamically.
|
| """
|
|
|
| import json
|
| import re
|
| from typing import Dict, List, Any, Optional
|
| from pathlib import Path
|
|
|
|
|
| class StabilityDataExtractor:
|
| """
|
| LLM-powered stability data extractor.
|
|
|
| Handles ANY format by using LLM to understand the data structure.
|
| Falls back to heuristic extraction if LLM is unavailable.
|
| """
|
|
|
|
|
| EXTRACTION_PROMPT = """你是药物稳定性数据提取专家。请从以下文本中提取稳定性数据。
|
|
|
| 【数据内容】
|
| {text_content}
|
|
|
| 【用户分析目标】
|
| {goal}
|
|
|
| 【任务】
|
| 请识别并提取:
|
| 1. 批次信息（批次名称/ID）；若数据按规格/包装/剂型分组（如 50mg vs 100mg、不同包装），
|
| 请在每个批次上给出 group_id（分组标识，如 "50mg"）与 factor_levels（如 {{"strength": "50mg"}}）。
|
| 2. 存储条件（如25°C/60%RH长期, 40°C/75%RH加速等）
|
| 3. 时间点（月）
|
| 4. 质量指标数值（如杂质含量、含量等）。对每个 CQA 给出 spec_type：
|
| - "upper"：上限型（越高越超标，如杂质 / 降解产物 / 水分）；
|
| - "lower"：下限型（越低越超标，如含量 / 效价 / Assay / Potency）。
|
| 并给出 role："degradant"（杂质/降解产物）或 "assay"（含量/效价），用于质量平衡核验。
|
| 5. 规格限度（仅当文档中**明确给出**可接受标准/限度时才提取；若文档未明确给出，必须返回 null，不得臆造默认值）
|
| 6. 减少设计声明：若文档/目标声明采用 ICH Q1D 括号法（Bracketing）或矩阵法（Matrixing），
|
| 请在顶层给出 design_type（"bracketing" | "matrixing" | "full"）与 reduced_factors（被减少的因素列表）。
|
| 7. 分组维度：若存在规格/包装/剂型分组，请在顶层给出 grouping_dimension
|
| （"strength" | "package" | "presentation"）。
|
|
|
| 【输出格式】
|
| 请严格按以下JSON格式输出:
|
| ```json
|
| {{
|
| "batches": [
|
| {{
|
| "batch_id": "批次ID",
|
| "batch_name": "批次名称",
|
| "group_id": null,
|
| "factor_levels": {{}},
|
| "conditions": [
|
| {{
|
| "condition_id": "条件描述（如 25C_60RH）",
|
| "condition_type": "longterm|accelerated|stress",
|
| "timepoints": [0, 3, 6, 9],
|
| "cqa_data": [
|
| {{
|
| "cqa_name": "指标名称（如总杂质）",
|
| "spec_type": "upper",
|
| "role": "degradant",
|
| "values": [0.1, 0.12, 0.15, 0.18]
|
| }}
|
| ]
|
| }}
|
| ]
|
| }}
|
| ],
|
| "specification_limit": null,
|
| "spec_provided": false,
|
| "primary_cqa": "主要质量指标名称",
|
| "design_type": "full",
|
| "reduced_factors": [],
|
| "grouping_dimension": null
|
| }}
|
| ```
|
|
|
| 【规格限度提取规则（务必遵守）】
|
| - 若文档中明确写明可接受标准 / 规格上限或下限（如"总杂质 ≤ 0.5%""含量 95.0%~105.0%"），则把数值填入 specification_limit 并将 spec_provided 设为 true。
|
| - 若文档**未**明确给出规格 / 限度，则 specification_limit 必须为 null 且 spec_provided 为 false。严禁凭空填写 0.5 等默认值。
|
|
|
| 如果无法识别数据结构，请返回空的batches数组并在"extraction_notes"字段说明原因。
|
| """
|
|
|
| def __init__(self, model_invoker=None):
|
| """
|
| Initialize extractor.
|
|
|
| Args:
|
| model_invoker: LLM invoker instance (lazy-loaded if not provided)
|
| """
|
| self._model_invoker = model_invoker
|
| self.extracted_data = {}
|
| self.metadata = {}
|
|
|
| @property
|
| def model_invoker(self):
|
| """Lazy-load model invoker."""
|
| if self._model_invoker is None:
|
| try:
|
| from layers.model_invoker import ModelInvoker
|
| self._model_invoker = ModelInvoker()
|
| except Exception as e:
|
| print(f"Warning: Could not load ModelInvoker: {e}")
|
| self._model_invoker = None
|
| return self._model_invoker
|
|
|
| def extract_from_text(self, text_content: str, goal: str = "") -> Dict[str, Any]:
|
| """
|
| Extract stability data from text using LLM.
|
|
|
| Args:
|
| text_content: Raw text from parsed files
|
| goal: Analysis goal from user
|
|
|
| Returns:
|
| Structured data dictionary with batches
|
| """
|
|
|
| result = {
|
| "batches": [],
|
| "specification_limit": 0.5,
|
| "spec_provided": False,
|
| "primary_cqa": "总杂质",
|
| "target_timepoints": self._extract_target_timepoints(goal),
|
| "design_type": "full",
|
| "reduced_factors": [],
|
| "grouping_dimension": None,
|
| "extraction_method": "none",
|
| "extraction_notes": ""
|
| }
|
|
|
| if not text_content or len(text_content.strip()) < 50:
|
| result["extraction_notes"] = "文本内容过短，无法提取数据"
|
| return result
|
|
|
|
|
| llm_result = self._extract_with_llm(text_content, goal)
|
| if llm_result and llm_result.get("batches"):
|
| result.update(llm_result)
|
| result["extraction_method"] = "llm"
|
| return result
|
|
|
|
|
| heuristic_result = self._extract_with_heuristics(text_content, goal)
|
| if heuristic_result and heuristic_result.get("batches"):
|
| result.update(heuristic_result)
|
| result["extraction_method"] = "heuristic"
|
| return result
|
|
|
| result["extraction_notes"] = "无法识别数据格式，请确保文件包含时间点和数值数据"
|
| return result
|
|
|
| def _extract_with_llm(self, text_content: str, goal: str) -> Optional[Dict]:
|
| """Use LLM to extract structured data."""
|
| if not self.model_invoker:
|
| return None
|
|
|
| try:
|
|
|
| max_chars = 8000
|
| truncated_text = text_content[:max_chars]
|
| if len(text_content) > max_chars:
|
| truncated_text += "\n... [文本已截断]"
|
|
|
| prompt = self.EXTRACTION_PROMPT.format(
|
| text_content=truncated_text,
|
| goal=goal or "分析稳定性数据"
|
| )
|
|
|
| response = self.model_invoker.invoke(
|
| system_prompt="你是专业的药物稳定性数据提取助手。请从文本中提取结构化的稳定性数据。",
|
| user_prompt=prompt,
|
| temperature=0.1
|
| )
|
|
|
| if response and hasattr(response, 'content'):
|
| content = response.content
|
| elif isinstance(response, str):
|
| content = response
|
| else:
|
| return None
|
|
|
|
|
| json_match = re.search(r'```json\s*([\s\S]*?)\s*```', content)
|
| if json_match:
|
| json_str = json_match.group(1)
|
| else:
|
|
|
| json_str = content.strip()
|
| if not json_str.startswith('{'):
|
| return None
|
|
|
| extracted = json.loads(json_str)
|
| return extracted
|
|
|
| except Exception as e:
|
| print(f"LLM extraction failed: {e}")
|
| return None
|
|
|
| def _extract_with_heuristics(self, text_content: str, goal: str) -> Optional[Dict]:
|
| """
|
| Fallback heuristic extraction using pattern recognition.
|
|
|
| Two strategies, tried in order:
|
| 1. **Column-oriented tables** (the common CSV / Excel paste layout where each
|
| row is one observation and columns are ``Batch, Condition, Time, <CQA…>``,
|
| optionally with a grouping column such as ``Strength`` / ``Package``).
|
| This layout is what real stability datasets use and is what the previous
|
| row-based heuristic completely failed to parse.
|
| 2. **Legacy row-based tables** (a header row listing several time points with
|
| measurement rows below) — kept as a fallback for unusual layouts.
|
| """
|
|
|
| column_result = self._extract_column_tables(text_content)
|
| if column_result and column_result.get("batches"):
|
| return column_result
|
|
|
|
|
| batches = []
|
| tables = self._find_time_series_tables(text_content)
|
| for i, table in enumerate(tables):
|
| batch = self._create_batch_from_table(table, i, text_content)
|
| if batch:
|
| batches.append(batch)
|
| if batches:
|
| return {"batches": batches}
|
|
|
| return None
|
|
|
|
|
|
|
|
|
|
|
| _NUM_RE = re.compile(r"-?\d+(?:\.\d+)?")
|
|
|
| def _extract_column_tables(self, text_content: str) -> Optional[Dict]:
|
| """Parse column-oriented stability tables into the structured schema.
|
|
|
| Returns a dict with ``batches`` (each carrying ``group_id`` /
|
| ``factor_levels`` when a grouping column is present), ``primary_cqa`` and an
|
| optional ``grouping_dimension``; or ``None`` when no such table is found.
|
| """
|
| if not text_content:
|
| return None
|
|
|
| tables = self._find_column_tables(text_content)
|
| if not tables:
|
| return None
|
|
|
| batches: List[Dict] = []
|
| grouping_dimension: Optional[str] = None
|
| for table in tables:
|
| built = self._build_batches_from_columns(table)
|
| if not built:
|
| continue
|
| tbl_batches, gdim = built
|
| batches.extend(tbl_batches)
|
| grouping_dimension = grouping_dimension or gdim
|
|
|
| if not batches:
|
| return None
|
|
|
| result: Dict[str, Any] = {
|
| "batches": batches,
|
| "primary_cqa": self._pick_primary_cqa(batches),
|
| }
|
| if grouping_dimension:
|
| result["grouping_dimension"] = grouping_dimension
|
| return result
|
|
|
| def _split_cells(self, line: str) -> List[str]:
|
| """Split a table line into cells, tolerating ``|`` / comma / tab / 2+ spaces."""
|
| if " | " in line:
|
| parts = line.split(" | ")
|
| elif "|" in line and "," not in line:
|
| parts = line.split("|")
|
| elif "," in line:
|
| parts = line.split(",")
|
| elif "\t" in line:
|
| parts = line.split("\t")
|
| else:
|
| parts = re.split(r"\s{2,}", line)
|
| return [p.strip() for p in parts if p.strip() != ""]
|
|
|
| def _is_time_col(self, header_cell: str) -> bool:
|
| """Heuristically decide whether a header cell denotes the time column."""
|
| h = (header_cell or "").lower()
|
| if any(k in h for k in ("time", "时间", "month", "月", "day", "天", "week", "周", "year", "年")):
|
| return True
|
| return h in ("t", "t(m)", "t(month)")
|
|
|
| def _is_group_col(self, header_cell: str) -> bool:
|
| """Whether a header cell denotes a grouping dimension (strength/package…)."""
|
| h = (header_cell or "").lower()
|
| return any(k in h for k in (
|
| "strength", "规格", "package", "包装", "container", "容器",
|
| "presentation", "剂型", "dosage", "dose", "fill", "装量",
|
| ))
|
|
|
| def _group_dimension(self, header_cell: str) -> str:
|
| """Map a grouping header to a canonical dimension id."""
|
| h = (header_cell or "").lower()
|
| if "strength" in h or "规格" in h or "dose" in h or "dosage" in h:
|
| return "strength"
|
| if "package" in h or "包装" in h or "container" in h or "容器" in h or "fill" in h or "装量" in h:
|
| return "package"
|
| if "presentation" in h or "剂型" in h:
|
| return "presentation"
|
| return "group"
|
|
|
| def _looks_like_header(self, cells: List[str]) -> bool:
|
| """A header has a time column and no purely-numeric cell."""
|
| if len(cells) < 2:
|
| return False
|
| if not any(self._is_time_col(c) for c in cells):
|
| return False
|
| for c in cells:
|
| if self._NUM_RE.fullmatch(c.strip()):
|
| return False
|
| return True
|
|
|
| def _row_has_number(self, cells: List[str]) -> bool:
|
| return any(self._NUM_RE.search(c) for c in cells)
|
|
|
| def _find_column_tables(self, text: str) -> List[Dict]:
|
| """Locate (header, data-rows) blocks in column-oriented text."""
|
| lines = text.split("\n")
|
| tables: List[Dict] = []
|
| i = 0
|
| n = len(lines)
|
| while i < n:
|
| raw = lines[i].strip()
|
| if not raw or raw.startswith("==="):
|
| i += 1
|
| continue
|
| cells = self._split_cells(raw)
|
| if self._looks_like_header(cells):
|
| rows: List[List[str]] = []
|
| j = i + 1
|
| while j < n:
|
| l2 = lines[j].strip()
|
| if not l2 or l2.startswith("==="):
|
| break
|
| c2 = self._split_cells(l2)
|
|
|
| if self._looks_like_header(c2):
|
| break
|
| if len(c2) < 2 or not self._row_has_number(c2):
|
| break
|
| rows.append(c2)
|
| j += 1
|
| if rows:
|
| tables.append({"header": cells, "rows": rows})
|
| i = j
|
| continue
|
| i += 1
|
| return tables
|
|
|
| def _classify_columns(self, header: List[str]) -> Dict[str, Any]:
|
| """Classify header cells into time / batch / condition / group / CQA columns."""
|
| cols: Dict[str, Any] = {
|
| "time": -1, "batch": -1, "condition": -1,
|
| "group": None, "group_dim": None, "cqa": [],
|
| }
|
| for idx, h in enumerate(header):
|
| hl = h.lower()
|
| if cols["time"] < 0 and self._is_time_col(h):
|
| cols["time"] = idx
|
| elif cols["batch"] < 0 and any(k in hl for k in ("batch", "lot", "批")):
|
| cols["batch"] = idx
|
| elif cols["condition"] < 0 and any(
|
| k in hl for k in ("condition", "storage", "条件", "储存", "存储")
|
| ):
|
| cols["condition"] = idx
|
| elif cols["group"] is None and self._is_group_col(h):
|
| cols["group"] = idx
|
| cols["group_dim"] = self._group_dimension(h)
|
| else:
|
| cols["cqa"].append((idx, h))
|
| return cols
|
|
|
| def _to_float(self, cell: Optional[str]) -> Optional[float]:
|
| if cell is None:
|
| return None
|
| m = self._NUM_RE.search(str(cell))
|
| if not m:
|
| return None
|
| try:
|
| return float(m.group(0))
|
| except ValueError:
|
| return None
|
|
|
| def _canon_cqa_name(self, header: str) -> str:
|
| """Map a CQA column header (with units) to a canonical Chinese CQA name."""
|
| base = re.sub(r"[\(（].*?[\)）]", "", header or "").strip()
|
| hl = (header or "").lower()
|
| if any(k in hl for k in ("assay", "content", "potency", "含量", "效价", "%lc")):
|
| return "含量"
|
| is_total = ("total" in hl) or ("总" in (header or ""))
|
| if any(k in hl for k in ("impur", "杂质", "related", "降解", "degrad")):
|
| return "总杂质" if is_total else "降解产物"
|
| if any(k in hl for k in ("moisture", "water", "水分")):
|
| return "水分"
|
| if any(k in hl for k in ("dissolution", "溶出")):
|
| return "溶出度"
|
| return base or "质量指标"
|
|
|
| def _canon_condition(self, cond: str) -> tuple:
|
| """Map a condition cell to ``(condition_id, condition_type)``."""
|
| c = str(cond or "").strip()
|
| cl = c.lower()
|
| if not c:
|
| return "storage", ""
|
| if "40" in c or "加速" in c or "accel" in cl:
|
| return c, "accelerated"
|
| if "25" in c or "长期" in c or "long" in cl:
|
| return c, "longterm"
|
| if "30" in c or "中间" in c or "intermediate" in cl:
|
| return c, "intermediate"
|
| if "60" in c or "高温" in c or "stress" in cl:
|
| return c, "stress"
|
| return c, ""
|
|
|
| def _build_batches_from_columns(self, table: Dict) -> Optional[tuple]:
|
| """Build batch dicts from a classified column table.
|
|
|
| Groups rows by (batch, condition, grouping-value); each unique combination
|
| becomes one batch with a single condition holding all detected CQAs.
|
| """
|
| header = table["header"]
|
| cols = self._classify_columns(header)
|
| if cols["time"] < 0 or not cols["cqa"]:
|
| return None
|
|
|
| grouping_dim = cols["group_dim"]
|
|
|
| groups: Dict[tuple, Dict[str, Any]] = {}
|
| order: List[tuple] = []
|
|
|
| for row in table["rows"]:
|
| def cell(idx):
|
| return row[idx] if (idx is not None and 0 <= idx < len(row)) else None
|
|
|
| tval = self._to_float(cell(cols["time"]))
|
| if tval is None:
|
| continue
|
| bid = cell(cols["batch"]) or "B1"
|
| cond = cell(cols["condition"]) or ""
|
| gval = cell(cols["group"]) if cols["group"] is not None else None
|
| key = (str(bid), str(cond), str(gval) if gval is not None else None)
|
| if key not in groups:
|
| groups[key] = {"times": [], "cqa": {}}
|
| order.append(key)
|
| groups[key]["times"].append(tval)
|
| for (cidx, cname) in cols["cqa"]:
|
| canon = self._canon_cqa_name(cname)
|
| groups[key]["cqa"].setdefault(canon, []).append(self._to_float(cell(cidx)))
|
|
|
| batches: List[Dict] = []
|
| for key in order:
|
| bid, cond, gval = key
|
| payload = groups[key]
|
| cqa_list: List[Dict] = []
|
| for cname, vals in payload["cqa"].items():
|
| if all(v is None for v in vals):
|
| continue
|
| cqa_list.append({"cqa_name": cname, "values": vals})
|
| if not cqa_list:
|
| continue
|
| cond_id, cond_type = self._canon_condition(cond)
|
|
|
|
|
| batch_id = gval if (gval is not None and grouping_dim) else bid
|
| batch = {
|
| "batch_id": str(batch_id),
|
| "batch_name": str(batch_id),
|
| "batch_type": "target",
|
| "conditions": [{
|
| "condition_id": cond_id,
|
| "condition_type": cond_type,
|
| "timepoints": payload["times"],
|
| "cqa_data": cqa_list,
|
| }],
|
| }
|
| if gval is not None and grouping_dim:
|
| batch["group_id"] = str(gval)
|
| batch["factor_levels"] = {grouping_dim: str(gval)}
|
| batches.append(batch)
|
|
|
| if not batches:
|
| return None
|
| return batches, grouping_dim
|
|
|
| def _pick_primary_cqa(self, batches: List[Dict]) -> str:
|
| """Pick a primary CQA: prefer a degradant/impurity, then assay, else first."""
|
| names: List[str] = []
|
| for b in batches:
|
| for c in b.get("conditions", []):
|
| for cqa in c.get("cqa_data", []):
|
| nm = cqa.get("cqa_name")
|
| if nm and nm not in names:
|
| names.append(nm)
|
| if not names:
|
| return "总杂质"
|
| for n in names:
|
| if n in ("总杂质", "降解产物") or "杂质" in n or "降解" in n:
|
| return n
|
| return names[0]
|
|
|
| def _find_time_series_tables(self, text: str) -> List[Dict]:
|
| """Find patterns that look like time-series data tables."""
|
| tables = []
|
| lines = text.split('\n')
|
|
|
|
|
| time_pattern = r'(\d+)\s*[MmHhDd月周天]'
|
|
|
| for i, line in enumerate(lines):
|
| time_matches = re.findall(time_pattern, line)
|
| if len(time_matches) >= 2:
|
|
|
| times = [int(t) for t in time_matches]
|
|
|
|
|
| data_rows = []
|
| for j in range(i+1, min(i+15, len(lines))):
|
| numbers = re.findall(r'(\d+\.?\d*)', lines[j])
|
| if len(numbers) >= len(times):
|
|
|
| try:
|
| values = [float(n) for n in numbers[:len(times)]]
|
| if all(0 <= v <= 200 for v in values):
|
| row_type = self._identify_row_type(lines[j])
|
| data_rows.append({
|
| "values": values,
|
| "type": row_type,
|
| "raw": lines[j]
|
| })
|
| except:
|
| pass
|
|
|
| if data_rows:
|
|
|
| context_start = max(0, i - 10)
|
| context = '\n'.join(lines[context_start:i+1])
|
|
|
| tables.append({
|
| "times": times,
|
| "rows": data_rows,
|
| "context": context,
|
| "line_number": i
|
| })
|
|
|
| return tables
|
|
|
| def _identify_row_type(self, line: str) -> str:
|
| """Identify what type of measurement a row represents."""
|
| line_lower = line.lower()
|
|
|
| if any(kw in line_lower for kw in ['杂质', 'impurity', '杂']):
|
| return 'impurity'
|
| elif any(kw in line_lower for kw in ['含量', 'assay', 'content']):
|
| return 'assay'
|
| elif any(kw in line_lower for kw in ['水分', 'moisture', 'water']):
|
| return 'moisture'
|
| elif any(kw in line_lower for kw in ['溶出', 'dissolution']):
|
| return 'dissolution'
|
|
|
| return 'unknown'
|
|
|
| def _create_batch_from_table(self, table: Dict, index: int, full_text: str) -> Optional[Dict]:
|
| """Create a batch structure from extracted table data."""
|
| context = table.get("context", "")
|
|
|
|
|
| batch_name = self._extract_batch_name(context, full_text, index)
|
|
|
|
|
| condition_info = self._extract_condition(context)
|
|
|
|
|
| cqa_list = []
|
| for row in table.get("rows", []):
|
| cqa_name = "总杂质" if row["type"] == "impurity" else (
|
| "含量" if row["type"] == "assay" else "质量指标"
|
| )
|
| cqa_list.append({
|
| "cqa_name": cqa_name,
|
| "values": row["values"]
|
| })
|
|
|
| if not cqa_list:
|
| return None
|
|
|
| return {
|
| "batch_id": batch_name.replace(" ", "_"),
|
| "batch_name": batch_name,
|
| "batch_type": "target",
|
| "conditions": [{
|
| "condition_id": condition_info["id"],
|
| "condition_type": condition_info["type"],
|
| "timepoints": table["times"],
|
| "cqa_data": cqa_list
|
| }]
|
| }
|
|
|
| def _extract_batch_name(self, context: str, full_text: str, index: int) -> str:
|
| """Extract batch name from context using various patterns."""
|
| patterns = [
|
| r'批[次号][:：\s]*([A-Za-z0-9\-_]+)',
|
| r'Batch[:：\s]*([A-Za-z0-9\-_]+)',
|
| r'([A-Z]{2,3}[-_]\d{4,}[-_]?[A-Z0-9]*)',
|
| r'([SF][-_]?\d{4}[-_]?\d+)',
|
| r'样品[:：\s]*(.{3,20})',
|
| ]
|
|
|
| for pattern in patterns:
|
| match = re.search(pattern, context, re.IGNORECASE)
|
| if match:
|
| name = match.group(1).strip()
|
| if len(name) >= 3:
|
| return name
|
|
|
|
|
| return f"批次{index + 1}"
|
|
|
| def _extract_condition(self, context: str) -> Dict[str, str]:
|
| """Extract storage condition from context."""
|
| context_lower = context.lower()
|
|
|
|
|
| if any(kw in context_lower for kw in ['40°c', '40℃', '40c', '加速']):
|
| return {"id": "40C_Accelerated", "type": "accelerated"}
|
| elif any(kw in context_lower for kw in ['25°c', '25℃', '25c', '长期']):
|
| return {"id": "25C_LongTerm", "type": "longterm"}
|
| elif any(kw in context_lower for kw in ['60°c', '60℃', '60c', '高温']):
|
| return {"id": "60C_Stress", "type": "stress"}
|
| elif any(kw in context_lower for kw in ['30°c', '30℃', '30c', '中间']):
|
| return {"id": "30C_Intermediate", "type": "intermediate"}
|
|
|
| return {"id": "Unknown_Condition", "type": "unknown"}
|
|
|
| def _extract_target_timepoints(self, goal: str) -> List[int]:
|
| """Extract target prediction timepoints from goal text."""
|
| timepoints = []
|
|
|
| patterns = [
|
| r'(\d+)\s*[个]?月',
|
| r'(\d+)\s*[Mm]',
|
| r'(\d+)\s*months?'
|
| ]
|
|
|
| for pattern in patterns:
|
| matches = re.findall(pattern, goal)
|
| timepoints.extend([int(m) for m in matches])
|
|
|
| timepoints = sorted(list(set(timepoints)))
|
|
|
| if not timepoints:
|
| timepoints = [24, 36]
|
|
|
| return timepoints
|
|
|
|
|
|
|
| def extract_stability_data(file_paths: List[str], goal: str) -> Dict[str, Any]:
|
| """
|
| Main entry point for data extraction.
|
| """
|
| from utils.file_parsers import parse_file
|
|
|
| extractor = StabilityDataExtractor()
|
| all_text = ""
|
|
|
| for path in file_paths:
|
| try:
|
| content = parse_file(path)
|
| if content:
|
| all_text += f"\n=== File: {path} ===\n{content}\n"
|
| except Exception as e:
|
| print(f"Error parsing {path}: {e}")
|
|
|
| return extractor.extract_from_text(all_text, goal)
|
|
|
