evyzacq
77 lines
2.7 KiB
Python
77 lines
2.7 KiB
Python
# IR Generator — delegates to the 3-stage pipeline
|
|
|
|
import logging
|
|
|
|
from server.services.ir_engine.pipeline import ir_pipeline
|
|
from server.services.prd_manager.service import prd_service
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class IRGenerator:
|
|
"""IR generator that runs the full 3-stage pipeline on parsed PRD documents."""
|
|
|
|
async def generate(self, prd_id: str, prd_text: str = "", skill_name: str = "default") -> dict:
|
|
"""Run the pipeline on a parsed PRD document."""
|
|
# Get the parsed document from the PRD service
|
|
prd = await prd_service.get_prd(prd_id)
|
|
if not prd:
|
|
return {"error": f"PRD not found: {prd_id}", "ir_id": ""}
|
|
|
|
# Fetch full parsed data from disk if available
|
|
parsed_doc = self._load_parsed(prd)
|
|
if not parsed_doc:
|
|
# Fallback: build minimal doc from PRD text
|
|
parsed_doc = {
|
|
"source": "",
|
|
"sections": [{"source": "正文", "blocks": [{"type": "para", "index": 1, "text": prd.get("full_text", "")}], "images": []}],
|
|
"image_sources": {},
|
|
"image_analysis": [],
|
|
"resolved_conflicts": [],
|
|
}
|
|
|
|
result = await ir_pipeline.run(prd_id, parsed_doc)
|
|
return {
|
|
"ir_id": result["ir_id"],
|
|
"prd_id": result["prd_id"],
|
|
"yaml_content": result["yaml_content"],
|
|
"ir_json": result["ir_json"],
|
|
"audit": result["audit"],
|
|
"audit_report": result["audit_report"],
|
|
"skill_used": skill_name,
|
|
"created_at": result["created_at"],
|
|
"pipeline_stats": result["pipeline_stats"],
|
|
}
|
|
|
|
async def get_ir(self, ir_id: str) -> dict | None:
|
|
return await ir_pipeline.get_ir(ir_id)
|
|
|
|
def _load_parsed(self, prd: dict) -> dict | None:
|
|
"""Load the full parsed JSON from disk if available."""
|
|
import json
|
|
parsed_path = prd.get("parsed_path", "")
|
|
if parsed_path:
|
|
try:
|
|
with open(parsed_path, "r", encoding="utf-8") as f:
|
|
return json.load(f)
|
|
except Exception:
|
|
pass
|
|
|
|
# Build from in-memory data
|
|
if prd.get("sections") or prd.get("full_text"):
|
|
sections = prd.get("sections", [])
|
|
if not sections and prd.get("full_text"):
|
|
sections = [{"source": "正文", "blocks": [{"type": "para", "index": 1, "text": prd["full_text"]}], "images": []}]
|
|
return {
|
|
"source": "",
|
|
"sections": sections,
|
|
"image_sources": {},
|
|
"image_analysis": prd.get("images", []),
|
|
"resolved_conflicts": [],
|
|
}
|
|
|
|
return None
|
|
|
|
|
|
ir_generator = IRGenerator()
|