"""Validation report generator for V2 extraction pipeline. Reads batch_extraction_status.json and generates a comprehensive markdown report with success metrics, failed files analysis, and recommendations. """ import json from pathlib import Path from datetime import datetime from typing import Dict, List, Any class ValidationReport: """Generate comprehensive validation report from status JSON.""" def __init__(self, status_file: Path): """Initialize with status JSON file. Args: status_file: Path to batch_extraction_status.json """ self.status_file = status_file with open(status_file, encoding='utf-8') as f: self.status = json.load(f) def generate_markdown_report(self, output_file: Path) -> Path: """Generate human-readable markdown report. Args: output_file: Path where report should be saved Returns: Path to generated report """ report_lines = [ "# V2 Extraction Pipeline - Validation Report", "", f"**Generated:** {datetime.now().isoformat()}", f"**Total Files:** {self.status.get('total_files', len(self.status.get('files', {})))}", f"**Status File:** {self.status_file}", "", "## Summary", "", ] # Add summary section summary = self.status.get("summary", {}) if summary: report_lines.extend([ f"- **Success Rate:** {summary.get('success_rate', 0):.2f}%", f"- **Completed:** {summary.get('completed', 0)}", f"- **Failed:** {summary.get('failed', 0)}", f"- **Pending:** {summary.get('pending', 0)}", f"- **Processing:** {summary.get('processing', 0)}", "", ]) # Add metrics section metrics = self.status.get("metrics", {}) if metrics: report_lines.extend([ "## Pipeline Metrics", "", f"- **V2 Success:** {metrics.get('v2_success', 0)}", f"- **V1 Fallback:** {metrics.get('v1_fallback', 0)}", "", ]) # Success by format table by_format = metrics.get("by_format", {}) if by_format: report_lines.extend([ "## Success by Format", "", "| Format | Total | Success | Failed | Success Rate |", "|--------|-------|---------|--------|--------------|", ]) for fmt, stats in sorted(by_format.items()): success_rate = (stats['success'] / stats['total'] * 100) if stats['total'] > 0 else 0 report_lines.append( f"| {fmt} | {stats['total']} | {stats['success']} | " f"{stats['failed']} | {success_rate:.1f}% |" ) report_lines.append("") # Failed files section failed_files = self._get_failed_files() if failed_files: report_lines.extend([ "## Failed Files", "", f"**Total Failed:** {len(failed_files)}", "", "| File | Error Category | Error Message | Attempts |", "|------|----------------|---------------|----------|", ]) for file_data in failed_files: error_msg = (file_data.get('error') or 'Unknown error')[:80] report_lines.append( f"| {file_data['filename']} | {file_data.get('error_category', 'unknown')} | " f"{error_msg}... | {file_data.get('attempts', 1)} |" ) report_lines.append("") # Error category breakdown error_categories = self._count_error_categories(failed_files) if error_categories: report_lines.extend([ "### Error Categories Breakdown", "", "| Category | Count |", "|----------|-------|", ]) for category, count in sorted(error_categories.items(), key=lambda x: x[1], reverse=True): report_lines.append(f"| {category} | {count} |") report_lines.append("") # Known limitations report_lines.extend([ "## Known Limitations", "", ]) corrupted_count = sum( 1 for f in failed_files if f.get('error_category') == 'file_corrupted' ) if corrupted_count > 0: report_lines.append( f"- **Corrupted Files:** {corrupted_count} files have ZIP structure damage or are malformed" ) password_count = sum( 1 for f in failed_files if f.get('error_category') == 'password_protected' ) if password_count > 0: report_lines.append( f"- **Password Protected:** {password_count} files require password for access" ) large_file_count = sum( 1 for f in failed_files if f.get('error_category') == 'file_too_large' ) if large_file_count > 0: report_lines.append( f"- **Large Files:** {large_file_count} files exceed 100MB size limit" ) # Recommendations report_lines.extend([ "", "## Recommendations", "", ]) success_rate = summary.get('success_rate', 0) if success_rate < 95: report_lines.extend([ f"⚠️ **Success rate ({success_rate:.1f}%) is below 95% target.**", "", "### Suggested Actions:", "", ]) if corrupted_count > 0: report_lines.append( f"1. **Investigate {corrupted_count} corrupted files** - " "may need source file repair or re-download" ) if password_count > 0: report_lines.append( f"2. **Obtain passwords for {password_count} protected files** - " "or request unprotected versions" ) if large_file_count > 0: report_lines.append( f"3. **Review {large_file_count} large files** - " "consider increasing size limit or splitting files" ) # Check for LLM errors llm_errors = sum( 1 for f in failed_files if f.get('error_category') == 'llm_error' ) if llm_errors > 0: report_lines.append( f"4. **Fix {llm_errors} LLM errors** - " "check Azure OpenAI credentials and quota" ) else: report_lines.extend([ "✅ **Success rate meets 95% target.**", "", "The V2 extraction pipeline is performing within acceptable parameters.", ]) # Processing time statistics report_lines.extend([ "", "## Processing Statistics", "", ]) total_duration, avg_duration = self._calculate_duration_stats() if total_duration is not None: report_lines.extend([ f"- **Total Processing Time:** {total_duration:.1f} seconds ({total_duration/60:.1f} minutes)", f"- **Average Per File:** {avg_duration:.1f} seconds", "", ]) # Timestamps started_at = self.status.get('started_at') completed_at = self.status.get('completed_at') if started_at: report_lines.append(f"- **Started:** {started_at}") if completed_at: report_lines.append(f"- **Completed:** {completed_at}") report_lines.append("") report_lines.append("---") report_lines.append("") report_lines.append("*Generated by ValidationReport class from batch_extraction_status.json*") # Write report output_file.parent.mkdir(parents=True, exist_ok=True) with open(output_file, "w", encoding='utf-8') as f: f.write("\n".join(report_lines)) return output_file def _get_failed_files(self) -> List[Dict[str, Any]]: """Get list of failed files with details.""" files = self.status.get("files", {}) failed = [ {**file_data, "file_path": file_path} for file_path, file_data in files.items() if file_data.get("status") == "failed" ] return sorted(failed, key=lambda x: x.get('filename', '')) def _count_error_categories(self, failed_files: List[Dict]) -> Dict[str, int]: """Count failed files by error category.""" categories = {} for file_data in failed_files: category = file_data.get('error_category', 'unknown') categories[category] = categories.get(category, 0) + 1 return categories def _calculate_duration_stats(self) -> tuple[float, float]: """Calculate total and average processing duration.""" files = self.status.get("files", {}) durations = [ file_data.get('duration_seconds', 0) for file_data in files.values() if file_data.get('duration_seconds') is not None ] if not durations: return None, None total = sum(durations) avg = total / len(durations) if durations else 0 return total, avg def main(): """Command-line interface for report generation.""" import argparse parser = argparse.ArgumentParser( description="Generate validation report from batch extraction status" ) parser.add_argument( "--status-file", default="batch_extraction_status.json", help="Path to status JSON file (default: batch_extraction_status.json)" ) parser.add_argument( "--output", default="docs/feats/wrapup_pipeline_v2/VALIDATION_REPORT.md", help="Output markdown file path" ) args = parser.parse_args() status_file = Path(args.status_file) if not status_file.exists(): print(f"[ERROR] Status file not found: {status_file}") return 1 report_gen = ValidationReport(status_file) output_path = report_gen.generate_markdown_report(Path(args.output)) print(f"[OK] Validation report generated: {output_path}") return 0 if __name__ == "__main__": import sys sys.exit(main())