Dbmaxwell's picture
feat: deploy FinReport Agent with LightOnOCR review
9703837 verified
Raw
History Blame Contribute Delete
15.4 kB
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