""" PRD Parser Utilities. This module contains parsing and normalization functions for PRD documents, extracted from prd.py to follow single responsibility principle. """ import json import re from typing import Any def extract_json_from_response(content: str) -> dict[str, Any] | None: """ Extract JSON from PRD response content. Tries multiple strategies: 1. Fenced code block: ```json ... ``` 2. Raw JSON object: outermost { ... } Returns parsed dict on success, None if no valid JSON found. """ if not content or not content.strip(): return None # Strategy 1: fenced ```json ... ``` blocks json_match = re.search(r"```(?:json)?\s*([\s\S]*?)\s*```", content) if json_match: raw = json_match.group(1).strip() try: parsed = json.loads(raw) if isinstance(parsed, dict): return parsed except json.JSONDecodeError: pass # Strategy 2: raw JSON object { ... } json_match = re.search(r"\{[\s\S]*\}", content) if json_match: raw = json_match.group(0).strip() try: parsed = json.loads(raw) if isinstance(parsed, dict): return parsed except json.JSONDecodeError: pass return None def normalize_llm_content(content: str | list[str | dict[str, Any]]) -> str: """ Normalize LLM content into a string. Handles various content formats that LLMs may return. """ if isinstance(content, str): return content parts: list[str] = [] for item in content: if isinstance(item, str): parts.append(item) continue if isinstance(item, dict): text = item.get("text") if isinstance(text, str): parts.append(text) continue try: parts.append(json.dumps(item)) except TypeError: parts.append(str(item)) return "\n".join(part for part in parts if part) def normalize_assumptions(value: str | list[str] | None) -> list[str]: """ Normalize assumptions stored as string or list into list[str]. """ if value is None: return [] if isinstance(value, list): return [str(item).strip() for item in value if str(item).strip()] if isinstance(value, str): if not value.strip(): return [] try: parsed = json.loads(value) if isinstance(parsed, list): return [str(item).strip() for item in parsed if str(item).strip()] except json.JSONDecodeError: pass return [line.strip() for line in value.splitlines() if line.strip()] return [] def parse_prd_sections(prd_content: str) -> dict[str, Any]: """ Parse PRD content and extract structured sections from markdown or JSON. Handles both JSON and markdown formats that the LLM may return. """ # First try JSON parsing parsed = extract_json_from_response(prd_content) if parsed: return { "product_vision": parsed.get("product_vision", ""), "key_features": parsed.get("features", {}), "user_stories": parsed.get("user_stories", []), "assumptions": parsed.get("assumptions", []), } # Parse markdown format - extract sections by headers sections = {} # Extract Product Vision vision_match = re.search( r"##?\s*1\.?\s*Product Vision\s*\n(.*?)(?=\n##|\Z)", prd_content, re.DOTALL | re.IGNORECASE, ) if not vision_match: vision_match = re.search( r"Product Vision\s*\n(.*?)(?=\n##|\Z)", prd_content, re.DOTALL | re.IGNORECASE, ) sections["product_vision"] = vision_match.group(1).strip() if vision_match else "" # Extract Target Users users_match = re.search( r"##?\s*Target Users?\s*\n(.*?)(?=\n##|\Z)", prd_content, re.DOTALL | re.IGNORECASE, ) sections["target_users"] = users_match.group(1).strip() if users_match else "" # Extract Key Features features_match = re.search( r"##?\s*3\.?\s*Key Features?\s*\n(.*?)(?=\n##|\Z)", prd_content, re.DOTALL | re.IGNORECASE, ) if not features_match: features_match = re.search( r"Key Features?\s*\n(.*?)(?=\n##|\Z)", prd_content, re.DOTALL | re.IGNORECASE, ) # Parse feature list from markdown features = {} if features_match: feature_text = features_match.group(1) # Find all feature titles (### F1: or ### Feature Name) feature_matches = re.findall(r"###\s*(?:F\d+:?\s*)?([^\n]+)", feature_text) for i, title in enumerate(feature_matches, 1): # Try to find priority priority = "should" priority_match = re.search( r"priority[:\s]*(\w+)", feature_text, re.IGNORECASE ) if priority_match: p = priority_match.group(1).lower() if "must" in p: priority = "must" elif "could" in p: priority = "could" features[f"F{i}"] = {"title": title.strip(), "priority": priority} sections["key_features"] = features # Extract User Stories - look for table format stories_match = re.search( r"##?\s*4\.?\s*User Stories?\s*\n(.*?)(?=\n##|\Z)", prd_content, re.DOTALL | re.IGNORECASE, ) user_stories = [] if stories_match: table_text = stories_match.group(1) # Parse markdown table rows rows = re.findall(r"\|\s*([^|]+)\s*\|", table_text) # Skip header row if present, extract story info for row in rows: if "---" in row or "ID" in row: continue user_stories.append({"content": row.strip()}) sections["user_stories"] = user_stories # Extract Assumptions assumptions_match = re.search( r"##?\s*5\.?\s*Assumptions.*?\n(.*?)(?=\n##|\Z)", prd_content, re.DOTALL | re.IGNORECASE, ) if not assumptions_match: assumptions_match = re.search( r"Assumptions.*?\n(.*?)(?=\n##|\Z)", prd_content, re.DOTALL | re.IGNORECASE ) assumptions = [] if assumptions_match: # Extract bullet points items = re.findall( r"[-*]\s*(.+?)(?=\n[-*]|\n\n|\Z)", assumptions_match.group(1) ) assumptions = [item.strip() for item in items] sections["assumptions"] = assumptions return { "product_vision": sections.get("product_vision", ""), "key_features": sections.get("key_features", {}), "user_stories": sections.get("user_stories", []), "assumptions": sections.get("assumptions", []), }