labor-arbitration-system/backend/app/extractors/rule_extractor.py

"""
基于规则的劳动仲裁要素抽取（独立模块，供混合抽取器等复用）。
案由六分类与 app.extractor.classify_template_primary_cause 一致。
"""
from __future__ import annotations

import re
from typing import Any

# 支持包内导入与将 app 目录加入 PYTHONPATH 后的裸导入（与任务说明一致）
try:
    from app.extractor import parse_amount
except ImportError:
    from extractor import parse_amount  # type: ignore

from app.extractor import (
    classify_template_primary_cause,
    extract_applicant_name,
    extract_claims,
    extract_contract_type,
    extract_entry_date,
    extract_law_refs,
    extract_leave_date,
    extract_month_salary,
    extract_overtime_desc,
    extract_respondent_name,
    extract_termination_reason,
    merge_dispute_template_fields,
)
from app.extractor import _amount_near_keywords as amount_near_keywords  # noqa: SLF001
from app.extractor import _clean_text as clean_text  # noqa: SLF001
from app.extractor import _yes_no_from_text  # noqa: SLF001

_DEFAULT_CAUSE = "劳动关系纠纷类"


class RuleBasedExtractor:
    """规则抽取器：案由分类 + 通用/劳动合同/社保/法条 + 仲裁请求（规则版）。"""

    def extract_all(self, text: str) -> dict[str, Any]:
        t = clean_text(text or "")
        primary_cause_type = classify_template_primary_cause(t) or _DEFAULT_CAUSE

        applicant = extract_applicant_name(t)
        respondent = extract_respondent_name(t)
        entry_date = extract_entry_date(t)
        leave_date = extract_leave_date(t)
        month_salary_standard = extract_month_salary(t)
        overtime_fact = extract_overtime_desc(t)
        termination_reason = extract_termination_reason(t)

        labor_contract_signed = _yes_no_from_text(
            t,
            ["签订劳动合同", "订立书面劳动合同", "有书面合同", "已签订劳动合同"],
            ["未签订劳动合同", "未订立书面劳动合同", "无书面合同", "未签订书面劳动合同"],
        )
        contract_type = extract_contract_type(t)
        double_wage_related = "是" if any(x in t for x in ("二倍工资", "双倍工资", "未签劳动合同")) else None

        social_insurance_enrolled = _yes_no_from_text(
            t,
            ["已缴纳社保", "参加社会保险", "缴纳五险", "有社保", "已参保"],
            ["未缴纳社保", "未参加社会保险", "未缴社保", "无社保", "未参保"],
        )
        social_insurance_amount = amount_near_keywords(t, ("保险待遇", "社保待遇", "补缴", "社会保险费"))
        if social_insurance_amount is None:
            for frag in re.split(r"[\n。；]+", t):
                if any(k in frag for k in ("社保", "社会保险", "五险")):
                    social_insurance_amount = parse_amount(frag)
                    if social_insurance_amount is not None:
                        break

        law_refs = extract_law_refs(t)
        claims = extract_claims(t)

        base: dict[str, Any] = {
            "primary_cause_type": primary_cause_type,
            "applicant_name": applicant,
            "respondent_name": respondent,
            "entry_date": entry_date,
            "leave_date": leave_date,
            "month_salary_standard": month_salary_standard,
            "overtime_fact": overtime_fact,
            "termination_reason": termination_reason,
            "labor_contract_signed": labor_contract_signed,
            "contract_type": contract_type,
            "double_wage_related": double_wage_related,
            "social_insurance_enrolled": social_insurance_enrolled,
            "social_insurance_amount": social_insurance_amount,
            "law_refs": law_refs,
            "claims": claims,
        }
        # 与现有模板扁平字段对齐，便于后续层级/画像复用
        base.update(merge_dispute_template_fields(t, base))
        base["primary_cause_type"] = base.get("tmpl_primary_cause") or primary_cause_type
        base["tmpl_primary_cause"] = base["primary_cause_type"]
        return base