getDomain/domainCheck/app/utils/detection_results.py

import json


def build_detection_result(
    *,
    status=None,
    state=None,
    message="",
    error=None,
    **payload,
):
    result = dict(payload)
    result["status"] = bool(status) if status is not None else None
    result["message"] = message or ""

    if error:
        result["state"] = "error"
        result["error"] = str(error)
        if not result["message"]:
            result["message"] = str(error)
    else:
        result["state"] = state or _default_state_for_status(status)
        result.pop("error", None)

    return result


def _default_state_for_status(status):
    if status is True:
        return "positive"
    if status is False:
        return "negative"
    return "ok"


def load_detection_result(value):
    if value is None:
        return {}
    if isinstance(value, dict):
        return value
    if isinstance(value, str):
        try:
            decoded = json.loads(value)
        except Exception:
            return {}
        return decoded if isinstance(decoded, dict) else {}
    return {}


def resolve_detection_status(value, *legacy_keys):
    data = load_detection_result(value)
    if not data:
        return False

    status = data.get("status")
    if status is not None:
        return bool(status)

    for key in legacy_keys:
        if data.get(key) is not None:
            return bool(data.get(key))
    return False


def build_manual_detection_result(status, *, legacy_key=None, message="人工更新"):
    payload = {}
    if legacy_key:
        payload[legacy_key] = bool(status)
    return build_detection_result(
        status=bool(status),
        state="manual",
        message=message,
        **payload,
    )


def normalize_detector_result(name, result):
    result = load_detection_result(result)
    if result.get("error"):
        return build_detection_result(error=result.get("error"), **_without_meta(result))

    if name == "baidu_history":
        has_history = bool(result.get("has_history"))
        has_gray = bool(result.get("has_gray"))
        return build_detection_result(
            status=has_history,
            state="risk" if has_gray else None,
            has_history=has_history,
            has_gray=has_gray,
        )

    if name in {"baidu_site", "qihu360_site", "google_site"}:
        has_index = bool(result.get("has_收录"))
        normalized = build_detection_result(
            status=has_index,
            has_收录=has_index,
            subdomains=list(result.get("subdomains", []) or []),
        )
        return normalized

    if name == "chinaz_info":
        return build_detection_result(
            status=None,
            title=result.get("title", ""),
            category=result.get("category", ""),
            has_sensitive=bool(result.get("has_sensitive")),
        )

    if name == "aizhan_info":
        return build_detection_result(
            status=None,
            title=result.get("title", ""),
            risk=result.get("risk", ""),
            has_sensitive=bool(result.get("has_sensitive")),
        )

    if name == "juziseo_info":
        history = normalize_detector_result("juziseo_history", result.get("history"))
        backlink = normalize_detector_result("juziseo_backlink", result.get("backlink"))
        nested_error = history.get("error") or backlink.get("error")
        return build_detection_result(
            status=None,
            error=nested_error,
            history=history,
            backlink=backlink,
        )

    if name == "juziseo_history":
        return build_detection_result(
            status=None,
            state="risk" if result.get("has_sensitive") or result.get("has_subdomains") else None,
            has_sensitive=bool(result.get("has_sensitive")),
            has_baidu_history=bool(result.get("has_baidu_history")),
            has_subdomains=bool(result.get("has_subdomains")),
            is_simplified=bool(result.get("is_simplified", True)),
        )

    if name == "juziseo_backlink":
        return build_detection_result(
            status=None,
            state="risk" if result.get("has_sensitive") or result.get("has_subdomains") else None,
            has_sensitive=bool(result.get("has_sensitive")),
            has_subdomains=bool(result.get("has_subdomains")),
        )

    if name == "jucha_info":
        whois = normalize_detector_result("jucha_whois", result.get("whois"))
        beian = normalize_detector_result("jucha_beian", result.get("beian"))
        intercept = normalize_detector_result("jucha_intercept", result.get("intercept"))
        nested_error = whois.get("error") or beian.get("error") or intercept.get("error")
        return build_detection_result(
            status=None,
            error=nested_error,
            whois=whois,
            beian=beian,
            intercept=intercept,
        )

    if name == "jucha_whois":
        whois_status = result.get("status", "")
        hold = whois_status in {"clientHold", "serverHold"}
        return build_detection_result(
            status=None,
            state="risk" if hold else None,
            whois_status=whois_status,
        )

    if name == "jucha_beian":
        has_beian = bool(result.get("has_beian"))
        return build_detection_result(
            status=has_beian,
            has_beian=has_beian,
            beian_year=result.get("beian_year", ""),
            is_enterprise=bool(result.get("is_enterprise")),
            beian_match=bool(result.get("beian_match")),
        )

    if name == "jucha_intercept":
        normal = bool(result.get("normal"))
        return build_detection_result(
            status=normal,
            state="risk" if not normal else None,
            normal=normal,
        )

    return build_detection_result(status=None, **result)


def _without_meta(result):
    return {
        key: value
        for key, value in result.items()
        if key not in {"status", "state", "message", "error"}
    }