pcb-afm-simulation-system/web/backend/app/services/report_generator.py

"""Report generation service (P4-M4).

Generates Word simulation reports from task results.
Includes cover, parameters, results summary, metrics, AI analysis.
"""
import json
import os
from datetime import datetime
from typing import Any, Dict, List, Optional

try:
    from docx import Document
    from docx.shared import Inches, Pt, RGBColor
    from docx.enum.text import WD_ALIGN_PARAGRAPH
    HAS_DOCX = True
except ImportError:
    HAS_DOCX = False


class ReportGenerator:
    """Generate simulation reports from task results."""

    def __init__(self, output_dir: Optional[str] = None):
        self.output_dir = output_dir or os.path.join(
            os.path.dirname(os.path.dirname(os.path.dirname(__file__))),
            "output", "reports"
        )
        os.makedirs(self.output_dir, exist_ok=True)

    def generate_report(self, task_data: Dict[str, Any],
                         ai_analysis: Optional[Dict[str, Any]] = None,
                         report_title: Optional[str] = None) -> str:
        """Generate a Word report from task data.

        Args:
            task_data: Task data including parameters, results, metrics
            ai_analysis: Optional AI analysis results
            report_title: Custom report title

        Returns:
            Path to generated report file
        """
        if not HAS_DOCX:
            return self._generate_json_report(task_data, ai_analysis, report_title)

        doc = Document()

        # Title
        title = report_title or f"Simulation Report - {task_data.get('task_name', 'Task')}"
        heading = doc.add_heading(title, level=0)
        heading.alignment = WD_ALIGN_PARAGRAPH.CENTER

        # Metadata
        doc.add_paragraph(f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
        doc.add_paragraph(f"Task ID: {task_data.get('task_id', 'N/A')}")
        doc.add_paragraph(f"Status: {task_data.get('status', 'N/A')}")

        # Parameters section
        doc.add_heading("Simulation Parameters", level=1)
        params = task_data.get("plan_data", {})
        if params:
            table = doc.add_table(rows=1, cols=2)
            table.style = "Table Grid"
            hdr = table.rows[0].cells
            hdr[0].text = "Parameter"
            hdr[1].text = "Value"
            for key, value in params.items():
                row = table.add_row().cells
                row[0].text = str(key)
                row[1].text = str(value)

        # Results summary
        doc.add_heading("Results Summary", level=1)
        results = task_data.get("result_metrics", {})
        if results:
            table = doc.add_table(rows=1, cols=2)
            table.style = "Table Grid"
            hdr = table.rows[0].cells
            hdr[0].text = "Metric"
            hdr[1].text = "Value"
            for key, value in results.items():
                row = table.add_row().cells
                row[0].text = str(key)
                row[1].text = str(value)

        # Per-point results
        doc.add_heading("Per-Point Results", level=1)
        points = task_data.get("points", [])
        if points:
            table = doc.add_table(rows=1, cols=4)
            table.style = "Table Grid"
            hdr = table.rows[0].cells
            hdr[0].text = "Point"
            hdr[1].text = "Parameters"
            hdr[2].text = "Status"
            hdr[3].text = "Duration (s)"
            for i, point in enumerate(points):
                row = table.add_row().cells
                row[0].text = str(i + 1)
                row[1].text = json.dumps(point.get("params", {}), ensure_ascii=False)
                row[2].text = point.get("status", "N/A")
                row[3].text = str(point.get("duration_s", "N/A"))

        # AI Analysis section
        if ai_analysis:
            doc.add_heading("AI Analysis", level=1)
            if "summary" in ai_analysis:
                doc.add_paragraph(ai_analysis["summary"])
            if "convergence" in ai_analysis:
                doc.add_heading("Convergence Assessment", level=2)
                conv = ai_analysis["convergence"]
                doc.add_paragraph(f"Converged: {conv.get('converged', 'N/A')}")
                doc.add_paragraph(f"Confidence: {conv.get('confidence', 'N/A')}")
            if "recommendations" in ai_analysis:
                doc.add_heading("Recommendations", level=2)
                for rec in ai_analysis["recommendations"]:
                    doc.add_paragraph(rec, style="List Bullet")

        # Save
        filename = f"report_{task_data.get('task_id', 'task')}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.docx"
        filepath = os.path.join(self.output_dir, filename)
        doc.save(filepath)
        return filepath

    def _generate_json_report(self, task_data: Dict[str, Any],
                                ai_analysis: Optional[Dict[str, Any]],
                                report_title: Optional[str]) -> str:
        """Fallback: generate JSON report when python-docx is unavailable."""
        report = {
            "title": report_title or f"Simulation Report - {task_data.get('task_name', 'Task')}",
            "generated_at": datetime.now().isoformat(),
            "task_data": task_data,
            "ai_analysis": ai_analysis,
        }
        filename = f"report_{task_data.get('task_id', 'task')}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
        filepath = os.path.join(self.output_dir, filename)
        with open(filepath, "w", encoding="utf-8") as f:
            json.dump(report, f, ensure_ascii=False, indent=2)
        return filepath

    def list_reports(self) -> List[Dict[str, Any]]:
        """List all generated reports."""
        reports = []
        if os.path.exists(self.output_dir):
            for f in os.listdir(self.output_dir):
                if f.endswith((".docx", ".json")):
                    filepath = os.path.join(self.output_dir, f)
                    reports.append({
                        "filename": f,
                        "path": filepath,
                        "size": os.path.getsize(filepath),
                        "created_at": datetime.fromtimestamp(os.path.getctime(filepath)).isoformat(),
                    })
        return sorted(reports, key=lambda x: x["created_at"], reverse=True)


_generator: Optional[ReportGenerator] = None

def get_report_generator() -> ReportGenerator:
    global _generator
    if _generator is None:
        _generator = ReportGenerator()
    return _generator