zkxw/backend/app/services/oa_confirmation_kit_sync.py

"""转正套件：审批通过后回写发起人 sys_user。"""
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__)

WRITEBACK_KEYS = {
    "hireDate": "hire_date",
    "probationMonths": "probation_months",
    "name": "nick_name",
    "email": "email",
    "dept": "dept_id",
    "phone": "phone_no",
}

_PHONE_RE = re.compile(r"^1[3-9]\d{9}$")


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 _kit_field_map(kit: Dict[str, Any]) -> Dict[str, str]:
    """semantic -> form field id"""
    props = kit.get("props") if isinstance(kit.get("props"), dict) else {}
    mapping: Dict[str, str] = {}
    prop_keys = {
        "hireDateField": "hireDate",
        "probationMonthsField": "probationMonths",
        "confirmationDateField": "confirmationDate",
        "postField": "post",
        "jobLevelField": "jobLevel",
        "nameField": "name",
        "emailField": "email",
        "deptField": "dept",
        "phoneField": "phone",
    }
    for pk, sem in prop_keys.items():
        fid = props.get(pk)
        if fid:
            mapping[sem] = str(fid)

    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 {}
            sem = str(cprops.get("_fc_kit_field") or "").strip()
            if sem and sem not in mapping:
                mapping[sem] = str(c["field"])
    return mapping


def _parse_hire_date(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]
    try:
        return datetime.strptime(s, "%Y-%m-%d").date()
    except ValueError:
        return None


def _dept_id(value: Any) -> Optional[str]:
    if value is None or value == "":
        return None
    if isinstance(value, dict):
        sid = str(value.get("id") or "").strip()
        return sid or None
    if isinstance(value, list) and value:
        return _dept_id(value[0])
    sid = str(value).strip()
    return sid or None


def _is_empty(value: Any) -> bool:
    if value is None or value == "":
        return True
    if isinstance(value, list) and not value:
        return True
    if isinstance(value, dict) and not str(value.get("id") or "").strip():
        return True
    return False


def extract_writeback_values(schema: Any, form_data: Dict[str, Any]) -> Dict[str, Any]:
    """从 schema+form_data 提取可回写列值（列名 -> 值）。空字段不进入结果。"""
    flat: List[Dict[str, Any]] = []
    _walk(_extract_rules(schema), flat)
    out: Dict[str, Any] = {}
    for rule in flat:
        if str(rule.get("type") or "") != "confirmationKit":
            continue
        fmap = _kit_field_map(rule)
        for sem, col in WRITEBACK_KEYS.items():
            field = fmap.get(sem)
            if not field:
                continue
            raw = form_data.get(field)
            if _is_empty(raw):
                continue
            if col == "hire_date":
                parsed = _parse_hire_date(raw)
                if parsed is None:
                    logger.warning("confirmation kit skip invalid hire_date=%r", raw)
                    continue
                out[col] = parsed
            elif col == "probation_months":
                try:
                    months = int(raw)
                except (TypeError, ValueError):
                    logger.warning("confirmation kit skip invalid probation_months=%r", raw)
                    continue
                if months < 0:
                    continue
                out[col] = months
            elif col == "dept_id":
                did = _dept_id(raw)
                if not did:
                    continue
                out[col] = did
            elif col == "phone_no":
                phone = str(raw).strip()
                if not _PHONE_RE.match(phone):
                    logger.warning("confirmation kit skip invalid phone=%r", phone)
                    continue
                out[col] = phone
            elif col == "email":
                email = str(raw).strip()
                if "@" not in email:
                    logger.warning("confirmation kit skip invalid email=%r", email)
                    continue
                out[col] = email
            elif col == "nick_name":
                name = str(raw).strip()
                if not name:
                    continue
                out[col] = name
    return out


def apply_confirmation_kit_user_sync(db: Session, inst: OaInstance) -> None:
    """扫描 confirmationKit，有值则更新发起人 sys_user；异常只打日志。"""
    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 {}
        values = extract_writeback_values(schema, form_data)
        if not values:
            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(
                "confirmation kit sync: initiator not found id=%s tenant=%s",
                inst.initiator_user_id,
                inst.tenant_no,
            )
            return

        for col, val in values.items():
            setattr(user, col, val)
        user.update_time = datetime.now()
        db.add(user)
    except Exception:
        logger.exception(
            "confirmation kit user sync failed instance=%s", getattr(inst, "id", None)
        )