zkxw/backend/app/services/oa_resignation_kit_sync.py

"""离职套件：审批通过后回写发起人 last_work_day。"""
from __future__ import annotations

import json
import logging
import re
from datetime import date, datetime
from typing import Any, Dict, List, Optional

from sqlalchemy.orm import Session

from app.models.models import OaAppVersion, OaInstance, SysUser

logger = logging.getLogger(__name__)

_YMD = re.compile(r"^(\d{4})-(\d{2})-(\d{2})$")


def _loads(raw: Any, default: Any = None) -> Any:
    if raw is None or raw == "":
        return default
    if isinstance(raw, (dict, list)):
        return raw
    try:
        return json.loads(raw)
    except Exception:
        return default


def _extract_rules(form_schema: Any) -> List[Dict[str, Any]]:
    if form_schema is None:
        return []
    if isinstance(form_schema, list):
        return [r for r in form_schema if isinstance(r, dict)]
    if isinstance(form_schema, dict):
        rules = form_schema.get("rule") or form_schema.get("rules") or form_schema.get("fields")
        if isinstance(rules, list):
            return [r for r in rules if isinstance(r, dict)]
        if form_schema.get("type") or form_schema.get("field"):
            return [form_schema]
    return []


def _walk(rules: List[Dict[str, Any]], out: List[Dict[str, Any]]) -> None:
    for rule in rules:
        out.append(rule)
        children = rule.get("children") or (rule.get("props") or {}).get("rule") or []
        if isinstance(children, list):
            _walk([c for c in children if isinstance(c, dict)], out)


def _last_work_day_field(kit: Dict[str, Any]) -> Optional[str]:
    props = kit.get("props") if isinstance(kit.get("props"), dict) else {}
    if props.get("lastWorkDayField"):
        return str(props["lastWorkDayField"])
    children = kit.get("children") or props.get("rule") or []
    if isinstance(children, list):
        for c in children:
            if not isinstance(c, dict) or not c.get("field"):
                continue
            cprops = c.get("props") if isinstance(c.get("props"), dict) else {}
            if str(cprops.get("_fc_kit_field") or "") == "lastWorkDay":
                return str(c["field"])
    return None


def _parse_ymd(value: Any) -> Optional[date]:
    if value is None or value == "":
        return None
    if isinstance(value, datetime):
        return value.date()
    if isinstance(value, date):
        return value
    s = str(value).strip()[:10]
    if not _YMD.match(s):
        return None
    try:
        return datetime.strptime(s, "%Y-%m-%d").date()
    except ValueError:
        return None


def extract_last_work_day(schema: Any, form_data: Dict[str, Any]) -> Optional[date]:
    flat: List[Dict[str, Any]] = []
    _walk(_extract_rules(schema), flat)
    for rule in flat:
        if str(rule.get("type") or "") != "resignationKit":
            continue
        field = _last_work_day_field(rule)
        if not field:
            continue
        parsed = _parse_ymd(form_data.get(field))
        if parsed is not None:
            return parsed
    return None


def apply_resignation_kit_user_sync(db: Session, inst: OaInstance) -> None:
    try:
        if not inst or not getattr(inst, "id", None):
            return
        form_data = _loads(getattr(inst, "form_data_json", None), {}) or {}
        if not isinstance(form_data, dict):
            form_data = {}
        ver = (
            db.query(OaAppVersion)
            .filter(OaAppVersion.id == inst.app_version_id, OaAppVersion.del_flag == 0)
            .first()
        )
        schema = _loads(getattr(ver, "form_schema_json", None) if ver else None, {}) or {}
        last_day = extract_last_work_day(schema, form_data)
        if last_day is None:
            return
        user = (
            db.query(SysUser)
            .filter(
                SysUser.id == inst.initiator_user_id,
                SysUser.tenant_no == inst.tenant_no,
                SysUser.del_flag == 0,
            )
            .first()
        )
        if not user:
            logger.warning(
                "resignation kit sync: initiator not found id=%s",
                inst.initiator_user_id,
            )
            return
        user.last_work_day = last_day
        user.update_time = datetime.now()
        db.add(user)
    except Exception:
        logger.exception(
            "resignation kit user sync failed instance=%s", getattr(inst, "id", None)
        )