"""자동 스킬 획득 — 운영 패턴 감지 + 스킬 자동 생성""" from __future__ import annotations import json, logging from datetime import datetime, timedelta from typing import Optional import httpx from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException from pydantic import BaseModel from sqlalchemy import select, desc, func from sqlalchemy.ext.asyncio import AsyncSession from core.auth import get_current_user, require_admin_role from database import get_db from models import User, MinedPattern, Skill logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/skill-miner", tags=["자동 스킬 획득"]) OLLAMA_URL = "http://localhost:11434" async def _generate_skill_code(pattern_cmds: list[str], desc: str) -> str: """Ollama로 스킬 코드 자동 생성.""" try: async with httpx.AsyncClient(timeout=20) as c: r = await c.post(f"{OLLAMA_URL}/api/generate", json={ "model": "llama3", "system": "Shell 명령 패턴을 보고 재사용 가능한 단일 명령으로 요약. 한 줄로만 답변.", "prompt": f"반복 명령:\n{chr(10).join(pattern_cmds)}\n설명: {desc}", "stream": False, }) return r.json().get("response", "# 자동 생성 실패").strip()[:500] except Exception as e: return f"# 자동 생성 실패: {e}" class PatternAnalyzeIn(BaseModel): commands: list[str]; server_id: Optional[int] = None; context: str = "" class PatternValidateIn(BaseModel): pattern_id: int; approve: bool; custom_code: Optional[str] = None @router.post("/analyze") async def analyze_pattern(body: PatternAnalyzeIn, background_tasks: BackgroundTasks, db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user)): """명령 시퀀스를 분석해 패턴 후보 등록.""" if len(body.commands) < 2: return {"ok": False, "message": "최소 2개 명령 필요"} # 중복 패턴 확인 cmd_key = json.dumps(sorted(body.commands)) existing = await db.execute( select(MinedPattern).where(MinedPattern.trigger == cmd_key[:200]) ) pattern = existing.scalar_one_or_none() if pattern: pattern.occurrence_count = (pattern.occurrence_count or 0) + 1 pattern.last_seen = datetime.utcnow() await db.commit() return {"pattern_id": pattern.id, "occurrence_count": pattern.occurrence_count, "auto_skill": pattern.occurrence_count >= 3} # 신규 패턴 등록 pattern = MinedPattern( pattern_type="COMMAND_SEQ", trigger=cmd_key[:200], action_sequence=json.dumps(body.commands), occurrence_count=1, last_seen=datetime.utcnow(), status="PENDING", context=body.context, ) db.add(pattern); await db.commit(); await db.refresh(pattern) # 3회 이상이면 자동 스킬 생성 제안 if pattern.occurrence_count >= 3: background_tasks.add_task(_auto_create_skill, pattern.id, body.commands, db) return {"pattern_id": pattern.id, "occurrence_count": 1, "auto_skill": False} async def _auto_create_skill(pattern_id: int, cmds: list, db: AsyncSession): """백그라운드: 패턴 → 스킬 자동 생성.""" code = await _generate_skill_code(cmds, f"패턴 #{pattern_id}") from sqlalchemy import update as sa_update async with db.begin(): await db.execute(sa_update(MinedPattern).where(MinedPattern.id == pattern_id) .values(status="SKILL_PROPOSED", generated_code=code)) @router.get("/patterns") async def list_patterns(db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user)): rows = await db.execute( select(MinedPattern).order_by(desc(MinedPattern.occurrence_count)).limit(30) ) patterns = rows.scalars().all() return [{ "id": p.id, "type": p.pattern_type, "trigger": p.trigger[:80], "occurrence_count": p.occurrence_count, "status": p.status, "last_seen": p.last_seen, "has_code": bool(p.generated_code), } for p in patterns] @router.post("/create") async def create_from_pattern(body: PatternValidateIn, db: AsyncSession = Depends(get_db), user: User = Depends(require_admin_role)): """패턴 → 스킬 변환 (관리자 승인).""" row = await db.execute(select(MinedPattern).where(MinedPattern.id == body.pattern_id)) pattern = row.scalar_one_or_none() if not pattern: raise HTTPException(404) if not body.approve: from sqlalchemy import update as sa_update await db.execute(sa_update(MinedPattern).where(MinedPattern.id == body.pattern_id) .values(status="REJECTED")) await db.commit() return {"ok": True, "status": "REJECTED"} code = body.custom_code or pattern.generated_code or json.loads(pattern.action_sequence or "[]")[0] skill = Skill( name=f"auto-{pattern.id}", description=f"자동 생성: {pattern.trigger[:50]}", category="MINED", code_template=code, parameters_schema='{"server_id":"int"}', created_from="MINED", success_count=0, fail_count=0, is_active=True, created_at=datetime.utcnow(), ) db.add(skill) from sqlalchemy import update as sa_update await db.execute(sa_update(MinedPattern).where(MinedPattern.id == body.pattern_id) .values(status="CONVERTED")) await db.commit() return {"ok": True, "skill_id": skill.id} @router.post("/validate") async def validate_skill(pattern_id: int, db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user)): """패턴 스킬 검증 (테스트 실행).""" row = await db.execute(select(MinedPattern).where(MinedPattern.id == pattern_id)) p = row.scalar_one_or_none() if not p: raise HTTPException(404) return {"pattern_id": pattern_id, "generated_code": p.generated_code or "없음", "status": p.status, "validation": "수동 검토 필요"} @router.get("/queue") async def skill_queue(db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user)): """스킬화 대기 패턴 목록.""" rows = await db.execute( select(MinedPattern).where(MinedPattern.status.in_(["PENDING", "SKILL_PROPOSED"])) .order_by(desc(MinedPattern.occurrence_count)) ) items = rows.scalars().all() return [{"id": p.id, "trigger": p.trigger[:60], "count": p.occurrence_count, "status": p.status, "has_code": bool(p.generated_code)} for p in items] @router.post("/mine-github") async def mine_github(repo_url: str = "anthropics/claude-plugins-official", user: User = Depends(require_admin_role)): """GitHub 저장소에서 스킬 패턴 마이닝 (시뮬레이션).""" return { "repo": repo_url, "status": "simulated", "found_patterns": [ "deploy-and-verify", "rollback-on-failure", "health-check-loop" ], "note": "실제 구현 시 GitHub API + Ollama 코드 분석 사용", }