""" 按 case_elements_hierarchy.json 的层级结构，将规则抽取结果（merge_dispute_template_fields 等扁平字段） 填入对应叶子节点；案由分类决定仅展示与填充该案由子树。 """ from __future__ import annotations import copy import json from pathlib import Path from typing import Any _HIERARCHY_PATH = Path(__file__).resolve().parent.parent / "data" / "case_elements_hierarchy.json" # 叶子路径 "案由>一层>二层>…" -> merge_dispute_template_fields / extract 产出的英文 hint 键 LEAF_PATH_TO_HINT_KEY: dict[str, str] = { # —— 通用要素（各案由顶层，值来自同一组扁平字段）—— "劳动关系纠纷类>申请人信息": "gen_applicant_info", "劳动关系纠纷类>被申请人信息": "gen_respondent_info", "劳动关系纠纷类>事实与理由": "gen_facts_and_reasons", "工伤保险待遇纠纷>申请人信息": "gen_applicant_info", "工伤保险待遇纠纷>被申请人信息": "gen_respondent_info", "工伤保险待遇纠纷>事实与理由": "gen_facts_and_reasons", "经济补偿金纠纷>申请人信息": "gen_applicant_info", "经济补偿金纠纷>被申请人信息": "gen_respondent_info", "经济补偿金纠纷>事实与理由": "gen_facts_and_reasons", "赔偿金纠纷>申请人信息": "gen_applicant_info", "赔偿金纠纷>被申请人信息": "gen_respondent_info", "赔偿金纠纷>事实与理由": "gen_facts_and_reasons", "生育保险待遇纠纷>申请人信息": "gen_applicant_info", "生育保险待遇纠纷>被申请人信息": "gen_respondent_info", "生育保险待遇纠纷>事实与理由": "gen_facts_and_reasons", # —— 劳动关系纠纷类 —— "劳动关系纠纷类>劳动报酬>劳动报酬发放周期": "lr1_pay_cycle", "劳动关系纠纷类>劳动报酬>劳动报酬金额": "lr1_pay_amount", "劳动关系纠纷类>劳动报酬>劳动报酬发放形式": "lr1_pay_form", "劳动关系纠纷类>社会保险>是否参加社会保险": "lr1_si_joined", "劳动关系纠纷类>社会保险>保险待遇金额": "lr1_si_benefit_amount", "劳动关系纠纷类>劳动关系>是否签订劳动合同>签订无固定限期劳动合同": "lr1_open_ended_contract", "劳动关系纠纷类>劳动关系>是否签订劳动合同>未签订劳动合同的二倍工资": "lr1_double_wage_no_contract", "劳动关系纠纷类>劳动关系>劳动关系存在时间": "lr1_relation_duration", # —— 工伤保险待遇纠纷 —— "工伤保险待遇纠纷>劳动关系>是否签订劳动合同": "wi_lr_contract_signed", "工伤保险待遇纠纷>劳动关系>劳动关系存在时间": "wi_lr_relation_duration", "工伤保险待遇纠纷>保险待遇>保险待遇发放时间>一次性伤残补助费": "wi_benefit_disability", "工伤保险待遇纠纷>保险待遇>保险待遇发放时间>辅助器具费": "wi_benefit_prosthetic", "工伤保险待遇纠纷>保险待遇>保险待遇发放时间>一次性医疗补助金": "wi_benefit_medical_allowance", "工伤保险待遇纠纷>保险待遇>保险待遇发放时间>交通食宿费": "wi_benefit_travel", "工伤保险待遇纠纷>保险待遇>保险待遇发放时间>工伤医疗/康复费用": "wi_benefit_rehab", "工伤保险待遇纠纷>保险待遇>保险待遇发放时间>住院治疗期间护理费": "wi_benefit_nursing", "工伤保险待遇纠纷>保险待遇>保险待遇发放时间>住院伙食补助费": "wi_benefit_meal", "工伤保险待遇纠纷>保险待遇>保险待遇金额": "wi_benefit_amount_total", "工伤保险待遇纠纷>保险待遇>保险待遇发放形式": "wi_benefit_pay_form", "工伤保险待遇纠纷>保险待遇>是否认可经济补偿": "wi_recognize_ec", "工伤保险待遇纠纷>社会保险>保险待遇金额": "wi_si_benefit_amt", "工伤保险待遇纠纷>社会保险>是否参加社会保险": "wi_si_joined", # —— 追索劳动报酬 —— "追索劳动报酬>申请人信息": "gen_applicant_info", "追索劳动报酬>被申请人信息": "gen_respondent_info", "追索劳动报酬>事实与理由": "gen_facts_and_reasons", "追索劳动报酬>劳动报酬>劳动报酬发放周期": "sr_pay_cycle", "追索劳动报酬>劳动报酬>克扣劳动报酬>主张金额": "sr_claim_deducted_pay", "追索劳动报酬>劳动报酬>克扣劳动报酬>生活费": "sr_claim_living_allowance", "追索劳动报酬>劳动报酬>加班工资": "sr_claim_overtime_pay", "追索劳动报酬>劳动报酬>实际支付工资标准": "sr_actual_pay_standard", "追索劳动报酬>劳动报酬>高温津贴金额": "sr_high_temp_allowance", "追索劳动报酬>劳动报酬>约定工资标准": "sr_agreed_pay_standard", "追索劳动报酬>劳动报酬>带薪年休假工资": "sr_annual_leave_pay", "追索劳动报酬>劳动报酬>未支付期间": "sr_unpaid_period", "追索劳动报酬>劳动报酬>加班工资金额": "sr_overtime_amount", # —— 经济补偿金纠纷 —— "经济补偿金纠纷>经济补偿金>离职前12个月平均工资": "ec_avg_salary_12m", "经济补偿金纠纷>经济补偿金>未签订劳动合同的二倍工资": "ec_double_wage_part", "经济补偿金纠纷>经济补偿金>主张金额>违法解除劳动合同的赔偿金": "ec_illegal_term_part", "经济补偿金纠纷>经济补偿金>主张金额>违法约定试用期的赔偿金": "ec_illegal_probation_part", "经济补偿金纠纷>经济补偿金>代通知金": "ec_notice_pay", "经济补偿金纠纷>经济补偿金>加付赔偿金": "ec_additional_damages", "经济补偿金纠纷>经济补偿金>劳动合同存在时间": "ec_contract_duration", "经济补偿金纠纷>劳动合同>离职原因": "ec_leave_reason", "经济补偿金纠纷>劳动合同>离职时间": "ec_leave_date", # —— 赔偿金纠纷 —— "赔偿金纠纷>赔偿金>主张金额": "dm_claim_amount", "赔偿金纠纷>赔偿金>违法解除劳动合同赔偿金": "dm_illegal_dismissal_damages", "赔偿金纠纷>劳动合同>劳动合同是否存在": "dm_contract_exists", "赔偿金纠纷>劳动合同>解除劳动合同原因": "dm_terminate_reason", "赔偿金纠纷>劳动合同>劳动合同是否继续履行": "dm_contract_continue", # —— 生育保险待遇纠纷 —— "生育保险待遇纠纷>金额>主张金额>产假工资/生育津贴": "mi_maternity_allowance_salary", "生育保险待遇纠纷>金额>主张金额>生育医疗费用": "mi_maternity_medical", "生育保险待遇纠纷>金额>加付赔偿金": "mi_additional_damages", "生育保险待遇纠纷>金额>交通食宿费": "mi_travel_accommodation", "生育保险待遇纠纷>劳动合同>劳动合同是否继续履行": "mi_contract_continue", "生育保险待遇纠纷>劳动合同>劳动合同解除原因": "mi_terminate_reason", } DEFAULT_CAUSE_TYPE = "劳动关系纠纷类" def load_hierarchy_templates() -> dict[str, Any]: if not _HIERARCHY_PATH.is_file(): return {} try: return json.loads(_HIERARCHY_PATH.read_text(encoding="utf-8")) except Exception: return {} def _path_key(path: tuple[str, ...]) -> str: return ">".join(path) def _leaf_value(path: tuple[str, ...], hints: dict[str, Any]) -> Any: key = _path_key(path) hk = LEAF_PATH_TO_HINT_KEY.get(key) if hk and hk in hints: v = hints[hk] if v is not None and v != "": return v return None def fill_hierarchy_node(node: Any, path: tuple[str, ...], hints: dict[str, Any]) -> Any: if node is None: return _leaf_value(path, hints) if isinstance(node, dict): return {k: fill_hierarchy_node(v, path + (k,), hints) for k, v in node.items()} return node def build_elements_hierarchy_for_cause(cause_type: str, hints: dict[str, Any]) -> dict[str, Any] | None: templates = load_hierarchy_templates() if cause_type not in templates: return None subtree = copy.deepcopy(templates[cause_type]) return fill_hierarchy_node(subtree, (cause_type,), hints) def count_hierarchy_leaves(obj: Any) -> int: if obj is None: return 1 if isinstance(obj, dict): if not obj: return 0 return sum(count_hierarchy_leaves(v) for v in obj.values()) return 0 def flatten_hierarchy_for_preview(obj: Any, path: tuple[str, ...] = ()) -> list[dict[str, Any]]: """扁平行：供 case_elements_table 简要预览。""" rows: list[dict[str, Any]] = [] if obj is None: rows.append({"path": list(path), "value": None}) elif isinstance(obj, dict): for k, v in obj.items(): rows.extend(flatten_hierarchy_for_preview(v, path + (k,))) else: rows.append({"path": list(path), "value": obj}) return rows