190 lines
5.5 KiB
Python
190 lines
5.5 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
|
|
ONTOLOGY_FIELD_ALIASES: dict[str, tuple[str, ...]] = {
|
|
"expense_type": ("reimbursement_type", "scene_label", "expenseType"),
|
|
"time_range": (
|
|
"business_time",
|
|
"businessTime",
|
|
"occurred_date",
|
|
"occurredDate",
|
|
"application_business_time",
|
|
"applicationBusinessTime",
|
|
"application_time",
|
|
"applicationTime",
|
|
),
|
|
"location": (
|
|
"business_location",
|
|
"businessLocation",
|
|
"application_location",
|
|
"applicationLocation",
|
|
),
|
|
"reason": (
|
|
"reason_value",
|
|
"reasonValue",
|
|
"business_reason",
|
|
"businessReason",
|
|
"application_reason",
|
|
"applicationReason",
|
|
),
|
|
"amount": (
|
|
"application_amount",
|
|
"applicationAmount",
|
|
"application_amount_label",
|
|
"applicationAmountLabel",
|
|
),
|
|
"transport_mode": (
|
|
"transport_type",
|
|
"transportType",
|
|
"transportMode",
|
|
"application_transport_mode",
|
|
"applicationTransportMode",
|
|
),
|
|
"attachments": ("attachment_names", "attachmentNames"),
|
|
"customer_name": ("customerName",),
|
|
"merchant_name": ("merchantName",),
|
|
"cost_center": ("costCenter",),
|
|
"department_name": ("department", "departmentName", "deptName"),
|
|
"employee_grade": ("grade", "user_grade", "employeeGrade", "position_grade"),
|
|
"employee_name": ("name", "user_name", "applicant", "claimant_name", "reporter_name"),
|
|
"employee_no": ("employeeNo",),
|
|
"employee_position": ("position", "employeePosition"),
|
|
"manager_name": ("managerName", "direct_manager_name", "directManagerName"),
|
|
"finance_owner_name": ("financeOwnerName",),
|
|
"finance_approver_name": ("financeApproverName",),
|
|
}
|
|
|
|
CANONICAL_ONTOLOGY_FIELDS = frozenset(ONTOLOGY_FIELD_ALIASES) | frozenset(
|
|
{
|
|
"participants",
|
|
"department",
|
|
"budget_period",
|
|
"budget_subject",
|
|
"budget_amount",
|
|
"cost_center",
|
|
"warning_threshold",
|
|
"control_action",
|
|
"employee_location",
|
|
"employee_risk_profile",
|
|
"document_id",
|
|
"application_claim_id",
|
|
"application_claim_no",
|
|
"application_days",
|
|
"application_date",
|
|
"application_lodging_daily_cap",
|
|
"application_subsidy_daily_cap",
|
|
"application_transport_policy",
|
|
"application_policy_estimate",
|
|
"application_rule_name",
|
|
"application_rule_version",
|
|
"original_amount",
|
|
"reimbursable_amount",
|
|
"employee_absorbed_amount",
|
|
}
|
|
)
|
|
|
|
ONTOLOGY_CONTEXT_METADATA_FIELDS = frozenset(
|
|
{
|
|
"_claim_no_retry_count",
|
|
"actor",
|
|
"application_edit_claim_id",
|
|
"application_edit_mode",
|
|
"applicationEditClaimId",
|
|
"applicationEditMode",
|
|
"application_fields",
|
|
"application_preview",
|
|
"application_stage",
|
|
"attachment_count",
|
|
"attachment_names",
|
|
"business_time_context",
|
|
"budget_details",
|
|
"budget_header",
|
|
"client_now_iso",
|
|
"client_timezone_offset_minutes",
|
|
"conversation_history",
|
|
"conversation_id",
|
|
"conversation_intent",
|
|
"conversation_scenario",
|
|
"conversation_state",
|
|
"document_type",
|
|
"draft_claim_id",
|
|
"dry_run_email",
|
|
"email",
|
|
"entry_source",
|
|
"expense_scene_selection",
|
|
"force",
|
|
"is_admin",
|
|
"ocr_documents",
|
|
"ocr_summary",
|
|
"report_type",
|
|
"request_context",
|
|
"requested_by_name",
|
|
"requested_by_username",
|
|
"review_action",
|
|
"review_document_form_values",
|
|
"review_form_values",
|
|
"role_codes",
|
|
"role",
|
|
"send_email",
|
|
"session_type",
|
|
"simulate_orchestrator_exception",
|
|
"simulate_tool_failure",
|
|
"time_range_raw",
|
|
"user_id",
|
|
"user_input_text",
|
|
"username",
|
|
}
|
|
)
|
|
|
|
REGISTERED_ONTOLOGY_CONTEXT_FIELDS = (
|
|
CANONICAL_ONTOLOGY_FIELDS
|
|
| ONTOLOGY_CONTEXT_METADATA_FIELDS
|
|
| frozenset(alias for aliases in ONTOLOGY_FIELD_ALIASES.values() for alias in aliases)
|
|
)
|
|
|
|
|
|
def normalize_ontology_form_values(values: Any) -> dict[str, str]:
|
|
if not isinstance(values, dict):
|
|
return {}
|
|
|
|
normalized: dict[str, str] = {}
|
|
for key, value in values.items():
|
|
cleaned_key = str(key or "").strip()
|
|
if not cleaned_key:
|
|
continue
|
|
normalized[cleaned_key] = str(value or "").strip()
|
|
|
|
for canonical_key, aliases in ONTOLOGY_FIELD_ALIASES.items():
|
|
if normalized.get(canonical_key):
|
|
continue
|
|
for alias in aliases:
|
|
if normalized.get(alias):
|
|
normalized[canonical_key] = normalized[alias]
|
|
break
|
|
|
|
return normalized
|
|
|
|
|
|
def normalize_ontology_context_json(context_json: Any) -> dict[str, Any]:
|
|
if not isinstance(context_json, dict):
|
|
return {}
|
|
|
|
normalized = dict(context_json)
|
|
for canonical_key, aliases in ONTOLOGY_FIELD_ALIASES.items():
|
|
if normalized.get(canonical_key):
|
|
continue
|
|
for alias in aliases:
|
|
if normalized.get(alias):
|
|
normalized[canonical_key] = normalized[alias]
|
|
break
|
|
form_values = normalize_ontology_form_values(normalized.get("review_form_values"))
|
|
if form_values:
|
|
normalized["review_form_values"] = form_values
|
|
return normalized
|
|
|
|
|
|
def is_registered_ontology_context_field(field_name: str) -> bool:
|
|
return str(field_name or "").strip() in REGISTERED_ONTOLOGY_CONTEXT_FIELDS
|