Refactor/code optimization

backend/nexus/nexus_repo.py

@@ -1,13 +1,14 @@
+import asyncio
 import os
 import time
 from pathlib import PurePosixPath
 from urllib.parse import quote
 
+import httpx
 import requests
 from fastapi import APIRouter, Depends, HTTPException, Request
 from fastapi.responses import StreamingResponse
 from requests.auth import HTTPBasicAuth
-from starlette.concurrency import run_in_threadpool
 
 from backend.auth.security import require_permission
 
@@ -21,6 +22,7 @@
 NEXUS_DASHBOARD_CACHE_TTL_SECONDS = int(os.getenv("NEXUS_DASHBOARD_CACHE_TTL_SECONDS", "30"))
 
 nexus_auth = HTTPBasicAuth(NEXUS_USERNAME, NEXUS_PASSWORD)
+httpx_nexus_auth = httpx.BasicAuth(NEXUS_USERNAME or "", NEXUS_PASSWORD or "")
 _dashboard_cache = {"expires_at": 0.0, "payload": None}
 
 
@@ -65,6 +67,35 @@ def fetch_nexus_assets():
     return all_assets
 
 
+async def fetch_nexus_assets_async(client):
+    nexus_url = f"{NEXUS_BASE_URL}/service/rest/v1/assets"
+    all_assets = []
+    continuation_token = None
+
+    while True:
+        params = {"repository": NEXUS_REPOSITORY}
+        if continuation_token:
+            params["continuationToken"] = continuation_token
+
+        response = await client.get(nexus_url, params=params)
+
+        if response.status_code != 200:
+            print(f"Nexus API Error: {response.status_code} - {response.text}")
+            break
+
+        data = response.json()
+        for item in data.get("items", []):
+            if item.get("path"):
+                item["path"] = item["path"].lstrip("/")
+            all_assets.append(item)
+        continuation_token = data.get("continuationToken")
+
+        if not continuation_token:
+            break
+
+    return all_assets
+
+
 def fetch_nexus_assets_by_name(names):
     nexus_url = f"{NEXUS_BASE_URL}/service/rest/v1/search/assets"
     matches = []
@@ -98,6 +129,39 @@ def fetch_nexus_assets_by_name(names):
     return matches
 
 
+async def fetch_nexus_assets_by_name_async(client, names):
+    nexus_url = f"{NEXUS_BASE_URL}/service/rest/v1/search/assets"
+
+    async def fetch_one(name):
+        if not name:
+            return []
+
+        matches = []
+        continuation_token = None
+        while True:
+            params = {"repository": NEXUS_REPOSITORY, "name": name}
+            if continuation_token:
+                params["continuationToken"] = continuation_token
+
+            response = await client.get(nexus_url, params=params)
+            if response.status_code != 200:
+                print(f"Nexus Search API Error: {response.status_code} - {response.text}")
+                break
+
+            data = response.json()
+            for item in data.get("items", []):
+                if item.get("path"):
+                    item["path"] = item["path"].lstrip("/")
+                matches.append(item)
+            continuation_token = data.get("continuationToken")
+            if not continuation_token:
+                break
+        return matches
+
+    results = await asyncio.gather(*(fetch_one(name) for name in names))
+    return [item for group in results for item in group]
+
+
 def fetch_nexus_blobstores():
     nexus_url = f"{NEXUS_BASE_URL}/service/rest/v1/blobstores"
     response = requests.get(
@@ -113,6 +177,17 @@ def fetch_nexus_blobstores():
     return response.json()
 
 
+async def fetch_nexus_blobstores_async(client):
+    nexus_url = f"{NEXUS_BASE_URL}/service/rest/v1/blobstores"
+    response = await client.get(nexus_url)
+
+    if response.status_code != 200:
+        print(f"Nexus Blob Store API Error: {response.status_code} - {response.text}")
+        return []
+
+    return response.json()
+
+
 def get_safe_item_name(item):
     path = item.get("path") or ""
     parts = path.split("/")
@@ -184,10 +259,30 @@ def fetch_dashboard_payload():
     return payload
 
 
+async def fetch_dashboard_payload_async(client):
+    now = time.monotonic()
+    cached_payload = _dashboard_cache["payload"]
+    if cached_payload is not None and _dashboard_cache["expires_at"] > now:
+        return cached_payload
+
+    assets, blobstores = await asyncio.gather(
+        fetch_nexus_assets_async(client),
+        fetch_nexus_blobstores_async(client),
+    )
+    payload = {
+        "items": assets,
+        "summary": build_dashboard_summary(assets, blobstores),
+    }
+    _dashboard_cache["payload"] = payload
+    _dashboard_cache["expires_at"] = now + NEXUS_DASHBOARD_CACHE_TTL_SECONDS
+    return payload
+
+
 @router.post("/api/nexus/list")
 async def nexus_list(_user: dict = Depends(require_permission("install_extension"))):
     try:
-        all_assets = await run_in_threadpool(fetch_nexus_assets)
+        async with httpx.AsyncClient(auth=httpx_nexus_auth, timeout=10) as client:
+            all_assets = await fetch_nexus_assets_async(client)
         print(f"총 {len(all_assets)}개의 자산을 Nexus에서 성공적으로 불러왔습니다.")
         return all_assets
     except Exception as e:
@@ -212,7 +307,8 @@ async def nexus_exists(
         candidate_names.append(f"{ext_id}-{version}.vsix")
 
     try:
-        matches = await run_in_threadpool(fetch_nexus_assets_by_name, candidate_names)
+        async with httpx.AsyncClient(auth=httpx_nexus_auth, timeout=10) as client:
+            matches = await fetch_nexus_assets_by_name_async(client, candidate_names)
     except Exception as e:
         print(f"Nexus exists lookup error: {e}")
         matches = []
@@ -277,7 +373,8 @@ def iterfile():
 @router.get("/api/nexus/dashboard")
 async def nexus_dashboard():
     try:
-        return await run_in_threadpool(fetch_dashboard_payload)
+        async with httpx.AsyncClient(auth=httpx_nexus_auth, timeout=10) as client:
+            return await fetch_dashboard_payload_async(client)
     except Exception as e:
         print(f"Nexus dashboard summary error: {e}")
         return {

backend/recevie_result.py

@@ -30,6 +30,7 @@ def _run_ai_judgment(web_payload: dict, target_dir: Path) -> None:
 
 # 데이터를 저장할 기본 루트 디렉토리
 BASE_SAVE_DIR = Path("analysis_result")
+POLICY_PATH = Path(__file__).resolve().parent / "admin" / "policy_settings.json"
 
 
 def safe_path_part(value: Any, default: str) -> str:
@@ -62,6 +63,146 @@ def get_nested(payload: Dict[str, Any], *keys: str, default: Any = None) -> Any:
     return current if current is not None else default
 
 
+def _normalize_decision(value: Any) -> str:
+    lowered = str(value or "").strip().lower()
+    if lowered in {"approve", "safe"}:
+        return "safe"
+    if lowered in {"reject", "rejected", "critical"}:
+        return "reject"
+    if lowered in {"review", "manual_review", "hold"}:
+        return "review"
+    return lowered or "undecided"
+
+
+def _read_policy() -> Dict[str, Any]:
+    default = {
+        "critical_auto_reject_enabled": True,
+        "low_auto_approve_enabled": False,
+        "fallback_decision": "review",
+    }
+    try:
+        if not POLICY_PATH.exists():
+            return default
+        with POLICY_PATH.open("r", encoding="utf-8") as file:
+            data = json.load(file)
+        if not isinstance(data, dict):
+            return default
+        return {**default, **data, "fallback_decision": "review"}
+    except Exception as policy_e:
+        print(f"[receive_result] policy read failed: {policy_e}")
+        return default
+
+
+def _apply_auto_policy(raw_decision: Any, recommended_decision: Any, risk_level: Any) -> tuple[str, Dict[str, Any]]:
+    policy = _read_policy()
+    raw = _normalize_decision(raw_decision)
+    recommended = _normalize_decision(recommended_decision)
+    risk = str(risk_level or "").strip().upper()
+
+    if recommended == "reject" or risk == "CRITICAL":
+        decision = "reject" if policy.get("critical_auto_reject_enabled") else "review"
+        reason = "critical_auto_reject" if decision == "reject" else "manual_review"
+    elif risk == "LOW" or recommended == "safe" or raw == "safe":
+        decision = "safe" if policy.get("low_auto_approve_enabled") else "review"
+        reason = "low_auto_approve" if decision == "safe" else "manual_review"
+    elif raw == "reject":
+        decision = raw
+        reason = "explicit_decision"
+    else:
+        decision = "review"
+        reason = "recommended_review"
+
+    return decision, {
+        "risk_level": risk or "UNKNOWN",
+        "recommended_decision": recommended,
+        "incoming_decision": raw,
+        "decision": decision,
+        "reason": reason,
+        "policy": {
+            "critical_auto_reject_enabled": bool(policy.get("critical_auto_reject_enabled")),
+            "low_auto_approve_enabled": bool(policy.get("low_auto_approve_enabled")),
+            "fallback_decision": "review",
+        },
+    }
+
+
+def _nexus_status_for_decision(decision: Any) -> str:
+    normalized = _normalize_decision(decision)
+    if normalized == "safe":
+        return "safe"
+    return "review"
+
+
+def _reconcile_nexus_location(
+    *,
+    decision: str,
+    auto_policy: Dict[str, Any],
+    browser: str,
+    ext_names: list[str],
+    version: str,
+    ext_id: str,
+) -> Dict[str, Any]:
+    target_status = _nexus_status_for_decision(decision)
+    candidate_statuses = [
+        _nexus_status_for_decision(auto_policy.get("incoming_decision")),
+        _nexus_status_for_decision(auto_policy.get("recommended_decision")),
+    ]
+    candidate_statuses = list(dict.fromkeys(candidate_statuses))
+
+    try:
+        from backend.admin.decision.nexus_file import (
+            build_nexus_path,
+            fetch_nexus_asset_paths,
+            move_nexus_file,
+        )
+
+        normalized_names = []
+        for name in ext_names:
+            text = str(name or "").strip()
+            if text and text not in normalized_names:
+                normalized_names.append(text)
+        if not normalized_names:
+            normalized_names.append("unknown_extension")
+
+        target_path = build_nexus_path(target_status, browser, normalized_names[0], version, ext_id)
+        asset_paths = fetch_nexus_asset_paths()
+        asset_lookup = {path.lower(): path for path in asset_paths}
+
+        if target_path.lower() in asset_lookup:
+            return {
+                "status": "already_target",
+                "target_path": asset_lookup[target_path.lower()],
+                "candidate_statuses": candidate_statuses,
+            }
+
+        for status in candidate_statuses:
+            for ext_name in normalized_names:
+                source_path = build_nexus_path(status, browser, ext_name, version, ext_id)
+                if source_path.lower() not in asset_lookup:
+                    continue
+                resolved_source = asset_lookup[source_path.lower()]
+                move_nexus_file(resolved_source, target_path)
+                return {
+                    "status": "moved",
+                    "source_path": resolved_source,
+                    "target_path": target_path,
+                    "candidate_statuses": candidate_statuses,
+                }
+
+        return {
+            "status": "source_not_found",
+            "target_path": target_path,
+            "candidate_statuses": candidate_statuses,
+        }
+    except Exception as nexus_e:
+        print(f"[receive_result] nexus reconcile failed: {nexus_e}")
+        return {
+            "status": "error",
+            "message": str(nexus_e),
+            "candidate_statuses": candidate_statuses,
+        }
+
+
 @router.post("/api/receive")
 async def receive_and_save_analysis(
     background_tasks: BackgroundTasks,
@@ -80,12 +221,17 @@ async def receive_and_save_analysis(
 
         # 1. 경로 구성을 위한 정보 추출
         # 기존 legacy top-level 필드 우선, 없으면 새 web_payload 구조에서 추출
-        decision = (
+        raw_decision = (
             payload.get("decision")
             or payload.get("judge")
             or overall.get("recommended_decision")
             or "undecided"
         )
+        decision, auto_policy = _apply_auto_policy(
+            raw_decision=raw_decision,
+            recommended_decision=overall.get("recommended_decision"),
+            risk_level=overall.get("risk_level"),
+        )
 
         browser = (
             payload.get("browser")
@@ -113,11 +259,20 @@ async def receive_and_save_analysis(
             or "unknown_id"
         )
 
+        raw_ext_name = str(ext_name or "").strip()
         decision = safe_path_part(decision, "undecided")
         browser = safe_path_part(browser, "unknown_browser")
         ext_name = safe_path_part(ext_name, "unknown_extension")
         version = safe_path_part(version, "unknown_version")
         ext_id = safe_path_part(ext_id, "unknown_id")
+        nexus_reconcile = _reconcile_nexus_location(
+            decision=decision,
+            auto_policy=auto_policy,
+            browser=browser,
+            ext_names=[ext_name, raw_ext_name],
+            version=version,
+            ext_id=ext_id,
+        )
 
         # 2. 저장 경로 생성
         # analysis_result/{decision}/{browser}/{extName}/{version}/{extID}
@@ -170,6 +325,8 @@ async def receive_and_save_analysis(
                 "recommended_decision": overall.get("recommended_decision", decision),
                 "decision_reason": overall.get("decision_reason", ""),
             },
+            "auto_policy": auto_policy,
+            "nexus_reconcile": nexus_reconcile,
             "metadata": {
                 "extID": ext_id,
                 "extName": ext_name,