labor-arbitration-system/backend/app/services/nlp_client.py

import re
from typing import Any

import requests

from app.config import settings


class NLPClient:
    def extract(self, text: str) -> dict[str, Any]:
        if settings.use_remote_nlp_service:
            return self._extract_remote(text)
        return self._extract_local_rule_based(text)

    def _extract_remote(self, text: str) -> dict[str, Any]:
        payload = {"text": text}
        response = requests.post(settings.nlp_service_url, json=payload, timeout=60)
        response.raise_for_status()
        return response.json()

    def _extract_local_rule_based(self, text: str) -> dict[str, Any]:
        cause = "工资报酬争议" if "工资" in text else "劳动争议"
        entry_match = re.search(r"(入职|到岗)[:：]?\s*([0-9]{4}[./-][0-9]{1,2}[./-][0-9]{1,2})", text)
        leave_match = re.search(r"(离职|解除)[:：]?\s*([0-9]{4}[./-][0-9]{1,2}[./-][0-9]{1,2})", text)
        amount_match = re.search(r"([0-9]{3,8}(?:\.[0-9]{1,2})?)\s*元", text)
        law_refs = re.findall(r"《[^》]+》", text)

        return {
            "parties": {
                "employer_nature": "企业",
                "worker_position": "待识别",
            },
            "case_cause": {"type": cause},
            "facts": {
                "entry_date": entry_match.group(2) if entry_match else None,
                "leave_date": leave_match.group(2) if leave_match else None,
                "overtime": "加班" in text,
                "termination_reason": "待识别",
            },
            "claims": {
                "amount": float(amount_match.group(1)) if amount_match else None,
                "types": ["工资", "经济补偿"] if "补偿" in text else ["工资"],
            },
            "laws": list(set(law_refs)),
        }