feat(aurora-smart): add dual-stack orchestration with policy, audit, and UI toggle
This commit is contained in:
Apple
@@ -253,6 +253,23 @@ _aurora_live_samples: Dict[str, collections.deque] = {}
|
||||
_aurora_live_last: Dict[str, Dict[str, Any]] = {}
|
||||
_aurora_live_last_loaded = False
|
||||
_aurora_live_last_path = (AURORA_DATA_DIR.parent / "sofiia-console-cache" / "aurora_live_last.json")
|
||||
_aurora_smart_runs: Dict[str, Dict[str, Any]] = {}
|
||||
_aurora_smart_runs_loaded = False
|
||||
_aurora_smart_runs_path = (AURORA_DATA_DIR.parent / "sofiia-console-cache" / "aurora_smart_runs.json")
|
||||
_aurora_smart_policy: Dict[str, Any] = {
|
||||
"updated_at": None,
|
||||
"strategies": {
|
||||
"local_only": {"count": 0, "avg_score": 0.0, "wins": 0, "losses": 0},
|
||||
"local_then_kling": {"count": 0, "avg_score": 0.0, "wins": 0, "losses": 0},
|
||||
},
|
||||
}
|
||||
_aurora_smart_policy_loaded = False
|
||||
_aurora_smart_policy_path = (AURORA_DATA_DIR.parent / "sofiia-console-cache" / "aurora_smart_policy.json")
|
||||
_AURORA_SMART_MAX_RUNS = max(20, int(os.getenv("AURORA_SMART_MAX_RUNS", "200")))
|
||||
_AURORA_SMART_LOCAL_POLL_SEC = max(2.0, float(os.getenv("AURORA_SMART_LOCAL_POLL_SEC", "3.0")))
|
||||
_AURORA_SMART_KLING_POLL_SEC = max(3.0, float(os.getenv("AURORA_SMART_KLING_POLL_SEC", "6.0")))
|
||||
_AURORA_SMART_LOCAL_MAX_SEC = max(60.0, float(os.getenv("AURORA_SMART_LOCAL_MAX_SEC", "10800")))
|
||||
_AURORA_SMART_KLING_MAX_SEC = max(60.0, float(os.getenv("AURORA_SMART_KLING_MAX_SEC", "3600")))
|
||||
MEDIA_COMFY_AGENT_URL = os.getenv(
|
||||
"MEDIA_COMFY_AGENT_URL",
|
||||
"http://comfy-agent:8880" if _is_container_runtime() else "http://127.0.0.1:8880",
|
||||
@@ -352,6 +369,10 @@ async def lifespan(app_: Any):
|
||||
|
||||
task = asyncio.create_task(_nodes_poll_loop())
|
||||
logger.info("Nodes poll loop started (interval=%ds)", _NODES_POLL_INTERVAL)
|
||||
try:
|
||||
_smart_resume_active_monitors()
|
||||
except Exception as e:
|
||||
logger.warning("aurora smart monitor resume failed: %s", e)
|
||||
yield
|
||||
task.cancel()
|
||||
try:
|
||||
@@ -972,6 +993,505 @@ def _aurora_persist_live_last_to_disk() -> None:
|
||||
logger.debug("aurora live-last persist failed: %s", e)
|
||||
|
||||
|
||||
def _smart_now_iso() -> str:
|
||||
return datetime.now(timezone.utc).isoformat()
|
||||
|
||||
|
||||
def _smart_is_terminal(status: Any) -> bool:
|
||||
return str(status or "").lower() in {"completed", "failed", "cancelled"}
|
||||
|
||||
|
||||
def _smart_media_type(file_name: str, content_type: str) -> str:
|
||||
name = str(file_name or "").lower()
|
||||
ctype = str(content_type or "").lower()
|
||||
video_ext = (".mp4", ".avi", ".mov", ".mkv", ".webm")
|
||||
audio_ext = (".mp3", ".wav", ".flac", ".m4a", ".aac", ".ogg")
|
||||
image_ext = (".jpg", ".jpeg", ".png", ".webp", ".tif", ".tiff", ".bmp")
|
||||
if ctype.startswith("video/") or name.endswith(video_ext):
|
||||
return "video"
|
||||
if ctype.startswith("audio/") or name.endswith(audio_ext):
|
||||
return "audio"
|
||||
if ctype.startswith("image/") or name.endswith(image_ext):
|
||||
return "photo"
|
||||
return "unknown"
|
||||
|
||||
|
||||
def _smart_trim_runs() -> None:
|
||||
if len(_aurora_smart_runs) <= _AURORA_SMART_MAX_RUNS:
|
||||
return
|
||||
ordered = sorted(
|
||||
_aurora_smart_runs.items(),
|
||||
key=lambda kv: str((kv[1] or {}).get("created_at") or ""),
|
||||
reverse=True,
|
||||
)
|
||||
keep = dict(ordered[:_AURORA_SMART_MAX_RUNS])
|
||||
_aurora_smart_runs.clear()
|
||||
_aurora_smart_runs.update(keep)
|
||||
|
||||
|
||||
def _smart_load_runs_from_disk() -> None:
|
||||
global _aurora_smart_runs_loaded
|
||||
if _aurora_smart_runs_loaded:
|
||||
return
|
||||
_aurora_smart_runs_loaded = True
|
||||
try:
|
||||
if not _aurora_smart_runs_path.exists():
|
||||
return
|
||||
payload = json.loads(_aurora_smart_runs_path.read_text(encoding="utf-8"))
|
||||
if isinstance(payload, dict):
|
||||
runs = payload.get("runs")
|
||||
else:
|
||||
runs = payload
|
||||
if isinstance(runs, dict):
|
||||
for run_id, run in runs.items():
|
||||
if isinstance(run_id, str) and isinstance(run, dict):
|
||||
_aurora_smart_runs[run_id] = run
|
||||
_smart_trim_runs()
|
||||
except Exception as exc:
|
||||
logger.debug("aurora smart-runs load failed: %s", exc)
|
||||
|
||||
|
||||
def _smart_persist_runs() -> None:
|
||||
try:
|
||||
_smart_trim_runs()
|
||||
_aurora_smart_runs_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
payload = {
|
||||
"updated_at": _smart_now_iso(),
|
||||
"runs": _aurora_smart_runs,
|
||||
}
|
||||
_aurora_smart_runs_path.write_text(
|
||||
json.dumps(payload, ensure_ascii=False, separators=(",", ":")),
|
||||
encoding="utf-8",
|
||||
)
|
||||
except Exception as exc:
|
||||
logger.debug("aurora smart-runs persist failed: %s", exc)
|
||||
|
||||
|
||||
def _smart_load_policy_from_disk() -> None:
|
||||
global _aurora_smart_policy_loaded
|
||||
if _aurora_smart_policy_loaded:
|
||||
return
|
||||
_aurora_smart_policy_loaded = True
|
||||
try:
|
||||
if not _aurora_smart_policy_path.exists():
|
||||
return
|
||||
payload = json.loads(_aurora_smart_policy_path.read_text(encoding="utf-8"))
|
||||
if isinstance(payload, dict):
|
||||
strategies = payload.get("strategies")
|
||||
if isinstance(strategies, dict):
|
||||
_aurora_smart_policy["strategies"] = strategies
|
||||
_aurora_smart_policy["updated_at"] = payload.get("updated_at")
|
||||
except Exception as exc:
|
||||
logger.debug("aurora smart-policy load failed: %s", exc)
|
||||
|
||||
|
||||
def _smart_persist_policy() -> None:
|
||||
try:
|
||||
_aurora_smart_policy["updated_at"] = _smart_now_iso()
|
||||
_aurora_smart_policy_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
_aurora_smart_policy_path.write_text(
|
||||
json.dumps(_aurora_smart_policy, ensure_ascii=False, separators=(",", ":")),
|
||||
encoding="utf-8",
|
||||
)
|
||||
except Exception as exc:
|
||||
logger.debug("aurora smart-policy persist failed: %s", exc)
|
||||
|
||||
|
||||
def _smart_strategy_stats(strategy: str) -> Dict[str, Any]:
|
||||
_smart_load_policy_from_disk()
|
||||
strategies = _aurora_smart_policy.setdefault("strategies", {})
|
||||
stats = strategies.get(strategy)
|
||||
if not isinstance(stats, dict):
|
||||
stats = {"count": 0, "avg_score": 0.0, "wins": 0, "losses": 0}
|
||||
strategies[strategy] = stats
|
||||
return stats
|
||||
|
||||
|
||||
def _smart_update_strategy_score(strategy: str, score: float) -> None:
|
||||
stats = _smart_strategy_stats(strategy)
|
||||
try:
|
||||
count = int(stats.get("count") or 0) + 1
|
||||
avg = float(stats.get("avg_score") or 0.0)
|
||||
stats["avg_score"] = round(((avg * (count - 1)) + float(score)) / max(1, count), 4)
|
||||
stats["count"] = count
|
||||
_smart_persist_policy()
|
||||
except Exception:
|
||||
return
|
||||
|
||||
|
||||
def _smart_update_strategy_outcome(strategy: str, success: bool) -> None:
|
||||
stats = _smart_strategy_stats(strategy)
|
||||
key = "wins" if success else "losses"
|
||||
stats[key] = int(stats.get(key) or 0) + 1
|
||||
_smart_persist_policy()
|
||||
|
||||
|
||||
def _smart_new_run_id() -> str:
|
||||
stamp = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S")
|
||||
return f"smart_{stamp}_{uuid.uuid4().hex[:6]}"
|
||||
|
||||
|
||||
def _smart_append_audit(run: Dict[str, Any], event: str, detail: Optional[Dict[str, Any]] = None) -> None:
|
||||
audit = run.setdefault("audit", [])
|
||||
if not isinstance(audit, list):
|
||||
audit = []
|
||||
run["audit"] = audit
|
||||
item: Dict[str, Any] = {"ts": _smart_now_iso(), "event": str(event)}
|
||||
if isinstance(detail, dict) and detail:
|
||||
item["detail"] = detail
|
||||
audit.append(item)
|
||||
if len(audit) > 200:
|
||||
del audit[:-200]
|
||||
run["updated_at"] = item["ts"]
|
||||
|
||||
|
||||
def _smart_analysis_features(analysis: Optional[Dict[str, Any]]) -> Dict[str, Any]:
|
||||
if not isinstance(analysis, dict):
|
||||
return {
|
||||
"faces": 0,
|
||||
"plates": 0,
|
||||
"noise": "unknown",
|
||||
"blur": "unknown",
|
||||
"quality_score": 0.0,
|
||||
}
|
||||
faces = len(analysis.get("faces") or []) if isinstance(analysis.get("faces"), list) else 0
|
||||
plates = len(analysis.get("license_plates") or []) if isinstance(analysis.get("license_plates"), list) else 0
|
||||
qa = analysis.get("quality_analysis") if isinstance(analysis.get("quality_analysis"), dict) else {}
|
||||
noise = str(qa.get("noise_level") or "unknown").lower()
|
||||
blur = str(qa.get("blur_level") or "unknown").lower()
|
||||
score = 0.0
|
||||
score += min(2.0, faces * 0.2)
|
||||
score += min(2.0, plates * 0.4)
|
||||
if noise in {"high", "very_high"}:
|
||||
score += 1.0
|
||||
if blur in {"high", "very_high"}:
|
||||
score += 1.0
|
||||
return {
|
||||
"faces": faces,
|
||||
"plates": plates,
|
||||
"noise": noise,
|
||||
"blur": blur,
|
||||
"quality_score": round(score, 3),
|
||||
}
|
||||
|
||||
|
||||
def _smart_decide_strategy(
|
||||
*,
|
||||
media_type: str,
|
||||
mode: str,
|
||||
requested_strategy: str,
|
||||
prefer_quality: bool,
|
||||
budget_tier: str,
|
||||
analysis: Optional[Dict[str, Any]],
|
||||
learning_enabled: bool,
|
||||
) -> Dict[str, Any]:
|
||||
strategy = str(requested_strategy or "auto").strip().lower()
|
||||
valid = {"auto", "local_only", "local_then_kling"}
|
||||
if strategy not in valid:
|
||||
strategy = "auto"
|
||||
|
||||
features = _smart_analysis_features(analysis)
|
||||
reasons: List[str] = []
|
||||
score = 0.0
|
||||
|
||||
if media_type != "video":
|
||||
chosen = "local_only"
|
||||
reasons.append("non-video media -> local stack only")
|
||||
return {"strategy": chosen, "reasons": reasons, "score": 0.0, "features": features}
|
||||
|
||||
if strategy in {"local_only", "local_then_kling"}:
|
||||
reasons.append(f"explicit strategy={strategy}")
|
||||
return {"strategy": strategy, "reasons": reasons, "score": features["quality_score"], "features": features}
|
||||
|
||||
score += float(features["quality_score"])
|
||||
if prefer_quality:
|
||||
score += 1.3
|
||||
reasons.append("prefer_quality=true")
|
||||
if str(mode).lower() == "forensic":
|
||||
score += 0.8
|
||||
reasons.append("forensic mode")
|
||||
|
||||
budget_norm = str(budget_tier or "normal").strip().lower()
|
||||
if budget_norm == "low":
|
||||
score -= 1.4
|
||||
reasons.append("budget_tier=low")
|
||||
elif budget_norm == "high":
|
||||
score += 0.6
|
||||
reasons.append("budget_tier=high")
|
||||
|
||||
if learning_enabled:
|
||||
stats = _smart_strategy_stats("local_then_kling")
|
||||
wins = int(stats.get("wins") or 0)
|
||||
losses = int(stats.get("losses") or 0)
|
||||
total = wins + losses
|
||||
if total >= 6:
|
||||
success_ratio = wins / max(1, total)
|
||||
if success_ratio >= 0.65:
|
||||
score += 0.5
|
||||
reasons.append(f"learned success ratio {success_ratio:.2f}")
|
||||
elif success_ratio <= 0.35:
|
||||
score -= 0.7
|
||||
reasons.append(f"learned low success ratio {success_ratio:.2f}")
|
||||
|
||||
chosen = "local_then_kling" if score >= 2.1 else "local_only"
|
||||
if not reasons:
|
||||
reasons.append("default heuristic")
|
||||
return {"strategy": chosen, "reasons": reasons, "score": round(score, 3), "features": features}
|
||||
|
||||
|
||||
def _smart_compact_result(result_payload: Dict[str, Any]) -> Dict[str, Any]:
|
||||
payload = {}
|
||||
if not isinstance(result_payload, dict):
|
||||
return payload
|
||||
payload["mode"] = result_payload.get("mode")
|
||||
payload["media_type"] = result_payload.get("media_type")
|
||||
payload["digital_signature"] = result_payload.get("digital_signature")
|
||||
output_files = result_payload.get("output_files")
|
||||
if isinstance(output_files, list):
|
||||
payload["output_files"] = output_files[:8]
|
||||
q = result_payload.get("quality_report")
|
||||
if isinstance(q, dict):
|
||||
payload["quality_report"] = q
|
||||
return payload
|
||||
|
||||
|
||||
async def _smart_fetch_run_status(run_id: str) -> Optional[Dict[str, Any]]:
|
||||
_smart_load_runs_from_disk()
|
||||
run = _aurora_smart_runs.get(run_id)
|
||||
if not isinstance(run, dict):
|
||||
return None
|
||||
return run
|
||||
|
||||
|
||||
async def _smart_monitor_run(run_id: str) -> None:
|
||||
run = await _smart_fetch_run_status(run_id)
|
||||
if not run:
|
||||
return
|
||||
|
||||
local = run.get("local") if isinstance(run.get("local"), dict) else {}
|
||||
local_job_id = str(local.get("job_id") or "")
|
||||
if not local_job_id:
|
||||
_smart_append_audit(run, "monitor.error", {"reason": "missing local job id"})
|
||||
run["status"] = "failed"
|
||||
run["phase"] = "failed"
|
||||
_smart_persist_runs()
|
||||
return
|
||||
|
||||
start = time.monotonic()
|
||||
while time.monotonic() - start <= _AURORA_SMART_LOCAL_MAX_SEC:
|
||||
try:
|
||||
st = await _aurora_request_json(
|
||||
"GET",
|
||||
f"/api/aurora/status/{quote(local_job_id, safe='')}",
|
||||
timeout=20.0,
|
||||
retries=2,
|
||||
retry_backoff_sec=0.25,
|
||||
)
|
||||
except Exception as exc:
|
||||
_smart_append_audit(run, "local.status.error", {"error": str(exc)[:220]})
|
||||
await asyncio.sleep(_AURORA_SMART_LOCAL_POLL_SEC)
|
||||
continue
|
||||
|
||||
status = str(st.get("status") or "").lower()
|
||||
if status in {"queued", "processing"}:
|
||||
run["phase"] = "local_processing"
|
||||
run["status"] = "processing"
|
||||
elif status == "completed":
|
||||
run["phase"] = "local_completed"
|
||||
run["status"] = "processing"
|
||||
else:
|
||||
run["phase"] = f"local_{status or 'unknown'}"
|
||||
run["status"] = status
|
||||
run["local"] = {
|
||||
**local,
|
||||
"job_id": local_job_id,
|
||||
"status": status,
|
||||
"progress": st.get("progress"),
|
||||
"current_stage": st.get("current_stage"),
|
||||
"eta_seconds": st.get("eta_seconds"),
|
||||
"live_fps": st.get("live_fps"),
|
||||
"error_message": st.get("error_message"),
|
||||
"updated_at": _smart_now_iso(),
|
||||
}
|
||||
_smart_persist_runs()
|
||||
|
||||
if status in {"queued", "processing"}:
|
||||
await asyncio.sleep(_AURORA_SMART_LOCAL_POLL_SEC)
|
||||
continue
|
||||
|
||||
if status != "completed":
|
||||
run["status"] = "failed"
|
||||
run["phase"] = "local_failed"
|
||||
_smart_append_audit(
|
||||
run,
|
||||
"local.failed",
|
||||
{"status": status, "error": str(st.get("error_message") or "")[:220]},
|
||||
)
|
||||
_smart_update_strategy_outcome(str(run.get("policy", {}).get("strategy") or "local_only"), False)
|
||||
_smart_persist_runs()
|
||||
return
|
||||
|
||||
_smart_append_audit(run, "local.completed", {"job_id": local_job_id})
|
||||
break
|
||||
else:
|
||||
run["status"] = "failed"
|
||||
run["phase"] = "local_timeout"
|
||||
_smart_append_audit(run, "local.timeout", {"max_sec": _AURORA_SMART_LOCAL_MAX_SEC})
|
||||
_smart_update_strategy_outcome(str(run.get("policy", {}).get("strategy") or "local_only"), False)
|
||||
_smart_persist_runs()
|
||||
return
|
||||
|
||||
try:
|
||||
local_result = await _aurora_request_json(
|
||||
"GET",
|
||||
f"/api/aurora/result/{quote(local_job_id, safe='')}",
|
||||
timeout=30.0,
|
||||
retries=2,
|
||||
retry_backoff_sec=0.25,
|
||||
)
|
||||
except Exception as exc:
|
||||
run["status"] = "failed"
|
||||
run["phase"] = "local_result_error"
|
||||
_smart_append_audit(run, "local.result.error", {"error": str(exc)[:240]})
|
||||
_smart_update_strategy_outcome(str(run.get("policy", {}).get("strategy") or "local_only"), False)
|
||||
_smart_persist_runs()
|
||||
return
|
||||
|
||||
run.setdefault("local", {})
|
||||
if isinstance(run["local"], dict):
|
||||
run["local"]["result"] = _smart_compact_result(local_result)
|
||||
run["local"]["result_ready"] = True
|
||||
run["selected_stack"] = "local"
|
||||
|
||||
policy = run.get("policy") if isinstance(run.get("policy"), dict) else {}
|
||||
strategy = str(policy.get("strategy") or "local_only")
|
||||
media_type = str(run.get("media_type") or "")
|
||||
kling = run.get("kling") if isinstance(run.get("kling"), dict) else {}
|
||||
if strategy != "local_then_kling" or media_type != "video":
|
||||
run["status"] = "completed"
|
||||
run["phase"] = "completed"
|
||||
_smart_append_audit(run, "smart.completed", {"selected_stack": "local", "reason": "strategy local_only or non-video"})
|
||||
_smart_update_strategy_outcome(strategy, True)
|
||||
_smart_persist_runs()
|
||||
return
|
||||
|
||||
run["phase"] = "kling_submitting"
|
||||
run["status"] = "processing"
|
||||
_smart_append_audit(run, "kling.submit.start")
|
||||
_smart_persist_runs()
|
||||
|
||||
try:
|
||||
submit = await _aurora_request_json(
|
||||
"POST",
|
||||
"/api/aurora/kling/enhance",
|
||||
data={
|
||||
"job_id": local_job_id,
|
||||
"prompt": str(kling.get("prompt") or "enhance video quality, improve sharpness and clarity"),
|
||||
"negative_prompt": str(kling.get("negative_prompt") or "noise, blur, artifacts, distortion"),
|
||||
"mode": str(kling.get("mode") or "pro"),
|
||||
"duration": str(kling.get("duration") or "5"),
|
||||
"cfg_scale": str(kling.get("cfg_scale") if kling.get("cfg_scale") is not None else "0.5"),
|
||||
},
|
||||
timeout=120.0,
|
||||
retries=1,
|
||||
retry_backoff_sec=0.25,
|
||||
)
|
||||
except Exception as exc:
|
||||
run["kling"] = {
|
||||
**kling,
|
||||
"status": "failed",
|
||||
"error": str(exc)[:320],
|
||||
}
|
||||
run["status"] = "completed"
|
||||
run["phase"] = "completed_with_kling_failure"
|
||||
run["selected_stack"] = "local"
|
||||
_smart_append_audit(run, "kling.submit.error", {"error": str(exc)[:220]})
|
||||
_smart_update_strategy_outcome(strategy, False)
|
||||
_smart_persist_runs()
|
||||
return
|
||||
|
||||
task_id = str(submit.get("kling_task_id") or "")
|
||||
run["kling"] = {
|
||||
**kling,
|
||||
"task_id": task_id,
|
||||
"status": str(submit.get("status") or "submitted").lower(),
|
||||
"submitted_at": _smart_now_iso(),
|
||||
}
|
||||
_smart_append_audit(run, "kling.submitted", {"task_id": task_id})
|
||||
_smart_persist_runs()
|
||||
|
||||
k_start = time.monotonic()
|
||||
while time.monotonic() - k_start <= _AURORA_SMART_KLING_MAX_SEC:
|
||||
try:
|
||||
kst = await _aurora_request_json(
|
||||
"GET",
|
||||
f"/api/aurora/kling/status/{quote(local_job_id, safe='')}",
|
||||
timeout=30.0,
|
||||
retries=1,
|
||||
retry_backoff_sec=0.2,
|
||||
)
|
||||
except Exception as exc:
|
||||
_smart_append_audit(run, "kling.status.error", {"error": str(exc)[:220]})
|
||||
await asyncio.sleep(_AURORA_SMART_KLING_POLL_SEC)
|
||||
continue
|
||||
|
||||
k_status = str(kst.get("status") or "").lower()
|
||||
k_url = kst.get("kling_result_url")
|
||||
run["phase"] = "kling_processing"
|
||||
run["kling"] = {
|
||||
**(run.get("kling") if isinstance(run.get("kling"), dict) else {}),
|
||||
"status": k_status,
|
||||
"result_url": k_url,
|
||||
"last_polled_at": _smart_now_iso(),
|
||||
}
|
||||
_smart_persist_runs()
|
||||
|
||||
if k_status in {"submitted", "queued", "running", "processing", "pending"}:
|
||||
await asyncio.sleep(_AURORA_SMART_KLING_POLL_SEC)
|
||||
continue
|
||||
|
||||
if k_status in {"succeed", "completed", "success"} and k_url:
|
||||
run["status"] = "completed"
|
||||
run["phase"] = "completed"
|
||||
run["selected_stack"] = "kling"
|
||||
_smart_append_audit(run, "smart.completed", {"selected_stack": "kling", "task_id": task_id})
|
||||
_smart_update_strategy_outcome(strategy, True)
|
||||
_smart_persist_runs()
|
||||
return
|
||||
|
||||
run["status"] = "completed"
|
||||
run["phase"] = "completed_with_kling_failure"
|
||||
run["selected_stack"] = "local"
|
||||
_smart_append_audit(
|
||||
run,
|
||||
"kling.terminal.non_success",
|
||||
{"status": k_status, "task_id": task_id},
|
||||
)
|
||||
_smart_update_strategy_outcome(strategy, False)
|
||||
_smart_persist_runs()
|
||||
return
|
||||
|
||||
run["status"] = "completed"
|
||||
run["phase"] = "completed_with_kling_timeout"
|
||||
run["selected_stack"] = "local"
|
||||
_smart_append_audit(run, "kling.timeout", {"max_sec": _AURORA_SMART_KLING_MAX_SEC})
|
||||
_smart_update_strategy_outcome(strategy, False)
|
||||
_smart_persist_runs()
|
||||
|
||||
|
||||
def _smart_resume_active_monitors() -> None:
|
||||
_smart_load_runs_from_disk()
|
||||
for run_id, run in list(_aurora_smart_runs.items()):
|
||||
if not isinstance(run, dict):
|
||||
continue
|
||||
if _smart_is_terminal(run.get("status")):
|
||||
continue
|
||||
try:
|
||||
asyncio.create_task(_smart_monitor_run(run_id))
|
||||
except Exception:
|
||||
continue
|
||||
|
||||
|
||||
@app.get("/api/aurora/health")
|
||||
async def api_aurora_health() -> Dict[str, Any]:
|
||||
return await _aurora_request_json("GET", "/health", timeout=10.0)
|
||||
@@ -1012,6 +1532,241 @@ async def api_aurora_upload(
|
||||
return payload
|
||||
|
||||
|
||||
@app.post("/api/aurora/process-smart")
|
||||
async def api_aurora_process_smart(
|
||||
file: UploadFile = File(...),
|
||||
mode: str = Form("tactical"),
|
||||
priority: str = Form("balanced"),
|
||||
export_options: str = Form(""),
|
||||
strategy: str = Form("auto"),
|
||||
prefer_quality: bool = Form(True),
|
||||
budget_tier: str = Form("normal"),
|
||||
learning_enabled: bool = Form(True),
|
||||
kling_prompt: str = Form("enhance video quality, improve sharpness and clarity"),
|
||||
kling_negative_prompt: str = Form("noise, blur, artifacts, distortion"),
|
||||
kling_mode: str = Form("pro"),
|
||||
kling_duration: str = Form("5"),
|
||||
kling_cfg_scale: float = Form(0.5),
|
||||
) -> Dict[str, Any]:
|
||||
_smart_load_runs_from_disk()
|
||||
_smart_load_policy_from_disk()
|
||||
|
||||
file_name = file.filename or "upload.bin"
|
||||
content_type = file.content_type or "application/octet-stream"
|
||||
media_type = _smart_media_type(file_name, content_type)
|
||||
|
||||
analysis: Optional[Dict[str, Any]] = None
|
||||
if media_type in {"video", "photo"}:
|
||||
try:
|
||||
await file.seek(0)
|
||||
files = {"file": (file_name, file.file, content_type)}
|
||||
analysis = await _aurora_request_json(
|
||||
"POST",
|
||||
"/api/aurora/analyze",
|
||||
files=files,
|
||||
timeout=120.0,
|
||||
retries=1,
|
||||
retry_backoff_sec=0.25,
|
||||
)
|
||||
except Exception as exc:
|
||||
analysis = None
|
||||
logger.warning("smart-process analyze skipped: %s", str(exc)[:220])
|
||||
|
||||
policy = _smart_decide_strategy(
|
||||
media_type=media_type,
|
||||
mode=mode,
|
||||
requested_strategy=strategy,
|
||||
prefer_quality=bool(prefer_quality),
|
||||
budget_tier=budget_tier,
|
||||
analysis=analysis,
|
||||
learning_enabled=bool(learning_enabled),
|
||||
)
|
||||
chosen_strategy = str(policy.get("strategy") or "local_only")
|
||||
policy.setdefault("requested_strategy", str(strategy or "auto"))
|
||||
policy["learning_enabled"] = bool(learning_enabled)
|
||||
policy["budget_tier"] = str(budget_tier or "normal")
|
||||
|
||||
await file.seek(0)
|
||||
files = {"file": (file_name, file.file, content_type)}
|
||||
local_payload = await _aurora_request_json(
|
||||
"POST",
|
||||
"/api/aurora/upload",
|
||||
files=files,
|
||||
data={
|
||||
"mode": mode,
|
||||
"priority": priority,
|
||||
"export_options": export_options,
|
||||
},
|
||||
timeout=120.0,
|
||||
)
|
||||
local_job_id = str(local_payload.get("job_id") or "")
|
||||
if not local_job_id:
|
||||
raise HTTPException(status_code=502, detail="Smart process failed: local job_id missing")
|
||||
|
||||
run_id = _smart_new_run_id()
|
||||
now = _smart_now_iso()
|
||||
run: Dict[str, Any] = {
|
||||
"run_id": run_id,
|
||||
"created_at": now,
|
||||
"updated_at": now,
|
||||
"status": "processing",
|
||||
"phase": "local_processing",
|
||||
"media_type": media_type,
|
||||
"selected_stack": None,
|
||||
"requested": {
|
||||
"mode": mode,
|
||||
"priority": priority,
|
||||
"export_options": export_options,
|
||||
"strategy": strategy,
|
||||
"prefer_quality": bool(prefer_quality),
|
||||
"budget_tier": budget_tier,
|
||||
"learning_enabled": bool(learning_enabled),
|
||||
},
|
||||
"policy": policy,
|
||||
"analysis_summary": _smart_analysis_features(analysis),
|
||||
"analysis": analysis if isinstance(analysis, dict) else None,
|
||||
"local": {
|
||||
"job_id": local_job_id,
|
||||
"status": "queued",
|
||||
"submit_payload": {
|
||||
"status_url": f"/api/aurora/status/{quote(local_job_id, safe='')}",
|
||||
"result_url": f"/api/aurora/result/{quote(local_job_id, safe='')}",
|
||||
},
|
||||
},
|
||||
"kling": {
|
||||
"enabled": chosen_strategy == "local_then_kling" and media_type == "video",
|
||||
"status": "pending",
|
||||
"prompt": kling_prompt,
|
||||
"negative_prompt": kling_negative_prompt,
|
||||
"mode": kling_mode,
|
||||
"duration": kling_duration,
|
||||
"cfg_scale": kling_cfg_scale,
|
||||
},
|
||||
"audit": [],
|
||||
}
|
||||
_smart_append_audit(
|
||||
run,
|
||||
"smart.submitted",
|
||||
{
|
||||
"local_job_id": local_job_id,
|
||||
"media_type": media_type,
|
||||
"strategy": chosen_strategy,
|
||||
"score": policy.get("score"),
|
||||
},
|
||||
)
|
||||
_aurora_smart_runs[run_id] = run
|
||||
_smart_persist_runs()
|
||||
|
||||
try:
|
||||
asyncio.create_task(_smart_monitor_run(run_id))
|
||||
except Exception as exc:
|
||||
_smart_append_audit(run, "monitor.spawn.error", {"error": str(exc)[:220]})
|
||||
_smart_persist_runs()
|
||||
|
||||
return {
|
||||
"smart_run_id": run_id,
|
||||
"status": run.get("status"),
|
||||
"phase": run.get("phase"),
|
||||
"media_type": media_type,
|
||||
"local_job_id": local_job_id,
|
||||
"policy": policy,
|
||||
"smart_status_url": f"/api/aurora/process-smart/{quote(run_id, safe='')}",
|
||||
"local_status_url": f"/api/aurora/status/{quote(local_job_id, safe='')}",
|
||||
"local_result_url": f"/api/aurora/result/{quote(local_job_id, safe='')}",
|
||||
}
|
||||
|
||||
|
||||
@app.get("/api/aurora/process-smart")
|
||||
async def api_aurora_process_smart_list(
|
||||
limit: int = Query(default=20, ge=1, le=200),
|
||||
status: Optional[str] = Query(default=None),
|
||||
) -> Dict[str, Any]:
|
||||
_smart_load_runs_from_disk()
|
||||
requested = str(status or "").strip().lower()
|
||||
rows = []
|
||||
for run in _aurora_smart_runs.values():
|
||||
if not isinstance(run, dict):
|
||||
continue
|
||||
run_status = str(run.get("status") or "")
|
||||
if requested and run_status.lower() != requested:
|
||||
continue
|
||||
local = run.get("local") if isinstance(run.get("local"), dict) else {}
|
||||
kling = run.get("kling") if isinstance(run.get("kling"), dict) else {}
|
||||
rows.append(
|
||||
{
|
||||
"run_id": run.get("run_id"),
|
||||
"status": run_status,
|
||||
"phase": run.get("phase"),
|
||||
"media_type": run.get("media_type"),
|
||||
"strategy": (run.get("policy") or {}).get("strategy") if isinstance(run.get("policy"), dict) else None,
|
||||
"selected_stack": run.get("selected_stack"),
|
||||
"created_at": run.get("created_at"),
|
||||
"updated_at": run.get("updated_at"),
|
||||
"local_job_id": local.get("job_id"),
|
||||
"local_status": local.get("status"),
|
||||
"kling_status": kling.get("status"),
|
||||
}
|
||||
)
|
||||
rows.sort(key=lambda x: str(x.get("created_at") or ""), reverse=True)
|
||||
return {"runs": rows[:limit], "count": min(limit, len(rows)), "total": len(rows)}
|
||||
|
||||
|
||||
@app.get("/api/aurora/process-smart/{run_id}")
|
||||
async def api_aurora_process_smart_status(run_id: str) -> Dict[str, Any]:
|
||||
run = await _smart_fetch_run_status(run_id)
|
||||
if not run:
|
||||
raise HTTPException(status_code=404, detail="smart run not found")
|
||||
return run
|
||||
|
||||
|
||||
@app.post("/api/aurora/process-smart/{run_id}/feedback")
|
||||
async def api_aurora_process_smart_feedback(
|
||||
run_id: str,
|
||||
payload: Optional[Dict[str, Any]] = Body(default=None),
|
||||
) -> Dict[str, Any]:
|
||||
run = await _smart_fetch_run_status(run_id)
|
||||
if not run:
|
||||
raise HTTPException(status_code=404, detail="smart run not found")
|
||||
body = payload if isinstance(payload, dict) else {}
|
||||
score_raw = body.get("score")
|
||||
score: Optional[float] = None
|
||||
try:
|
||||
if score_raw is not None:
|
||||
score = float(score_raw)
|
||||
except Exception:
|
||||
score = None
|
||||
selected_stack = str(body.get("selected_stack") or "").strip().lower() or None
|
||||
notes = str(body.get("notes") or "").strip()
|
||||
|
||||
feedback = {
|
||||
"ts": _smart_now_iso(),
|
||||
"score": score,
|
||||
"selected_stack": selected_stack,
|
||||
"notes": notes[:1000] if notes else None,
|
||||
}
|
||||
run["feedback"] = feedback
|
||||
strategy = str((run.get("policy") or {}).get("strategy") or "local_only")
|
||||
if score is not None:
|
||||
score = max(1.0, min(5.0, score))
|
||||
_smart_update_strategy_score(strategy, score)
|
||||
if selected_stack in {"local", "kling"}:
|
||||
run["selected_stack"] = selected_stack
|
||||
_smart_append_audit(run, "feedback.received", {"score": score, "selected_stack": selected_stack})
|
||||
_smart_persist_runs()
|
||||
return {
|
||||
"ok": True,
|
||||
"run_id": run_id,
|
||||
"feedback": feedback,
|
||||
"policy": _aurora_smart_policy,
|
||||
}
|
||||
|
||||
|
||||
@app.get("/api/aurora/process-smart/policy/stats")
|
||||
async def api_aurora_process_smart_policy_stats() -> Dict[str, Any]:
|
||||
_smart_load_policy_from_disk()
|
||||
return _aurora_smart_policy
|
||||
|
||||
|
||||
@app.post("/api/aurora/analyze")
|
||||
async def api_aurora_analyze(file: UploadFile = File(...)) -> Dict[str, Any]:
|
||||
await file.seek(0)
|
||||
|
||||
Reference in New Issue
Block a user