116 lines
3.3 KiB
Python
116 lines
3.3 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
@dataclass
|
|
class PcfBlock:
|
|
"""PCF 中的一个顶层段。"""
|
|
|
|
block_type: str
|
|
value: str = ""
|
|
fields: dict[str, list[str]] = field(default_factory=dict)
|
|
line_no: int = 0
|
|
|
|
def first(self, key: str, default: str = "") -> str:
|
|
values = self.fields.get(key, [])
|
|
return values[0] if values else default
|
|
|
|
|
|
@dataclass
|
|
class WeldRecord:
|
|
"""内部标准焊口记录。"""
|
|
|
|
source_file: str
|
|
pipeline_reference: str = ""
|
|
piping_spec: str = ""
|
|
line_no: str = ""
|
|
weld_no: str = ""
|
|
diameter: str = ""
|
|
outside_diameter: str = ""
|
|
wall_thickness: str = ""
|
|
spec: str = ""
|
|
weld_area_raw: str = ""
|
|
weld_area: str = ""
|
|
contractor_raw: str = ""
|
|
component_identifier: str = ""
|
|
master_component_identifier: str = ""
|
|
skey: str = ""
|
|
uci: str = ""
|
|
material_1: str = ""
|
|
material_2: str = ""
|
|
issues: list[str] = field(default_factory=list)
|
|
raw_fields: dict[str, list[str]] = field(default_factory=dict)
|
|
|
|
def set_if_empty(self, field_name: str, value: str) -> None:
|
|
if hasattr(self, field_name) and not getattr(self, field_name) and value:
|
|
setattr(self, field_name, value)
|
|
|
|
|
|
@dataclass
|
|
class UnknownFieldCandidate:
|
|
"""等待 DeepSeek 识别的非标准字段候选。"""
|
|
|
|
source_field: str
|
|
section: str
|
|
sample_values: list[str]
|
|
context: str
|
|
|
|
|
|
@dataclass
|
|
class MappingDecision:
|
|
"""DeepSeek 字段映射结果。"""
|
|
|
|
source_field: str
|
|
target_field: str
|
|
confidence: float
|
|
reason: str = ""
|
|
sample_values: list[str] = field(default_factory=list)
|
|
|
|
@property
|
|
def is_usable(self) -> bool:
|
|
return bool(self.source_field and self.target_field and self.confidence >= 0)
|
|
|
|
|
|
@dataclass
|
|
class JobResult:
|
|
job_id: str
|
|
input_files: int
|
|
weld_count: int
|
|
warning_count: int
|
|
error_count: int
|
|
output_xls: Path
|
|
issues_csv: Path
|
|
report_json: Path
|
|
messages: list[str] = field(default_factory=list)
|
|
llm_enabled: bool = False
|
|
llm_mappings: list[MappingDecision] = field(default_factory=list)
|
|
import_headers: list[str] = field(default_factory=list)
|
|
import_columns: list[dict[str, str]] = field(default_factory=list)
|
|
import_rows: list[dict[str, str]] = field(default_factory=list)
|
|
issue_rows: list[dict[str, str]] = field(default_factory=list)
|
|
|
|
def to_public_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"job_id": self.job_id,
|
|
"input_files": self.input_files,
|
|
"weld_count": self.weld_count,
|
|
"warning_count": self.warning_count,
|
|
"error_count": self.error_count,
|
|
"messages": self.messages,
|
|
"llm_enabled": self.llm_enabled,
|
|
"tables": {
|
|
"import_headers": self.import_headers,
|
|
"import_columns": self.import_columns,
|
|
"import_rows": self.import_rows,
|
|
"issue_rows": self.issue_rows,
|
|
},
|
|
"downloads": {
|
|
"xls": f"/api/jobs/{self.job_id}/download/xls",
|
|
"issues": f"/api/jobs/{self.job_id}/download/issues",
|
|
"report": f"/api/jobs/{self.job_id}/download/report",
|
|
},
|
|
}
|