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"} }