| from __future__ import annotations |
|
|
| from datetime import date |
| from decimal import Decimal |
| from pathlib import Path |
|
|
| import xlsxwriter |
|
|
| from .excel import ( |
| BORDER, |
| LIGHT, |
| NAVY, |
| PALE_BLUE, |
| SLATE, |
| WHITE, |
| _add_title, |
| _as_datetime, |
| _configure_print, |
| _formats, |
| _status_format, |
| ) |
| from .models import GenericDocumentReport, Severity, ValidationResult |
|
|
|
|
| def _excel_value(value: Decimal | date | str | None) -> float | str: |
| if isinstance(value, Decimal): |
| return float(value) |
| return "" if value is None else value |
|
|
|
|
| def _generic_formats(workbook: xlsxwriter.Workbook, formats: dict) -> dict: |
| formats.update( |
| { |
| "row_number": workbook.add_format( |
| { |
| "num_format": '#,##0.######;[Red]-#,##0.######;[Color10]-', |
| "border": 1, |
| "border_color": BORDER, |
| } |
| ), |
| "row_number_bold": workbook.add_format( |
| { |
| "bold": True, |
| "font_color": NAVY, |
| "bg_color": PALE_BLUE, |
| "num_format": '#,##0.######;[Red]-#,##0.######;[Color10]-', |
| "border": 1, |
| "border_color": BORDER, |
| } |
| ), |
| "hero": workbook.add_format( |
| { |
| "bold": True, |
| "font_size": 26, |
| "font_color": WHITE, |
| "bg_color": NAVY, |
| "valign": "vcenter", |
| } |
| ), |
| "metric": workbook.add_format( |
| { |
| "bold": True, |
| "font_size": 18, |
| "font_color": NAVY, |
| "bg_color": LIGHT, |
| "align": "center", |
| "valign": "vcenter", |
| "border": 1, |
| "border_color": BORDER, |
| } |
| ), |
| } |
| ) |
| for level in range(4): |
| formats[f"row_label_{level}"] = workbook.add_format( |
| { |
| "font_color": NAVY, |
| "border": 1, |
| "border_color": BORDER, |
| "text_wrap": True, |
| "valign": "vcenter", |
| "indent": level, |
| } |
| ) |
| formats[f"row_section_{level}"] = workbook.add_format( |
| { |
| "bold": True, |
| "font_color": NAVY, |
| "bg_color": PALE_BLUE, |
| "border": 1, |
| "border_color": BORDER, |
| "text_wrap": True, |
| "valign": "vcenter", |
| "indent": level, |
| } |
| ) |
| return formats |
|
|
|
|
| def _write_summary(workbook, report, validation, formats) -> None: |
| sheet = workbook.add_worksheet("Özet") |
| sheet.hide_gridlines(2) |
| sheet.merge_range("A1:G2", report.document_title, formats["hero"]) |
| sheet.set_row(0, 34) |
| sheet.set_row(1, 18) |
| sheet.set_column("A:A", 18) |
| sheet.set_column("B:B", 24) |
| sheet.set_column("C:C", 43) |
| sheet.set_column("D:D", 16) |
| sheet.set_column("E:G", 13) |
|
|
| sheet.write("A4", "Belge türü", formats["label"]) |
| sheet.merge_range("B4:D4", report.document_type, formats["text"]) |
| sheet.write("E4", "İşlem durumu", formats["label"]) |
| sheet.merge_range("F4:G4", validation.status, _status_format(formats, validation)) |
| sheet.write("A5", "Kaynak dosya", formats["label"]) |
| sheet.merge_range("B5:D5", report.source_filename, formats["text"]) |
| sheet.write("E5", "Planlama modu", formats["label"]) |
| planner_label = ( |
| "HF yapılandırma ajanı" |
| if report.planner_mode == "hf_structured_agent" |
| else "Deterministik yedek plan" |
| ) |
| sheet.merge_range("F5:G5", planner_label, formats["text"]) |
| sheet.write("A6", "Belge özeti", formats["label"]) |
| sheet.merge_range("B6:G6", report.summary, formats["wrapped_text"]) |
| sheet.set_row(5, 34) |
|
|
| sheet.merge_range("A8:B9", len(report.tables), formats["metric"]) |
| sheet.merge_range( |
| "C8:D9", |
| sum(len(table.rows) for table in report.tables), |
| formats["metric"], |
| ) |
| sheet.merge_range( |
| "E8:G9", |
| f"{validation.error_count} hata · {validation.warning_count} uyarı", |
| formats["metric"], |
| ) |
| sheet.merge_range("A10:B10", "Keşfedilen tablo", formats["subtitle"]) |
| sheet.merge_range("C10:D10", "Aktarılan satır", formats["subtitle"]) |
| sheet.merge_range("E10:G10", "Kontrol sonucu", formats["subtitle"]) |
|
|
| headers = ["#", "Çalışma Sayfası", "Tablo Başlığı", "Kaynak Sayfa", "Satır", "Sütun", "Güven"] |
| header_row = 12 |
| for column, header in enumerate(headers): |
| sheet.write(header_row, column, header, formats["table_header"]) |
| for index, table in enumerate(report.tables, start=1): |
| row = header_row + index |
| sheet.write_number(row, 0, index, formats["integer"]) |
| sheet.write_url( |
| row, |
| 1, |
| f"internal:'{table.sheet_name}'!A1", |
| formats["text_border"], |
| table.sheet_name, |
| ) |
| sheet.write(row, 2, table.title, formats["wrapped_text"]) |
| sheet.write(row, 3, ", ".join(map(str, table.source_pages)), formats["text_border"]) |
| sheet.write_number(row, 4, len(table.rows), formats["integer"]) |
| sheet.write_number(row, 5, len(table.headers), formats["integer"]) |
| sheet.write_number(row, 6, table.confidence, formats["percent"]) |
| sheet.add_table( |
| header_row, |
| 0, |
| header_row + len(report.tables), |
| 6, |
| { |
| "name": "DiscoveredTables", |
| "style": "Table Style Medium 2", |
| "columns": [ |
| {"header": header, "header_format": formats["table_header"]} |
| for header in headers |
| ], |
| }, |
| ) |
| sheet.freeze_panes(header_row + 1, 0) |
| _configure_print(sheet, repeat_header_row=header_row) |
|
|
|
|
| def _is_section_row(label: str, hierarchy_level: int) -> bool: |
| letters = "".join(character for character in label if character.isalpha()) |
| return hierarchy_level == 0 and bool(letters) and letters == letters.upper() |
|
|
|
|
| def _write_table(workbook, table, formats, table_index: int) -> None: |
| sheet = workbook.add_worksheet(table.sheet_name) |
| has_notes = any(row.note_reference for row in table.rows) |
| value_offset = 2 if has_notes else 1 |
| source_column = value_offset + len(table.headers) |
| last_column = source_column |
| subtitle = ( |
| f"Kaynak sayfa: {', '.join(map(str, table.source_pages))}" |
| + (f" · Birim: {table.unit}" if table.unit else "") |
| ) |
| _add_title(sheet, table.title, subtitle, last_column, formats) |
| if table.description: |
| sheet.merge_range(2, 0, 2, last_column, table.description, formats["subtitle"]) |
| sheet.set_row(2, 28) |
| headers = ["Finansal Kalem"] |
| if has_notes: |
| headers.append("Dipnot Ref.") |
| headers.extend(table.headers) |
| headers.append("Kaynak Sayfa") |
| header_row = 3 |
| for column, header in enumerate(headers): |
| sheet.write(header_row, column, header, formats["table_header"]) |
| for row_index, item in enumerate(table.rows, start=header_row + 1): |
| section = _is_section_row(item.label, item.hierarchy_level) |
| style = "row_section" if section else "row_label" |
| label_format = formats[f"{style}_{min(item.hierarchy_level, 3)}"] |
| sheet.write(row_index, 0, item.label, label_format) |
| if has_notes: |
| sheet.write(row_index, 1, item.note_reference or "", formats["text_border"]) |
| for offset, value in enumerate(item.values): |
| column = value_offset + offset |
| if isinstance(value, Decimal): |
| number_format = formats["row_number_bold"] if section else formats["row_number"] |
| sheet.write_number(row_index, column, float(value), number_format) |
| elif isinstance(value, date): |
| sheet.write_datetime( |
| row_index, |
| column, |
| _as_datetime(value), |
| formats["date"], |
| ) |
| else: |
| sheet.write(row_index, column, _excel_value(value), formats["text_border"]) |
| sheet.write_number(row_index, source_column, item.source.page, formats["integer"]) |
| if len(item.label) > 72: |
| sheet.set_row(row_index, 30) |
| last_row = header_row + len(table.rows) |
| table_name = f"FinancialTable{table_index}" |
| sheet.add_table( |
| header_row, |
| 0, |
| last_row, |
| last_column, |
| { |
| "name": table_name, |
| "style": "Table Style Medium 2", |
| "columns": [ |
| {"header": header, "header_format": formats["table_header"]} |
| for header in headers |
| ], |
| }, |
| ) |
| sheet.set_column(0, 0, 48) |
| if has_notes: |
| sheet.set_column(1, 1, 12) |
| sheet.set_column(value_offset, source_column - 1, 18) |
| sheet.set_column(source_column, source_column, 13) |
| if {"Döviz", "ISIN", "Toplam Değer"}.issubset(table.headers): |
| sheet.set_column(0, 0, 15) |
| compact_widths = { |
| "Döviz": 9, |
| "İhraççı Kurum": 28, |
| "ISIN": 17, |
| "Vade Tarihi": 13, |
| "Vadeye Kalan Gün": 12, |
| "Nominal Faiz (%)": 13, |
| "Faiz Ödeme Sayısı": 12, |
| "Nominal Değer": 15, |
| "Birim Alış Fiyatı": 15, |
| "Satın Alış Tarihi": 14, |
| "Repo Teminatı": 15, |
| "Borsa Sözleşme No": 17, |
| "Net Dönüş": 14, |
| "Günlük Birim Değer": 15, |
| "Toplam Değer": 16, |
| "Grup (%)": 11, |
| "FPD (%)": 11, |
| "FTD (%)": 11, |
| } |
| for offset, header in enumerate(table.headers): |
| sheet.set_column( |
| value_offset + offset, |
| value_offset + offset, |
| compact_widths.get(header, 14), |
| ) |
| sheet.set_column(source_column, source_column, 11) |
| sheet.set_row(header_row, 46) |
| sheet.freeze_panes(header_row + 1, value_offset) |
| _configure_print(sheet, repeat_header_row=header_row) |
|
|
|
|
| def _write_findings(workbook, report, validation, formats) -> None: |
| sheet = workbook.add_worksheet("Kontroller") |
| _add_title( |
| sheet, |
| "Çıkarım Kontrolleri", |
| f"{validation.error_count} hata · {validation.warning_count} uyarı · {validation.info_count} bilgi", |
| 7, |
| formats, |
| ) |
| sheet.write("A3", "SHA-256", formats["label"]) |
| sheet.merge_range("B3:H3", report.source_sha256, formats["hash"]) |
| headers = [ |
| "Seviye", |
| "Kod", |
| "Açıklama", |
| "PDF Sayfası", |
| "Kayıt", |
| "Beklenen", |
| "Gerçekleşen", |
| "Çalışma Sayfası", |
| ] |
| header_row = 5 |
| for column, header in enumerate(headers): |
| sheet.write(header_row, column, header, formats["table_header"]) |
| for row_index, finding in enumerate(validation.findings, start=header_row + 1): |
| level = { |
| Severity.ERROR: "Hata", |
| Severity.WARNING: "Uyarı", |
| Severity.INFO: "Bilgi", |
| }[finding.severity] |
| sheet.write(row_index, 0, level, formats["text_border"]) |
| sheet.write(row_index, 1, finding.code, formats["text_border"]) |
| sheet.write(row_index, 2, finding.message, formats["wrapped_text"]) |
| sheet.write(row_index, 3, finding.page or "", formats["text_border"]) |
| sheet.write(row_index, 4, finding.record_key or "", formats["text_border"]) |
| sheet.write(row_index, 5, finding.expected or "", formats["text_border"]) |
| sheet.write(row_index, 6, finding.actual or "", formats["text_border"]) |
| sheet.write(row_index, 7, finding.sheet or "", formats["text_border"]) |
| if validation.findings: |
| sheet.add_table( |
| header_row, |
| 0, |
| header_row + len(validation.findings), |
| 7, |
| { |
| "name": "GenericValidation", |
| "style": "Table Style Medium 2", |
| "columns": [ |
| {"header": header, "header_format": formats["table_header"]} |
| for header in headers |
| ], |
| }, |
| ) |
| sheet.set_column("A:B", 20) |
| sheet.set_column("C:C", 75) |
| sheet.set_column("D:H", 18) |
| sheet.freeze_panes(header_row + 1, 0) |
| _configure_print(sheet, repeat_header_row=header_row) |
|
|
|
|
| def _write_provenance(workbook, report, formats) -> None: |
| sheet = workbook.add_worksheet("Kaynak İzleri") |
| _add_title( |
| sheet, |
| "Satır Kaynak İzleri", |
| "Her finansal satır PDF sayfası ve koordinat kutusuyla eşleştirilir", |
| 8, |
| formats, |
| ) |
| headers = ["Çalışma Sayfası", "Tablo", "Satır Etiketi", "PDF Sayfası", "X0", "Y0", "X1", "Y1", "Yöntem"] |
| header_row = 3 |
| for column, header in enumerate(headers): |
| sheet.write(header_row, column, header, formats["table_header"]) |
| records = [ |
| (table, row) |
| for table in report.tables |
| for row in table.rows |
| ] |
| for row_index, (table, row) in enumerate(records, start=header_row + 1): |
| values = [table.sheet_name, table.title, row.label] |
| for column, value in enumerate(values): |
| sheet.write(row_index, column, value, formats["text_border"]) |
| sheet.write_number(row_index, 3, row.source.page, formats["integer"]) |
| for offset, coordinate in enumerate(row.source.bbox): |
| sheet.write_number(row_index, 4 + offset, coordinate, formats["decimal"]) |
| sheet.write(row_index, 8, row.source.method.value, formats["text_border"]) |
| sheet.add_table( |
| header_row, |
| 0, |
| header_row + len(records), |
| 8, |
| { |
| "name": "GenericProvenance", |
| "style": "Table Style Medium 2", |
| "columns": [ |
| {"header": header, "header_format": formats["table_header"]} |
| for header in headers |
| ], |
| }, |
| ) |
| sheet.set_column("A:B", 26) |
| sheet.set_column("C:C", 55) |
| sheet.set_column("D:I", 15) |
| sheet.freeze_panes(header_row + 1, 0) |
| _configure_print(sheet, repeat_header_row=header_row) |
|
|
|
|
| def write_generic_workbook( |
| report: GenericDocumentReport, |
| validation: ValidationResult, |
| output_path: str | Path, |
| ) -> Path: |
| path = Path(output_path).expanduser().resolve() |
| path.parent.mkdir(parents=True, exist_ok=True) |
| workbook = xlsxwriter.Workbook(path, {"nan_inf_to_errors": True}) |
| workbook.set_properties( |
| { |
| "title": report.document_title, |
| "subject": "FinReport Agent tarafından kaynak PDF'den oluşturuldu", |
| "author": "FinReport Agent", |
| "company": "FinReport Agent", |
| "comments": f"Kaynak SHA-256: {report.source_sha256}", |
| } |
| ) |
| formats = _generic_formats(workbook, _formats(workbook)) |
| try: |
| _write_summary(workbook, report, validation, formats) |
| for index, table in enumerate(report.tables, start=1): |
| _write_table(workbook, table, formats, index) |
| _write_findings(workbook, report, validation, formats) |
| _write_provenance(workbook, report, formats) |
| finally: |
| workbook.close() |
| return path |
|
|