test-mcp/analysis_service.py

import re
import math
import statistics
from datetime import datetime, timedelta
from sql_queries import DashboardQueries
from prediction_service import SOIPredictionService

class AnalysisService:
    """프로젝트 통계 및 활동성 분석 전문 서비스"""

    @staticmethod
    def calculate_operational_consistency(history_rows, days_stagnant):
        """운영 일관성 지수(OCI) 산출 로직 (장기 정체 패널티 포함)
        최근 30일간 활동 리듬 분석 + 현재 방치 기간에 따른 강력한 감쇄
        """
        if not history_rows or len(history_rows) < 2:
            return 0.0

        # 1. 최근 30일 이력 기반 Base Score 산출
        now = datetime.now().date()
        recent_30 = [h for h in history_rows if (now - h['crawl_date']).days <= 30]
        
        # 주차별 활동 여부 (4주)
        weeks_active = [False, False, False, False]
        for h in recent_30:
            days_ago = (now - h['crawl_date']).days
            week_idx = min(3, days_ago // 7)
            weeks_active[week_idx] = True
        
        base_consistency = (sum(weeks_active) / 4) * 70
        
        # 활동 밀도 (변화 발생일 비율)
        effort_days = 0
        for i in range(1, len(recent_30)):
            if recent_30[i]['file_count'] != recent_30[i-1]['file_count']:
                effort_days += 1
        
        density_score = (effort_days / max(1, len(recent_30))) * 30
        base_oci = base_consistency + density_score

        # 2. [핵심] 장기 정체 패널티 적용
        # 방치일이 100일 이상이면 OCI는 0점으로 수렴 (성실도 무효화)
        stagnation_factor = max(0, (100 - days_stagnant) / 100.0)
        
        final_oci = base_oci * stagnation_factor
        
        return round(final_oci, 1)

    @staticmethod
    def calculate_activity_status(target_date_dt, log, file_count):
        """개별 프로젝트의 활동 상태 및 방치일 산출"""
        status, days = "unknown", 999
        file_val = int(file_count) if file_count else 0
        has_log = log and log != "데이터 없음" and log != "X"
        
        if file_val == 0:
            status = "unknown"
        elif has_log:
            if "폴더자동삭제" in log.replace(" ", ""):
                status = "stale"
                days = 999
            else:
                match = re.search(r'(\d{4})\.(\d{2})\.(\d{2})', log)
                if match:
                    log_date = datetime.strptime(match.group(0), "%Y.%m.%d")
                    diff = (target_date_dt - log_date).days
                    status = "active" if diff <= 7 else "warning" if diff <= 14 else "stale"
                    days = diff
                else:
                    status = "stale"
        else:
            status = "stale"
            
        return status, days

    @staticmethod
    def get_project_activity_logic(cursor, date_str):
        """활동도 분석 리포트 생성 로직"""
        if not date_str or date_str == "-":
            cursor.execute(DashboardQueries.GET_LAST_CRAWL_DATE)
            res = cursor.fetchone()
            target_date_val = res['last_date'] if res['last_date'] else datetime.now().date()
        else:
            target_date_val = datetime.strptime(date_str.replace(".", "-"), "%Y-%m-%d").date()

        target_date_dt = datetime.combine(target_date_val, datetime.min.time())
        cursor.execute(DashboardQueries.GET_PROJECT_LIST_FOR_ANALYSIS, (target_date_val,))
        rows = cursor.fetchall()
        
        analysis = {"summary": {"active": 0, "warning": 0, "stale": 0, "unknown": 0}, "details": []}
        for r in rows:
            status, days = AnalysisService.calculate_activity_status(target_date_dt, r['recent_log'], r['file_count'])
            analysis["summary"][status] += 1
            analysis["details"].append({"name": r['short_nm'] or r['project_nm'], "status": status, "days_ago": days})
            
        return analysis

    @staticmethod
    def get_p_zsr_analysis_logic(cursor):
        """절대적 방치 실태 고발 및 운영 일관성(OCI) 분석 로직"""
        cursor.execute(DashboardQueries.GET_LAST_CRAWL_DATE)
        res_date = cursor.fetchone()
        if not res_date or not res_date['last_date']:
            return []
        last_date = res_date['last_date']
        
        cursor.execute("""
            SELECT m.project_id, m.project_nm, m.short_nm, m.department, m.master, 
                   h.recent_log, h.file_count, m.continent, m.country 
            FROM projects_master m
            LEFT JOIN projects_history h ON m.project_id = h.project_id AND h.crawl_date = %s
            ORDER BY m.project_id ASC
        """, (last_date,))
        projects = cursor.fetchall()
        
        if not projects: return []

        results = []
        total_soi = 0
        
        for p in projects:
            file_count = int(p['file_count']) if p['file_count'] else 0
            log = p['recent_log']
            
            # 방치일 계산
            days_stagnant = 14
            if log and log != "데이터 없음":
                match = re.search(r'(\d{4})\.(\d{2})\.(\d{2})', log)
                if match:
                    log_date = datetime.strptime(match.group(0), "%Y.%m.%d").date()
                    days_stagnant = (last_date - log_date).days
            
            is_auto_delete = log and "폴더자동삭제" in log.replace(" ", "")
            
            # AI-Hazard 추론 로직 (Dynamic Lambda)
            scale_impact = min(0.04, math.log10(file_count + 1) * 0.008) if file_count > 0 else 0
            ai_lambda = 0.04 + scale_impact
            
            # 지수 감쇄 적용
            soi_score = math.exp(-ai_lambda * days_stagnant) * 100

            # ECV 패널티
            existence_confidence = 1.0
            if file_count == 0: existence_confidence = 0.05 
            elif file_count < 10: existence_confidence = 0.4 
            
            # Log Quality Scoring
            log_quality_factor = 1.0
            if log and log != "데이터 없음":
                if any(k in log for k in ["업로드", "수정", "등록", "변환", "파일", "업데이트"]): log_quality_factor = 1.0
                elif any(k in log for k in ["폴더", "생성", "삭제", "이동"]): log_quality_factor = 0.7
                elif any(k in log for k in ["참가자", "권한", "추가", "변경", "메일"]): log_quality_factor = 0.4
                else: log_quality_factor = 0.6
            
            soi_score = soi_score * existence_confidence * log_quality_factor
            if is_auto_delete: soi_score = 0.1 

            # [운영 일관성 분석 (OCI)]
            history_rows = SOIPredictionService.get_historical_soi(cursor, p['project_id'])
            oci_score = AnalysisService.calculate_operational_consistency(history_rows, days_stagnant)
            
            # 실무 투입 에너지 계산
            effort_days = 0
            if len(history_rows) > 1:
                for i in range(1, len(history_rows)):
                    if history_rows[i]['file_count'] != history_rows[i-1]['file_count']:
                        effort_days += 1
            
            work_effort_rate = round((effort_days / max(1, len(history_rows))) * 100, 1)
            total_soi += soi_score
            
            # VCI 산출
            REPLACEMENT_LEVEL = 70.0
            asset_weight = (file_count / 200.0) + 0.5
            p_war_score = (soi_score - REPLACEMENT_LEVEL) * asset_weight

            results.append({
                "project_nm": p['short_nm'] or p['project_nm'],
                "file_count": file_count,
                "days_stagnant": days_stagnant,
                "risk_count": round(p_war_score, 2), 
                "p_war": round(soi_score, 1),
                "oci_score": oci_score, # 운영 일관성 지수 추가
                "is_auto_delete": is_auto_delete,
                "master": p['master'],
                "dept": p['department'],
                "ai_lambda": round(ai_lambda, 4),
                "log_quality": log_quality_factor,
                "work_effort": work_effort_rate,
                "avg_info": {
                    "avg_files": 0,
                    "avg_stagnant": 0,
                    "avg_risk": round(total_soi / len(projects), 1)
                }
            })
        
        results.sort(key=lambda x: x['p_war'])
        return results