466 lines
16 KiB
Python
466 lines
16 KiB
Python
from __future__ import annotations
|
|
|
|
import csv
|
|
import json
|
|
import shutil
|
|
import uuid
|
|
import zipfile
|
|
from pathlib import Path
|
|
from urllib.parse import unquote, urlparse
|
|
from urllib.request import url2pathname
|
|
|
|
import httpx
|
|
from fastapi import UploadFile
|
|
|
|
from .config import AppConfig, ROOT_DIR
|
|
from .deepseek import DeepSeekMapper, persist_high_confidence_mappings
|
|
from .excel_writer import XlsTemplateWriter
|
|
from .models import JobResult, WeldRecord
|
|
from .pcf_parser import PcfParser
|
|
|
|
|
|
JOBS_DIR = ROOT_DIR / "outputs" / "jobs"
|
|
UPLOADS_DIR = ROOT_DIR / "outputs" / "uploads"
|
|
MAX_ATTACHMENT_BYTES = 100 * 1024 * 1024
|
|
|
|
|
|
class JobStore:
|
|
def __init__(self) -> None:
|
|
self.results: dict[str, JobResult] = {}
|
|
|
|
def set(self, result: JobResult) -> None:
|
|
self.results[result.job_id] = result
|
|
|
|
def get(self, job_id: str) -> JobResult | None:
|
|
return self.results.get(job_id)
|
|
|
|
|
|
job_store = JobStore()
|
|
|
|
|
|
def safe_filename(name: str) -> str:
|
|
keep = []
|
|
for ch in Path(name).name:
|
|
if ch.isalnum() or ch in ".-_() ":
|
|
keep.append(ch)
|
|
else:
|
|
keep.append("_")
|
|
return "".join(keep) or "attachment.pcf"
|
|
|
|
|
|
async def process_uploads(files: list[UploadFile]) -> JobResult:
|
|
job_id = uuid.uuid4().hex[:12]
|
|
upload_dir = UPLOADS_DIR / job_id
|
|
upload_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
saved_files: list[Path] = []
|
|
messages: list[str] = []
|
|
for upload in files:
|
|
filename = safe_filename(upload.filename or "upload.pcf")
|
|
target = upload_dir / filename
|
|
if target.suffix.lower() != ".pcf":
|
|
messages.append(f"拒绝非 PCF 文件：{filename}")
|
|
continue
|
|
with target.open("wb") as f:
|
|
shutil.copyfileobj(upload.file, f)
|
|
saved_files.append(target)
|
|
|
|
return await process_pcf_paths(job_id=job_id, pcf_files=saved_files, initial_messages=messages)
|
|
|
|
|
|
async def analyze_attachment_urls(urls: list[str]) -> dict:
|
|
job_id = uuid.uuid4().hex[:12]
|
|
download_dir = UPLOADS_DIR / job_id
|
|
download_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
pcf_files, messages = await download_attachment_urls(urls, download_dir)
|
|
analysis = await analyze_pcf_paths(pcf_files=pcf_files, initial_messages=messages)
|
|
return analysis_to_json_payload(job_id=job_id, input_files=len(pcf_files), analysis=analysis)
|
|
|
|
|
|
async def process_pcf_paths(
|
|
job_id: str,
|
|
pcf_files: list[Path],
|
|
initial_messages: list[str] | None = None,
|
|
) -> JobResult:
|
|
job_dir = JOBS_DIR / job_id
|
|
job_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
analysis = await analyze_pcf_paths(pcf_files=pcf_files, initial_messages=initial_messages)
|
|
config = analysis["config"]
|
|
records = analysis["records"]
|
|
candidates = analysis["candidates"]
|
|
decisions = analysis["decisions"]
|
|
messages = analysis["messages"]
|
|
import_headers = analysis["import_headers"]
|
|
import_columns = analysis["import_columns"]
|
|
import_rows = analysis["import_rows"]
|
|
issue_rows = analysis["issue_rows"]
|
|
mapper_enabled = analysis["llm_enabled"]
|
|
|
|
output_xls = job_dir / "焊接数据导入结果.xls"
|
|
issues_csv = job_dir / "问题清单.csv"
|
|
report_json = job_dir / "解析报告.json"
|
|
|
|
writer = XlsTemplateWriter(config)
|
|
writer.write(records, output_xls)
|
|
write_issues_csv(issue_rows, issues_csv)
|
|
write_report_json(
|
|
records=records,
|
|
candidates=candidates,
|
|
decisions=decisions,
|
|
accepted_mappings=analysis["accepted_mappings"],
|
|
messages=messages,
|
|
report_path=report_json,
|
|
llm_enabled=mapper_enabled,
|
|
)
|
|
|
|
warning_count = analysis["warning_count"]
|
|
error_count = analysis["error_count"]
|
|
|
|
result = JobResult(
|
|
job_id=job_id,
|
|
input_files=len(pcf_files),
|
|
weld_count=len(records),
|
|
warning_count=warning_count,
|
|
error_count=error_count,
|
|
output_xls=output_xls,
|
|
issues_csv=issues_csv,
|
|
report_json=report_json,
|
|
messages=messages,
|
|
llm_enabled=mapper_enabled,
|
|
llm_mappings=decisions,
|
|
import_headers=import_headers,
|
|
import_columns=import_columns,
|
|
import_rows=import_rows,
|
|
issue_rows=issue_rows,
|
|
)
|
|
job_store.set(result)
|
|
return result
|
|
|
|
|
|
async def analyze_pcf_paths(
|
|
pcf_files: list[Path],
|
|
initial_messages: list[str] | None = None,
|
|
) -> dict:
|
|
messages = list(initial_messages or [])
|
|
config = AppConfig.load()
|
|
parser = PcfParser(config)
|
|
records, candidates, parse_messages = parser.parse_files(pcf_files)
|
|
messages.extend(parse_messages)
|
|
|
|
mapper = DeepSeekMapper(config)
|
|
decisions = []
|
|
accepted_mappings = {}
|
|
if mapper.enabled and candidates:
|
|
try:
|
|
decisions = await mapper.map_unknown_fields(candidates)
|
|
accepted_mappings = persist_high_confidence_mappings(config, decisions)
|
|
if accepted_mappings:
|
|
config = AppConfig.load()
|
|
parser = PcfParser(config)
|
|
records, _, parse_messages = parser.parse_files(pcf_files, accepted_mappings)
|
|
messages.extend(parse_messages)
|
|
except Exception as exc:
|
|
messages.append(f"DeepSeek 字段映射失败，已降级为规则解析：{exc}")
|
|
elif candidates:
|
|
messages.append("未在 config/mapping.yaml 配置 deepseek.api_key，已跳过非标准字段智能映射")
|
|
|
|
writer = XlsTemplateWriter(config)
|
|
import_headers, import_columns, import_rows = writer.build_import_rows(records)
|
|
issue_rows = build_issue_rows(records, candidates, decisions, config.confidence_threshold)
|
|
|
|
warning_count = sum(len(r.issues) for r in records) + len(messages)
|
|
error_count = 0 if records else 1
|
|
if not records:
|
|
messages.append("未解析到任何焊口记录")
|
|
|
|
return {
|
|
"config": config,
|
|
"records": records,
|
|
"candidates": candidates,
|
|
"decisions": decisions,
|
|
"accepted_mappings": accepted_mappings,
|
|
"messages": messages,
|
|
"llm_enabled": mapper.enabled,
|
|
"import_headers": import_headers,
|
|
"import_columns": import_columns,
|
|
"import_rows": import_rows,
|
|
"issue_rows": issue_rows,
|
|
"warning_count": warning_count,
|
|
"error_count": error_count,
|
|
}
|
|
|
|
|
|
async def download_attachment_urls(urls: list[str], target_dir: Path) -> tuple[list[Path], list[str]]:
|
|
pcf_files: list[Path] = []
|
|
messages: list[str] = []
|
|
async with httpx.AsyncClient(follow_redirects=True, timeout=60) as client:
|
|
for index, url in enumerate(urls, start=1):
|
|
parsed = urlparse(url)
|
|
if is_local_attachment(url, parsed):
|
|
local_files, local_messages = collect_local_attachment(url, target_dir, index)
|
|
pcf_files.extend(local_files)
|
|
messages.extend(local_messages)
|
|
continue
|
|
|
|
if parsed.scheme not in {"http", "https"}:
|
|
messages.append(f"附件引用协议不支持：{url}")
|
|
continue
|
|
|
|
downloaded, download_messages = await download_remote_attachment(client, url, target_dir, index)
|
|
pcf_files.extend(downloaded)
|
|
messages.extend(download_messages)
|
|
return pcf_files, messages
|
|
|
|
|
|
def is_local_attachment(raw: str, parsed) -> bool:
|
|
if parsed.scheme == "file":
|
|
return True
|
|
if parsed.scheme in {"http", "https"}:
|
|
return False
|
|
if len(raw) >= 3 and raw[1] == ":" and raw[2] in {"\\", "/"}:
|
|
return True
|
|
if raw.startswith("\\\\"):
|
|
return True
|
|
return parsed.scheme == ""
|
|
|
|
|
|
def collect_local_attachment(raw: str, target_dir: Path, index: int) -> tuple[list[Path], list[str]]:
|
|
messages: list[str] = []
|
|
pcf_files: list[Path] = []
|
|
source = local_path_from_reference(raw)
|
|
|
|
if not source.exists() or not source.is_file():
|
|
messages.append(f"本地附件不存在或不是文件：{raw}")
|
|
return pcf_files, messages
|
|
if source.stat().st_size > MAX_ATTACHMENT_BYTES:
|
|
messages.append(f"本地附件超过大小限制：{source}")
|
|
return pcf_files, messages
|
|
|
|
target = target_dir / safe_filename(source.name or f"attachment_{index}.pcf")
|
|
if source.resolve() != target.resolve():
|
|
shutil.copyfile(source, target)
|
|
else:
|
|
target = source
|
|
|
|
if target.suffix.lower() == ".pcf":
|
|
pcf_files.append(target)
|
|
elif target.suffix.lower() == ".zip":
|
|
extracted, extract_messages = extract_pcf_zip(target, target_dir / f"zip_{index}")
|
|
pcf_files.extend(extracted)
|
|
messages.extend(extract_messages)
|
|
else:
|
|
messages.append(f"本地附件不是 PCF 或 ZIP，已跳过：{source.name}")
|
|
return pcf_files, messages
|
|
|
|
|
|
def local_path_from_reference(raw: str) -> Path:
|
|
parsed = urlparse(raw)
|
|
if parsed.scheme == "file":
|
|
if parsed.netloc and parsed.netloc not in {"localhost", "127.0.0.1"}:
|
|
return Path(f"//{parsed.netloc}{url2pathname(parsed.path)}")
|
|
return Path(url2pathname(unquote(parsed.path)))
|
|
return Path(raw)
|
|
|
|
|
|
async def download_remote_attachment(
|
|
client: httpx.AsyncClient,
|
|
url: str,
|
|
target_dir: Path,
|
|
index: int,
|
|
) -> tuple[list[Path], list[str]]:
|
|
pcf_files: list[Path] = []
|
|
messages: list[str] = []
|
|
try:
|
|
response = await client.get(url)
|
|
response.raise_for_status()
|
|
except Exception as exc:
|
|
messages.append(f"附件下载失败：{url}，原因：{exc}")
|
|
return pcf_files, messages
|
|
|
|
content = response.content
|
|
if len(content) > MAX_ATTACHMENT_BYTES:
|
|
messages.append(f"附件超过大小限制：{url}")
|
|
return pcf_files, messages
|
|
|
|
filename = filename_from_response(url, response.headers, index)
|
|
saved = target_dir / filename
|
|
saved.write_bytes(content)
|
|
|
|
if saved.suffix.lower() == ".pcf":
|
|
pcf_files.append(saved)
|
|
elif saved.suffix.lower() == ".zip":
|
|
extracted, extract_messages = extract_pcf_zip(saved, target_dir / f"zip_{index}")
|
|
pcf_files.extend(extracted)
|
|
messages.extend(extract_messages)
|
|
else:
|
|
messages.append(f"附件不是 PCF 或 ZIP，已跳过：{filename}")
|
|
return pcf_files, messages
|
|
|
|
|
|
def filename_from_response(url: str, headers: httpx.Headers, index: int) -> str:
|
|
disposition = headers.get("content-disposition", "")
|
|
marker = "filename="
|
|
if marker in disposition:
|
|
raw = disposition.split(marker, 1)[1].strip().strip('"')
|
|
return safe_filename(unquote(raw))
|
|
parsed_name = Path(unquote(urlparse(url).path)).name
|
|
return safe_filename(parsed_name or f"attachment_{index}.pcf")
|
|
|
|
|
|
def extract_pcf_zip(zip_path: Path, target_dir: Path) -> tuple[list[Path], list[str]]:
|
|
target_dir.mkdir(parents=True, exist_ok=True)
|
|
pcf_files: list[Path] = []
|
|
messages: list[str] = []
|
|
try:
|
|
with zipfile.ZipFile(zip_path) as archive:
|
|
for member in archive.infolist():
|
|
if member.is_dir():
|
|
continue
|
|
name = safe_filename(Path(member.filename).name)
|
|
if not name.lower().endswith(".pcf"):
|
|
continue
|
|
target = target_dir / name
|
|
with archive.open(member) as src, target.open("wb") as dst:
|
|
shutil.copyfileobj(src, dst)
|
|
pcf_files.append(target)
|
|
except zipfile.BadZipFile:
|
|
messages.append(f"ZIP 附件无法解压：{zip_path.name}")
|
|
if not pcf_files:
|
|
messages.append(f"ZIP 附件中未找到 PCF 文件：{zip_path.name}")
|
|
return pcf_files, messages
|
|
|
|
|
|
def analysis_to_json_payload(job_id: str, input_files: int, analysis: dict) -> dict:
|
|
records: list[WeldRecord] = analysis["records"]
|
|
return {
|
|
"job_id": job_id,
|
|
"input_files": input_files,
|
|
"weld_count": len(records),
|
|
"warning_count": analysis["warning_count"],
|
|
"error_count": analysis["error_count"],
|
|
"messages": analysis["messages"],
|
|
"llm_enabled": analysis["llm_enabled"],
|
|
"tables": {
|
|
"import_headers": analysis["import_headers"],
|
|
"import_columns": analysis["import_columns"],
|
|
"import_rows": analysis["import_rows"],
|
|
"issue_rows": analysis["issue_rows"],
|
|
},
|
|
"report": build_report_payload(
|
|
records=records,
|
|
candidates=analysis["candidates"],
|
|
decisions=analysis["decisions"],
|
|
accepted_mappings=analysis["accepted_mappings"],
|
|
messages=analysis["messages"],
|
|
llm_enabled=analysis["llm_enabled"],
|
|
),
|
|
}
|
|
|
|
|
|
def build_issue_rows(
|
|
records: list[WeldRecord],
|
|
candidates,
|
|
decisions,
|
|
threshold: float,
|
|
) -> list[dict[str, str]]:
|
|
rows: list[dict[str, str]] = []
|
|
for record in records:
|
|
for issue in record.issues:
|
|
rows.append(
|
|
{
|
|
"类型": "解析警告",
|
|
"文件": record.source_file,
|
|
"焊口代号": record.weld_no,
|
|
"字段": "",
|
|
"内容": issue,
|
|
"处理状态": "需复核",
|
|
}
|
|
)
|
|
high_conf = {d.source_field for d in decisions if d.confidence >= threshold}
|
|
for candidate in candidates:
|
|
rows.append(
|
|
{
|
|
"类型": "未知字段",
|
|
"文件": "",
|
|
"焊口代号": "",
|
|
"字段": candidate.source_field,
|
|
"内容": "; ".join(candidate.sample_values),
|
|
"处理状态": "已自动映射" if candidate.source_field in high_conf else "未映射/需复核",
|
|
}
|
|
)
|
|
return rows
|
|
|
|
|
|
def write_issues_csv(issue_rows: list[dict[str, str]], path: Path) -> None:
|
|
with path.open("w", encoding="utf-8-sig", newline="") as f:
|
|
writer = csv.writer(f)
|
|
headers = ["类型", "文件", "焊口代号", "字段", "内容", "处理状态"]
|
|
writer.writerow(headers)
|
|
for row in issue_rows:
|
|
writer.writerow([row.get(header, "") for header in headers])
|
|
|
|
|
|
def write_report_json(
|
|
records: list[WeldRecord],
|
|
candidates,
|
|
decisions,
|
|
accepted_mappings: dict[str, str],
|
|
messages: list[str],
|
|
report_path: Path,
|
|
llm_enabled: bool,
|
|
) -> None:
|
|
payload = build_report_payload(
|
|
records=records,
|
|
candidates=candidates,
|
|
decisions=decisions,
|
|
accepted_mappings=accepted_mappings,
|
|
messages=messages,
|
|
llm_enabled=llm_enabled,
|
|
)
|
|
with report_path.open("w", encoding="utf-8") as f:
|
|
json.dump(payload, f, ensure_ascii=False, indent=2)
|
|
|
|
|
|
def build_report_payload(
|
|
records: list[WeldRecord],
|
|
candidates,
|
|
decisions,
|
|
accepted_mappings: dict[str, str],
|
|
messages: list[str],
|
|
llm_enabled: bool,
|
|
) -> dict:
|
|
return {
|
|
"summary": {
|
|
"weld_count": len(records),
|
|
"source_files": sorted({r.source_file for r in records}),
|
|
"llm_enabled": llm_enabled,
|
|
"unknown_field_count": len(candidates),
|
|
},
|
|
"messages": messages,
|
|
"accepted_mappings": accepted_mappings,
|
|
"llm_mappings": [
|
|
{
|
|
"source_field": d.source_field,
|
|
"target_field": d.target_field,
|
|
"confidence": d.confidence,
|
|
"reason": d.reason,
|
|
"sample_values": d.sample_values,
|
|
}
|
|
for d in decisions
|
|
],
|
|
"records_preview": [
|
|
{
|
|
"source_file": r.source_file,
|
|
"pipeline_reference": r.pipeline_reference,
|
|
"line_no": r.line_no,
|
|
"weld_no": r.weld_no,
|
|
"diameter": r.diameter,
|
|
"weld_area": r.weld_area,
|
|
"issues": r.issues,
|
|
}
|
|
for r in records[:20]
|
|
],
|
|
}
|