from __future__ import annotations import json from typing import Any import httpx from .config import AppConfig from .models import MappingDecision, UnknownFieldCandidate from .pcf_parser import STANDARD_TARGET_FIELDS class DeepSeekMapper: def __init__(self, config: AppConfig): self.config = config self.api_key = str(config.deepseek.get("api_key", "") or "").strip() self.base_url = config.deepseek.get("base_url", "https://api.deepseek.com").rstrip("/") self.model = config.deepseek.get("model", "deepseek-v4-flash") @property def enabled(self) -> bool: return bool(self.api_key) async def map_unknown_fields( self, candidates: list[UnknownFieldCandidate] ) -> list[MappingDecision]: if not self.enabled or not candidates: return [] payload = { "standard_fields": sorted(STANDARD_TARGET_FIELDS), "candidates": [ { "source_field": c.source_field, "section": c.section, "sample_values": c.sample_values, "context": c.context, } for c in candidates[:50] ], } prompt = ( "你是 PCF 管道文件字段映射助手。请把未知 PCF 字段映射到标准字段。" "只在语义和样例值明确时映射;不确定时 target_field 返回空字符串,confidence 低于 0.85。" "必须返回 JSON,格式为 {\"mappings\": [{\"source_field\": str, \"target_field\": str, " "\"confidence\": number, \"reason\": str, \"sample_values\": [str]}]}。\n\n" f"输入:{json.dumps(payload, ensure_ascii=False)}" ) response = await self._chat_json(prompt) raw_mappings = response.get("mappings", response if isinstance(response, list) else []) decisions: list[MappingDecision] = [] for item in raw_mappings: if not isinstance(item, dict): continue try: decision = MappingDecision( source_field=str(item.get("source_field", "")).strip(), target_field=str(item.get("target_field", "")).strip(), confidence=float(item.get("confidence", 0)), reason=str(item.get("reason", "")).strip(), sample_values=[str(v) for v in item.get("sample_values", [])], ) except (TypeError, ValueError): continue if decision.is_usable: decisions.append(decision) return decisions async def _chat_json(self, prompt: str) -> dict[str, Any] | list[Any]: url = f"{self.base_url}/chat/completions" headers = { "Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json", } body = { "model": self.model, "messages": [ {"role": "system", "content": "只输出合法 JSON,不要输出 Markdown。"}, {"role": "user", "content": prompt}, ], "response_format": {"type": "json_object"}, "temperature": 0, } async with httpx.AsyncClient(timeout=60) as client: resp = await client.post(url, headers=headers, json=body) resp.raise_for_status() content = resp.json()["choices"][0]["message"]["content"] return json.loads(content) def persist_high_confidence_mappings( config: AppConfig, decisions: list[MappingDecision], threshold: float | None = None, ) -> dict[str, str]: threshold = config.confidence_threshold if threshold is None else threshold accepted: dict[str, str] = {} changed = False for decision in decisions: if ( decision.confidence >= threshold and decision.target_field in STANDARD_TARGET_FIELDS and decision.source_field not in config.field_mappings ): current = config.dynamic_field_mappings.get(decision.source_field) if current and current != decision.target_field: continue config.dynamic_field_mappings[decision.source_field] = decision.target_field accepted[decision.source_field] = decision.target_field changed = True if changed: config.save() return accepted