Export statistics to CSV file. Args: output_path: Output file path (default: work_dir/stats.csv) Returns: Path to exported CSV file
(self, output_path: Optional[str] = None)
| 408 | self.logger.info("=" * 60) |
| 409 | |
| 410 | def export_csv(self, output_path: Optional[str] = None) -> str: |
| 411 | """ |
| 412 | Export statistics to CSV file. |
| 413 | |
| 414 | Args: |
| 415 | output_path: Output file path (default: work_dir/stats.csv) |
| 416 | |
| 417 | Returns: |
| 418 | Path to exported CSV file |
| 419 | """ |
| 420 | if output_path is None: |
| 421 | output_path = os.path.join(self.work_dir, "pipeline_stats.csv") |
| 422 | |
| 423 | output_dir = os.path.dirname(output_path) |
| 424 | if output_dir: |
| 425 | os.makedirs(output_dir, exist_ok=True) |
| 426 | |
| 427 | # Collect all operator names |
| 428 | all_operators = sorted(self._operator_counters.keys()) |
| 429 | |
| 430 | with open(output_path, "w", newline="", encoding="utf-8") as f: |
| 431 | fieldnames = ["file_path", "source", "total", "rejected", "kept"] + list( |
| 432 | all_operators |
| 433 | ) |
| 434 | writer = csv.DictWriter(f, fieldnames=fieldnames) |
| 435 | writer.writeheader() |
| 436 | |
| 437 | # Write per-dataset stats |
| 438 | for ds in self._file_stats.values(): |
| 439 | row = { |
| 440 | "file_path": ds.path, |
| 441 | "source": ds.source, |
| 442 | "total": ds.total_count, |
| 443 | "rejected": ds.rejected_count, |
| 444 | "kept": ds.kept_count, |
| 445 | } |
| 446 | for op in all_operators: |
| 447 | row[op] = 0 |
| 448 | writer.writerow(row) |
| 449 | |
| 450 | # Write global summary row |
| 451 | summary_row = { |
| 452 | "file_path": "TOTAL", |
| 453 | "source": "ALL", |
| 454 | "total": self._pipeline_stats.total_input, |
| 455 | "rejected": self._pipeline_stats.total_rejected, |
| 456 | "kept": self._pipeline_stats.total_output, |
| 457 | } |
| 458 | for op in all_operators: |
| 459 | summary_row[op] = self._operator_counters[op]["rejected"] |
| 460 | writer.writerow(summary_row) |
| 461 | |
| 462 | self.logger.info(f"Statistics exported to: {output_path}") |
| 463 | return output_path |
| 464 | |
| 465 | def export_json(self, output_path: Optional[str] = None) -> str: |
| 466 | """ |