""" ORM models + Pydantic schemas for GUARDiA ITSM """ from __future__ import annotations import hashlib import json from datetime import datetime, date from enum import Enum from typing import Any, Optional, List from pydantic import BaseModel, ConfigDict from sqlalchemy import ( Boolean, Column, Date, DateTime, Float, ForeignKey, Integer, JSON, String, Text, func ) from sqlalchemy.orm import relationship, backref from database import Base # ── Enums ────────────────────────────────────────────────────────────────────── class SRStatus(str, Enum): RECEIVED = "RECEIVED" PARSED = "PARSED" PENDING_APPROVAL = "PENDING_APPROVAL" APPROVED = "APPROVED" IN_PROGRESS = "IN_PROGRESS" PENDING_PM_VALIDATION = "PENDING_PM_VALIDATION" COMPLETED = "COMPLETED" FAILED_ROLLBACK = "FAILED_ROLLBACK" REJECTED = "REJECTED" class SRType(str, Enum): DEPLOY = "DEPLOY" RESTART = "RESTART" LOG = "LOG" INQUIRY = "INQUIRY" OTHER = "OTHER" class Priority(str, Enum): CRITICAL = "CRITICAL" HIGH = "HIGH" MEDIUM = "MEDIUM" LOW = "LOW" class ApprovalResult(str, Enum): PENDING = "PENDING" APPROVED = "APPROVED" REJECTED = "REJECTED" # ── ORM Models ───────────────────────────────────────────────────────────────── class Institution(Base): __tablename__ = "tb_inst_meta" id = Column(Integer, primary_key=True, index=True) inst_code = Column(String(20), unique=True, nullable=False, index=True) inst_name = Column(String(100), nullable=False) org_type = Column(String(50)) contact_pm = Column(String(100)) address = Column(String(300)) region = Column(String(50)) phone = Column(String(20)) contract_start = Column(Date) contract_end = Column(Date) sla_hours = Column(Integer, default=4) note = Column(Text) is_active = Column(Boolean, default=True) created_at = Column(DateTime, default=func.now()) updated_at = Column(DateTime, default=func.now(), onupdate=func.now()) servers = relationship("Server", back_populates="institution") sr_requests = relationship("SRRequest", back_populates="institution") contacts = relationship("InstContact", back_populates="institution", cascade="all, delete-orphan") class InstContact(Base): """기관 담당자 연락처.""" __tablename__ = "tb_inst_contact" id = Column(Integer, primary_key=True, index=True) inst_id = Column(Integer, ForeignKey("tb_inst_meta.id"), nullable=False) contact_name = Column(String(50), nullable=False) dept = Column(String(100)) position = Column(String(50)) role = Column(String(30), nullable=False, default="ENGINEER") # MANAGER | ENGINEER | PM | SECURITY | HELPDESK email = Column(String(100)) phone = Column(String(20)) mobile = Column(String(20)) is_primary = Column(Boolean, default=False) is_active = Column(Boolean, default=True) note = Column(Text) created_at = Column(DateTime, default=func.now()) institution = relationship("Institution", back_populates="contacts") class Server(Base): __tablename__ = "tb_server_info" id = Column(Integer, primary_key=True, index=True) inst_id = Column(Integer, ForeignKey("tb_inst_meta.id"), nullable=False) server_name = Column(String(100), nullable=False) server_role = Column(String(20)) # WEB / WAS / DB / ESB / BATCH os_type = Column(String(50)) os_version = Column(String(50)) ip_addr = Column(String(45)) # NOT exposed in API responses ssh_user = Column(String(50)) # NOT exposed os_pw_enc = Column(Text) # AES-256 encrypted, NEVER exposed port = Column(Integer, default=22) ssh_method = Column(String(20), default="PASSWORD") # PASSWORD | KEY | KEY_WITH_PASS ssh_key_path = Column(String(255)) # SSH 키 경로 (관제 PC 상의 경로) # WEB 정보 web_server = Column(String(50)) # Apache | Nginx | IIS | WebtoB web_version = Column(String(30)) web_port = Column(Integer) web_conf_path = Column(String(255)) # WAS 정보 was_type = Column(String(50)) # Tomcat | JBoss | WebLogic | JEUS | WebSphere was_version = Column(String(30)) was_port = Column(Integer) was_home = Column(String(255)) was_deploy_path = Column(String(255)) jdk_version = Column(String(20)) heap_min = Column(String(20)) # e.g. 512m heap_max = Column(String(20)) # e.g. 2048m # DB 정보 db_type = Column(String(30)) # PostgreSQL | Oracle | MySQL | MSSQL | Tibero db_version = Column(String(30)) db_port = Column(Integer) db_name = Column(String(100)) db_schema_name = Column(String(100)) # 유지보수 정보 cpu_cores = Column(Integer) memory_gb = Column(Integer) disk_total_gb = Column(Integer) maintenance_contact = Column(String(50)) # 담당자 사번 install_date = Column(Date) eol_date = Column(Date) ssl_cert_path = Column(String(255)) ssl_expire_date = Column(Date) backup_path = Column(String(255)) note = Column(Text) is_active = Column(Boolean, default=True) last_check = Column(DateTime) created_at = Column(DateTime, default=func.now()) updated_at = Column(DateTime, default=func.now(), onupdate=func.now()) institution = relationship("Institution", back_populates="servers") class SRRequest(Base): __tablename__ = "tb_sr_request" id = Column(Integer, primary_key=True, index=True) sr_id = Column(String(30), unique=True, nullable=False, index=True) inst_id = Column(Integer, ForeignKey("tb_inst_meta.id")) sr_type = Column(String(20), default=SRType.OTHER) title = Column(String(200), nullable=False) description = Column(Text) status = Column(String(30), default=SRStatus.RECEIVED) priority = Column(String(20), default=Priority.MEDIUM) requested_by = Column(String(100)) assigned_to = Column(String(100)) target_server = Column(String(100)) created_at = Column(DateTime, default=func.now()) updated_at = Column(DateTime, default=func.now(), onupdate=func.now()) # ── A-2: SLA 타이머 & 자동 에스컬레이션 ───────────────────────────────── sla_deadline = Column(DateTime, nullable=True) # SLA 마감 시각 sla_breached = Column(Boolean, default=False) # 마감 초과 여부 escalated_at = Column(DateTime, nullable=True) # 에스컬레이션 발생 시각 escalated_to = Column(String(100), nullable=True) # 에스컬레이션 수신자 # ── G-7: AI 자동 분류 제안 ─────────────────────────────────────────────── ai_suggestion = Column(Text, nullable=True) # JSON: AI 분류 결과 # ── G-9: Jira 연동 ────────────────────────────────────────────────────── jira_key = Column(String(30), nullable=True) # Jira 이슈 키 (GUARDIA-42) institution = relationship("Institution", back_populates="sr_requests") approvals = relationship("ApprovalFlow", back_populates="sr_request") audit_logs = relationship("AuditLog", back_populates="sr_request") class ApprovalFlow(Base): __tablename__ = "tb_approval_flow" id = Column(Integer, primary_key=True, index=True) sr_id = Column(String(30), ForeignKey("tb_sr_request.sr_id"), nullable=False) approver = Column(String(100), nullable=False) result = Column(String(20), default=ApprovalResult.PENDING) comment = Column(Text) decided_at = Column(DateTime) created_at = Column(DateTime, default=func.now()) # ── G-11: 다중 승인 체계 ───────────────────────────────────────────────── delegate_to = Column(Integer, ForeignKey("tb_user.id"), nullable=True) delegate_until = Column(DateTime, nullable=True) deadline_at = Column(DateTime, nullable=True) signature = Column(Text, nullable=True) # 전자서명 해시 sr_request = relationship("SRRequest", back_populates="approvals") class AuditLog(Base): __tablename__ = "tb_audit_log" id = Column(Integer, primary_key=True, index=True) sr_id = Column(String(30), ForeignKey("tb_sr_request.sr_id"), nullable=True) actor = Column(String(100)) action = Column(String(100), nullable=False) detail = Column(Text) prev_hash = Column(String(64)) log_hash = Column(String(64)) created_at = Column(DateTime, default=func.now()) # ── D-5: 불변 감사 로그 확장 필드 ───────────────────────────────────────── entity_type = Column(String(50), nullable=True) # SR|RFC|PAM|USER|LDAP|VULN 등 entity_id = Column(String(50), nullable=True) # 대상 엔티티 ID ip_addr_hash = Column(String(64), nullable=True) # 클라이언트 IP SHA-256 (원본 저장 금지) severity = Column(String(20), default="INFO") # INFO|WARN|ERROR|CRITICAL sr_request = relationship("SRRequest", back_populates="audit_logs") class OpsTask(Base): __tablename__ = "tb_ops_task" id = Column(Integer, primary_key=True, index=True) sr_id = Column(String(30), ForeignKey("tb_sr_request.sr_id")) task_name = Column(String(200), nullable=False) task_order = Column(Integer, default=0) status = Column(String(30), default="PENDING") result_msg = Column(Text) started_at = Column(DateTime) finished_at = Column(DateTime) created_at = Column(DateTime, default=func.now()) # ── Self-Improving Learning Loop ORM Models ──────────────────────────────────── class RecurrencePattern(Base): """재발 패턴 추적 — 같은 문제가 반복될 때 자동 감지.""" __tablename__ = "tb_recurrence_pattern" id = Column(Integer, primary_key=True, index=True) pattern_hash = Column(String(16), unique=True, index=True) pattern_key = Column(String(200)) sr_type = Column(String(20), index=True) inst_id = Column(Integer, ForeignKey("tb_inst_meta.id"), nullable=True) keyword_signature = Column(Text) # 핵심 토큰 집합 (공백 구분) tech_keywords = Column(String(200)) # 기술 키워드 (콤마 구분) occurrence_count = Column(Integer, default=1) first_seen_at = Column(DateTime, default=func.now()) last_seen_at = Column(DateTime, default=func.now()) sr_ids = Column(JSON) # 연관 SR ID 목록 escalated = Column(Boolean, default=False) problem_id = Column(String(30), nullable=True) created_at = Column(DateTime, default=func.now()) updated_at = Column(DateTime, default=func.now(), onupdate=func.now()) class SolutionFeedback(Base): """KB 솔루션 효과 추적 — 해결책이 실제로 효과 있었는지 피드백.""" __tablename__ = "tb_solution_feedback" id = Column(Integer, primary_key=True, index=True) sr_id = Column(String(30), ForeignKey("tb_sr_request.sr_id"), index=True) kb_id = Column(Integer, nullable=True) # KBDocument.id (FK 미설정 — 유연성) kb_doc_id = Column(String(30), nullable=True, index=True) applied_at = Column(DateTime, default=func.now()) resolved = Column(Boolean, default=True) recurred_within_days = Column(Integer, nullable=True) effectiveness_score = Column(Integer, default=0) # +1 성공, -1 재발 feedback_note = Column(Text, nullable=True) checked_at = Column(DateTime, nullable=True) created_at = Column(DateTime, default=func.now()) class AdaptiveThreshold(Base): """적응형 임계값 — 이상 탐지 오탐/누락 기반 자동 보정.""" __tablename__ = "tb_adaptive_threshold" id = Column(Integer, primary_key=True, index=True) source = Column(String(200), index=True) metric_type = Column(String(50), index=True) base_threshold = Column(Float, nullable=False) adapted_threshold = Column(Float, nullable=False) true_positive = Column(Integer, default=0) false_positive = Column(Integer, default=0) missed_count = Column(Integer, default=0) last_adapted_at = Column(DateTime, nullable=True) adaptation_count = Column(Integer, default=0) created_at = Column(DateTime, default=func.now()) updated_at = Column(DateTime, default=func.now(), onupdate=func.now()) class LessonLearned(Base): """검증된 교훈 — 효과 점수 10 이상 KB 또는 패턴 마이닝으로 자동 승격.""" __tablename__ = "tb_lesson_learned" id = Column(Integer, primary_key=True, index=True) lesson_id = Column(String(40), unique=True, index=True) title = Column(String(200), nullable=False) category = Column(String(50)) problem_pattern = Column(Text) root_cause = Column(Text) effective_solution = Column(Text) prevention = Column(Text) confidence_score = Column(Integer, default=0) source_kb_ids = Column(JSON) source_sr_ids = Column(JSON) promoted_from_kb_id = Column(Integer, nullable=True) is_verified = Column(Boolean, default=False) created_at = Column(DateTime, default=func.now()) updated_at = Column(DateTime, default=func.now(), onupdate=func.now()) # ── Pydantic Schemas ─────────────────────────────────────────────────────────── class InstitutionOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int inst_code: str inst_name: str org_type: Optional[str] contact_pm: Optional[str] address: Optional[str] region: Optional[str] phone: Optional[str] contract_start: Optional[date] contract_end: Optional[date] sla_hours: int note: Optional[str] is_active: bool created_at: datetime updated_at: Optional[datetime] class InstitutionCreate(BaseModel): inst_code: str inst_name: str org_type: Optional[str] = None contact_pm: Optional[str] = None address: Optional[str] = None region: Optional[str] = None phone: Optional[str] = None contract_start: Optional[date] = None contract_end: Optional[date] = None sla_hours: int = 4 note: Optional[str] = None class InstitutionUpdate(BaseModel): inst_name: Optional[str] = None org_type: Optional[str] = None contact_pm: Optional[str] = None address: Optional[str] = None region: Optional[str] = None phone: Optional[str] = None contract_start: Optional[date] = None contract_end: Optional[date] = None sla_hours: Optional[int] = None note: Optional[str] = None is_active: Optional[bool] = None class InstContactOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int inst_id: int contact_name: str dept: Optional[str] position: Optional[str] role: str email: Optional[str] phone: Optional[str] mobile: Optional[str] is_primary: bool is_active: bool note: Optional[str] created_at: datetime class InstContactCreate(BaseModel): contact_name: str dept: Optional[str] = None position: Optional[str] = None role: str = "ENGINEER" email: Optional[str] = None phone: Optional[str] = None mobile: Optional[str] = None is_primary: bool = False note: Optional[str] = None class InstContactUpdate(BaseModel): contact_name: Optional[str] = None dept: Optional[str] = None position: Optional[str] = None role: Optional[str] = None email: Optional[str] = None phone: Optional[str] = None mobile: Optional[str] = None is_primary: Optional[bool] = None is_active: Optional[bool] = None note: Optional[str] = None class ServerOut(BaseModel): """Server info — sensitive fields (ip_addr, ssh_user, os_pw_enc) intentionally omitted.""" model_config = ConfigDict(from_attributes=True) id: int server_name: str server_role: Optional[str] os_type: Optional[str] os_version: Optional[str] ssh_method: Optional[str] web_server: Optional[str] web_version: Optional[str] web_port: Optional[int] web_conf_path: Optional[str] was_type: Optional[str] was_version: Optional[str] was_port: Optional[int] was_home: Optional[str] was_deploy_path: Optional[str] jdk_version: Optional[str] heap_min: Optional[str] heap_max: Optional[str] db_type: Optional[str] db_version: Optional[str] db_port: Optional[int] db_name: Optional[str] db_schema_name: Optional[str] cpu_cores: Optional[int] memory_gb: Optional[int] disk_total_gb: Optional[int] maintenance_contact: Optional[str] install_date: Optional[date] eol_date: Optional[date] ssl_expire_date: Optional[date] ssl_cert_path: Optional[str] backup_path: Optional[str] note: Optional[str] is_active: bool last_check: Optional[datetime] created_at: datetime updated_at: Optional[datetime] class ServerUpdate(BaseModel): server_name: Optional[str] = None server_role: Optional[str] = None os_type: Optional[str] = None os_version: Optional[str] = None ssh_method: Optional[str] = None web_server: Optional[str] = None web_version: Optional[str] = None web_port: Optional[int] = None web_conf_path: Optional[str] = None was_type: Optional[str] = None was_version: Optional[str] = None was_port: Optional[int] = None was_home: Optional[str] = None was_deploy_path: Optional[str] = None jdk_version: Optional[str] = None heap_min: Optional[str] = None heap_max: Optional[str] = None db_type: Optional[str] = None db_version: Optional[str] = None db_port: Optional[int] = None db_name: Optional[str] = None db_schema_name: Optional[str] = None cpu_cores: Optional[int] = None memory_gb: Optional[int] = None disk_total_gb: Optional[int] = None maintenance_contact: Optional[str] = None install_date: Optional[date] = None eol_date: Optional[date] = None ssl_expire_date: Optional[date] = None ssl_cert_path: Optional[str] = None backup_path: Optional[str] = None note: Optional[str] = None is_active: Optional[bool] = None class SRCreate(BaseModel): title: str description: Optional[str] = None sr_type: SRType = SRType.OTHER priority: Priority = Priority.MEDIUM requested_by: str assigned_to: Optional[str] = None target_server: Optional[str] = None inst_code: Optional[str] = None class SROut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int sr_id: str sr_type: str title: str description: Optional[str] status: str priority: str requested_by: str assigned_to: Optional[str] target_server: Optional[str] created_at: datetime updated_at: datetime # A-2 SLA 필드 sla_deadline: Optional[datetime] = None sla_breached: bool = False escalated_at: Optional[datetime] = None escalated_to: Optional[str] = None class SRStatusUpdate(BaseModel): status: SRStatus actor: str comment: Optional[str] = None class ApprovalCreate(BaseModel): approver: str result: ApprovalResult comment: Optional[str] = None class ApprovalOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int sr_id: str approver: str result: str comment: Optional[str] decided_at: Optional[datetime] created_at: datetime class AuditLogOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int sr_id: Optional[str] actor: Optional[str] action: str detail: Optional[str] log_hash: Optional[str] prev_hash: Optional[str] entity_type: Optional[str] entity_id: Optional[str] severity: Optional[str] created_at: datetime # ── Audit hash helper ────────────────────────────────────────────────────────── def compute_log_hash(prev_hash: Optional[str], actor: str, action: str, detail: str, ts: str) -> str: payload = json.dumps( {"prev": prev_hash or "", "actor": actor, "action": action, "detail": detail, "ts": ts}, ensure_ascii=False, sort_keys=True ) return hashlib.sha256(payload.encode()).hexdigest() # ── WorkLog ORM ──────────────────────────────────────────────────────────────── # ── User / Auth Models ───────────────────────────────────────────────────────── class UserRole(str, Enum): ADMIN = "ADMIN" ENGINEER = "ENGINEER" PM = "PM" DEPLOY_ENGINEER = "DEPLOY_ENGINEER" # Priority 4 신규: 배포 전담 엔지니어 CUSTOMER = "CUSTOMER" class User(Base): __tablename__ = "tb_user" id = Column(Integer, primary_key=True, index=True) username = Column(String(50), unique=True, nullable=False, index=True) display_name = Column(String(100)) role = Column(String(20), nullable=False, default=UserRole.ENGINEER) hashed_pw = Column(String(200), nullable=False) must_change_pw = Column(Boolean, default=True) inst_code = Column(String(20), nullable=True) # CUSTOMER 역할의 소속 기관 email = Column(String(100), nullable=True) phone = Column(String(20), nullable=True) dept = Column(String(100), nullable=True) position = Column(String(50), nullable=True) is_active = Column(Boolean, default=True) created_at = Column(DateTime, default=func.now()) last_login_at = Column(DateTime, nullable=True) # ── A-5: MFA (TOTP) ────────────────────────────────────────────────────── mfa_enabled = Column(Boolean, default=False) # MFA 활성 여부 totp_secret_enc = Column(String(200), nullable=True) # AES-GCM 암호화된 TOTP 시크릿 # ── D-1: LDAP/AD 연동 ───────────────────────────────────────────────────── auth_type = Column(String(20), default="LOCAL") # LOCAL | LDAP department = Column(String(100), nullable=True) # LDAP department # -- Login security: account lockout failed_login_count = Column(Integer, default=0) locked_until = Column(DateTime, nullable=True) class UserOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int username: str display_name: Optional[str] role: str must_change_pw: bool inst_code: Optional[str] is_active: bool mfa_enabled: bool = False class WorkActionType(str, Enum): CMDB_CHECK = "CMDB_CHECK" # 자산 확인 SSH_CONNECT = "SSH_CONNECT" # SSH 접속 SSH_EXEC = "SSH_EXEC" # 명령 실행 SOURCE_MOD = "SOURCE_MOD" # 소스/파일 배포 HEALTH_CHECK = "HEALTH_CHECK" # 헬스체크 RESULT = "RESULT" # 결과 기록 COMPLETE = "COMPLETE" # 완료 처리 class WorkLog(Base): __tablename__ = "tb_work_log" id = Column(Integer, primary_key=True, index=True) sr_id = Column(String(30), ForeignKey("tb_sr_request.sr_id"), nullable=False) engineer = Column(String(100), default="GUARDiA-AI") action_type = Column(String(30), nullable=False) content = Column(Text) # 수행 내용 result = Column(Text) # 결과 출력 is_success = Column(Boolean, default=True) created_at = Column(DateTime, default=func.now()) class Rating(Base): __tablename__ = "tb_rating" id = Column(Integer, primary_key=True, index=True) sr_id = Column(String(30), ForeignKey("tb_sr_request.sr_id"), unique=True) customer = Column(String(100)) stars = Column(Integer) # 1~5 comment = Column(Text) created_at = Column(DateTime, default=func.now()) class LicenseRecord(Base): """GUARDiA 라이선스 등록 이력 테이블.""" __tablename__ = "tb_license" id = Column(Integer, primary_key=True, index=True) license_key = Column(Text, nullable=False) license_id = Column(String(50), unique=True, nullable=False, index=True) edition = Column(String(20), nullable=False) customer = Column(String(200), nullable=False) issued_at = Column(DateTime, nullable=True) expires_at = Column(DateTime, nullable=False) limits = Column(JSON, nullable=True) is_active = Column(Boolean, default=True) is_trial = Column(Boolean, default=False) activated_by = Column(String(50), nullable=True) activated_at = Column(DateTime, default=func.now()) # ── G-8: 보안 패치 추적 ────────────────────────────────────────────────────── class VulnPatchRecord(Base): """취약점 패치 이력 테이블.""" __tablename__ = "tb_vuln_patch" id = Column(Integer, primary_key=True, index=True) server_id = Column(Integer, ForeignKey("tb_server_info.id"), nullable=True) scan_id = Column(String(50), nullable=True) cve_id = Column(String(30), nullable=False, index=True) cvss_score = Column(Float, nullable=True) severity = Column(String(20), nullable=True) # CRITICAL|HIGH|MEDIUM|LOW status = Column(String(20), default="OPEN") # OPEN|PATCHED|WONTFIX|MITIGATED patch_note = Column(Text, nullable=True) patched_at = Column(DateTime, nullable=True) patched_by = Column(String(50), nullable=True) created_at = Column(DateTime, default=func.now()) updated_at = Column(DateTime, default=func.now(), onupdate=func.now()) # ── G-10: PWA Push 구독 ────────────────────────────────────────────────────── class PushSubscription(Base): """PWA 푸시 구독 정보 테이블.""" __tablename__ = "tb_push_subscription" id = Column(Integer, primary_key=True, index=True) user_id = Column(Integer, ForeignKey("tb_user.id"), nullable=False, index=True) endpoint = Column(Text, nullable=False, unique=True) p256dh = Column(Text, nullable=False) auth = Column(Text, nullable=False) user_agent = Column(String(200), nullable=True) created_at = Column(DateTime, default=func.now()) # ── Pydantic Schemas (new) ───────────────────────────────────────────────────── class WorkLogOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int sr_id: str engineer: Optional[str] action_type: str content: Optional[str] result: Optional[str] is_success: bool created_at: datetime class WorkStepIn(BaseModel): engineer: str = "GUARDiA-AI" action_type: WorkActionType content: str result: Optional[str] = None is_success: bool = True class RatingCreate(BaseModel): customer: str stars: int # 1~5 comment: Optional[str] = None class RatingOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int sr_id: str customer: Optional[str] stars: int comment: Optional[str] created_at: datetime # ── Engineer Profile ──────────────────────────────────────────────────────────── class EngineerProfile(Base): """엔지니어 스킬 프로필 — 자동 배정에 사용.""" __tablename__ = "tb_engineer_profile" id = Column(Integer, primary_key=True, index=True) username = Column(String(50), unique=True, nullable=False, index=True) display_name = Column(String(100)) skill_types = Column(String(200)) # comma-sep SRType, e.g. "DEPLOY,RESTART" inst_affinity = Column(String(200)) # comma-sep inst_code, e.g. "MOF,MOIS" max_workload = Column(Integer, default=5) is_available = Column(Boolean, default=True) created_at = Column(DateTime, default=func.now()) class EngineerProfileOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int username: str display_name: Optional[str] skill_types: Optional[str] inst_affinity: Optional[str] max_workload: int is_available: bool # ── Knowledge Base ───────────────────────────────────────────────────────────── class KBDocument(Base): """기술 지식 문서 — 장애 증상·원인·해결 방법.""" __tablename__ = "tb_kb_document" id = Column(Integer, primary_key=True, index=True) doc_id = Column(String(30), unique=True, nullable=False, index=True) category = Column(String(50)) title = Column(String(200), nullable=False) symptoms = Column(Text) cause = Column(Text) solution = Column(Text) commands = Column(Text) tags = Column(String(300)) sr_type = Column(String(20)) quality_score = Column(Integer, default=0) published = Column(Boolean, default=False) source_sr_id = Column(String(20), nullable=True, index=True) author = Column(String(100), nullable=True) updated_at = Column(DateTime, default=func.now(), onupdate=func.now()) created_at = Column(DateTime, default=func.now()) class KBDocumentOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int doc_id: str category: Optional[str] title: str symptoms: Optional[str] cause: Optional[str] solution: Optional[str] commands: Optional[str] tags: Optional[str] sr_type: Optional[str] class KBSearchResult(BaseModel): doc: KBDocumentOut score: float matched_keywords: List[str] # ── Shell Script Library ──────────────────────────────────────────────────────── class ShellScript(Base): """쉘 스크립트 라이브러리 — SM·정기점검·수시점검 등 모든 운영 스크립트 관리.""" __tablename__ = "tb_shell_script" id = Column(Integer, primary_key=True, index=True) script_name = Column(String(200), nullable=False) category = Column(String(30), nullable=False) # SM|REGULAR|ADHOC|DEPLOY|SECURITY|MONITORING sub_category = Column(String(50)) target_layer = Column(String(20), default="ALL") # WEB|WAS|DB|ESB|ALL os_type = Column(String(20), default="LINUX") # LINUX|WINDOWS|ALL description = Column(Text, nullable=False) script_body = Column(Text, nullable=False) parameters = Column(Text) # JSON string: [{"name":"TARGET","desc":"...","required":true}] sample_output = Column(Text) is_dangerous = Column(Boolean, default=False) requires_approval = Column(Boolean, default=False) author = Column(String(50)) version = Column(String(20), default="1.0.0") tags = Column(String(300)) # comma-separated inst_id = Column(Integer, ForeignKey("tb_inst_meta.id"), nullable=True) # NULL = 공용 스크립트 is_active = Column(Boolean, default=True) use_count = Column(Integer, default=0) created_at = Column(DateTime, default=func.now()) updated_at = Column(DateTime, default=func.now(), onupdate=func.now()) class ShellScriptOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int script_name: str category: str sub_category: Optional[str] target_layer: str os_type: str description: str script_body: str parameters: Optional[str] sample_output: Optional[str] is_dangerous: bool requires_approval: bool author: Optional[str] version: str tags: Optional[str] inst_id: Optional[int] is_active: bool use_count: int created_at: datetime updated_at: Optional[datetime] class ShellScriptCreate(BaseModel): script_name: str category: str sub_category: Optional[str] = None target_layer: str = "ALL" os_type: str = "LINUX" description: str script_body: str parameters: Optional[str] = None # JSON string sample_output: Optional[str] = None is_dangerous: bool = False requires_approval: bool = False author: Optional[str] = None version: str = "1.0.0" tags: Optional[str] = None inst_id: Optional[int] = None class ShellScriptUpdate(BaseModel): script_name: Optional[str] = None category: Optional[str] = None sub_category: Optional[str] = None target_layer: Optional[str] = None os_type: Optional[str] = None description: Optional[str] = None script_body: Optional[str] = None parameters: Optional[str] = None sample_output: Optional[str] = None is_dangerous: Optional[bool] = None requires_approval: Optional[bool] = None author: Optional[str] = None version: Optional[str] = None tags: Optional[str] = None is_active: Optional[bool] = None # ── Work Timetable ────────────────────────────────────────────────────────────── class WorkTimetable(Base): """작업 타임테이블 — SM·정기점검·PM·SR 기반 모든 작업 이력.""" __tablename__ = "tb_work_timetable" id = Column(Integer, primary_key=True, index=True) work_type = Column(String(30), nullable=False) # REGULAR_CHECK | PM | SR | ADHOC | DEPLOY | EMERGENCY title = Column(String(200), nullable=False) inst_id = Column(Integer, ForeignKey("tb_inst_meta.id"), nullable=True) server_id = Column(Integer, ForeignKey("tb_server_info.id"), nullable=True) sr_id = Column(String(30), nullable=True) script_id = Column(Integer, ForeignKey("tb_shell_script.id"), nullable=True) scheduled_at = Column(DateTime, nullable=False) started_at = Column(DateTime, nullable=True) completed_at = Column(DateTime, nullable=True) content = Column(Text, nullable=False) command_or_shell = Column(Text, nullable=True) result = Column(Text, nullable=True) result_status = Column(String(20), default="PENDING") # PENDING | SUCCESS | FAILED | PARTIAL | CANCELLED assignee = Column(String(50), nullable=True) reviewer = Column(String(50), nullable=True) note = Column(Text, nullable=True) created_by = Column(String(50), nullable=True) created_at = Column(DateTime, default=func.now()) updated_at = Column(DateTime, default=func.now(), onupdate=func.now()) institution = relationship("Institution", foreign_keys=[inst_id]) server = relationship("Server", foreign_keys=[server_id]) script = relationship("ShellScript", foreign_keys=[script_id]) class WorkTimetableOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int work_type: str title: str inst_id: Optional[int] server_id: Optional[int] sr_id: Optional[str] script_id: Optional[int] scheduled_at: datetime started_at: Optional[datetime] completed_at: Optional[datetime] content: str command_or_shell: Optional[str] result: Optional[str] result_status: str assignee: Optional[str] reviewer: Optional[str] note: Optional[str] created_by: Optional[str] created_at: datetime updated_at: Optional[datetime] class WorkTimetableCreate(BaseModel): work_type: str title: str inst_id: Optional[int] = None server_id: Optional[int] = None sr_id: Optional[str] = None script_id: Optional[int] = None scheduled_at: datetime started_at: Optional[datetime] = None completed_at: Optional[datetime] = None content: str command_or_shell: Optional[str] = None result: Optional[str] = None result_status: str = "PENDING" assignee: Optional[str] = None reviewer: Optional[str] = None note: Optional[str] = None class WorkTimetableUpdate(BaseModel): work_type: Optional[str] = None title: Optional[str] = None inst_id: Optional[int] = None server_id: Optional[int] = None sr_id: Optional[str] = None script_id: Optional[int] = None scheduled_at: Optional[datetime] = None started_at: Optional[datetime] = None completed_at: Optional[datetime] = None content: Optional[str] = None command_or_shell: Optional[str] = None result: Optional[str] = None result_status: Optional[str] = None assignee: Optional[str] = None reviewer: Optional[str] = None note: Optional[str] = None # ── SR Attachment ─────────────────────────────────────────────────────────────── class SRAttachment(Base): """SR 첨부 파일 — 요청 시 업로드된 파일 메타데이터.""" __tablename__ = "tb_sr_attachment" id = Column(Integer, primary_key=True, index=True) sr_id = Column(String(30), ForeignKey("tb_sr_request.sr_id"), nullable=False, index=True) original_name = Column(String(255), nullable=False) # 원본 파일명 stored_name = Column(String(255), nullable=False) # 저장 파일명 (충돌 방지) file_path = Column(String(512), nullable=False) # 서버 내부 경로 — API 응답에 노출 금지 file_size = Column(Integer, nullable=False) # bytes content_type = Column(String(100)) uploaded_by = Column(String(100), nullable=False) created_at = Column(DateTime, default=func.now()) sr_request = relationship("SRRequest", foreign_keys=[sr_id]) class SRAttachmentOut(BaseModel): """첨부파일 응답 스키마 — file_path 노출 금지.""" model_config = ConfigDict(from_attributes=True) id: int sr_id: str original_name: str file_size: int content_type: Optional[str] uploaded_by: str created_at: datetime # ── Notification Log ──────────────────────────────────────────────────────────── class NotificationLog(Base): """알림 발송 이력 — 이메일·메신저 전송 결과 추적.""" __tablename__ = "tb_notification_log" id = Column(Integer, primary_key=True, index=True) sr_id = Column(String(30), nullable=True, index=True) channel = Column(String(20), nullable=False) # EMAIL | MESSENGER recipient = Column(String(200)) # 수신자 (이메일 주소 또는 채널명) subject = Column(String(300)) status = Column(String(20), nullable=False) # SENT | FAILED | SKIPPED error_msg = Column(Text, nullable=True) sent_at = Column(DateTime, default=func.now()) class NotificationLogOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int sr_id: Optional[str] channel: str recipient: Optional[str] subject: Optional[str] status: str error_msg: Optional[str] sent_at: datetime # ── Project Source (바이브 코딩 / 배포 파이프라인) ────────────────────────────── class Project(Base): """소스 프로젝트 등록 — 바이브 코딩·배포 파이프라인 연결.""" __tablename__ = "tb_project" id = Column(Integer, primary_key=True, index=True) project_name = Column(String(100), unique=True, nullable=False, index=True) description = Column(Text, nullable=True) source_path = Column(String(512), nullable=True) # 로컬 개발 PC 소스 절대 경로 repo_url = Column(String(512), nullable=True) # Git 원격 주소 branch = Column(String(100), default="main") build_cmd = Column(Text, nullable=True) # 예: mvn clean package -DskipTests test_cmd = Column(Text, nullable=True) # 예: mvn test deploy_server_id = Column(Integer, ForeignKey("tb_server_info.id"), nullable=True) deploy_path = Column(String(512), nullable=True) # 서버 내 배포 경로 was_restart_cmd = Column(Text, nullable=True) # 예: systemctl restart tomcat9 health_check_url = Column(String(512), nullable=True) # 예: http://localhost:8080/health jenkins_job_name = Column(String(200), nullable=True) # Jenkins Job 이름 (미설정 시 "{project_name}-deploy" 사용) sr_type_hint = Column(String(20), nullable=True) # 연관 SR 유형 힌트 # ── B-3: 코드 리뷰 연동 필드 ────────────────────────────────────────────── project_dir = Column(String(200), nullable=True) # C:\GUARDiA\projects\ 하위 폴더명 tech_stack = Column(String(50), nullable=True) # java/python/php/javascript/mixed last_review_at = Column(DateTime, nullable=True) last_review_score= Column(Integer, nullable=True) # 0-100 종합 점수 created_by = Column(String(100), nullable=True) created_at = Column(DateTime, default=func.now()) updated_at = Column(DateTime, default=func.now(), onupdate=func.now()) deploy_server = relationship("Server", foreign_keys=[deploy_server_id]) class ProjectOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int project_name: str description: Optional[str] source_path: Optional[str] repo_url: Optional[str] branch: str build_cmd: Optional[str] test_cmd: Optional[str] deploy_server_id: Optional[int] deploy_path: Optional[str] was_restart_cmd: Optional[str] health_check_url: Optional[str] jenkins_job_name: Optional[str] sr_type_hint: Optional[str] project_dir: Optional[str] tech_stack: Optional[str] last_review_at: Optional[datetime] last_review_score: Optional[int] created_by: Optional[str] created_at: datetime updated_at: Optional[datetime] class ProjectCreate(BaseModel): project_name: str description: Optional[str] = None source_path: Optional[str] = None repo_url: Optional[str] = None branch: str = "main" build_cmd: Optional[str] = None test_cmd: Optional[str] = None deploy_server_id: Optional[int] = None deploy_path: Optional[str] = None was_restart_cmd: Optional[str] = None health_check_url: Optional[str] = None jenkins_job_name: Optional[str] = None sr_type_hint: Optional[str] = None project_dir: Optional[str] = None tech_stack: Optional[str] = None class ProjectUpdate(BaseModel): project_name: Optional[str] = None description: Optional[str] = None source_path: Optional[str] = None repo_url: Optional[str] = None branch: Optional[str] = None build_cmd: Optional[str] = None test_cmd: Optional[str] = None deploy_server_id: Optional[int] = None deploy_path: Optional[str] = None was_restart_cmd: Optional[str] = None health_check_url: Optional[str] = None jenkins_job_name: Optional[str] = None sr_type_hint: Optional[str] = None # ── Vibe Session (Claude CLI 바이브 코딩 세션) ────────────────────────────────── class VibeSessionStatus(str, Enum): PENDING = "PENDING" CODING = "CODING" BUILDING = "BUILDING" TESTING = "TESTING" DEPLOYING = "DEPLOYING" COMPLETED = "COMPLETED" FAILED = "FAILED" CANCELLED = "CANCELLED" class VibeSession(Base): """Claude CLI 바이브 코딩 세션 — SR → 코딩 → 빌드 → 배포 파이프라인 추적.""" __tablename__ = "tb_vibe_session" id = Column(Integer, primary_key=True, index=True) sr_id = Column(String(30), ForeignKey("tb_sr_request.sr_id"), nullable=True, index=True) project_id = Column(Integer, ForeignKey("tb_project.id"), nullable=True) claude_session_id = Column(String(128), nullable=True, index=True) # Claude CLI 세션 ID (claude --session-id 또는 SDK 세션 ID) workspace_path = Column(String(512), nullable=True) # 실제 작업 디렉터리 status = Column(String(20), nullable=False, default=VibeSessionStatus.PENDING) started_by = Column(String(100), nullable=True) started_at = Column(DateTime, default=func.now()) coded_at = Column(DateTime, nullable=True) # 코딩 완료 시각 built_at = Column(DateTime, nullable=True) # 빌드 완료 시각 tested_at = Column(DateTime, nullable=True) # 테스트 완료 시각 deployed_at = Column(DateTime, nullable=True) # 배포 완료 시각 build_log = Column(Text, nullable=True) # 빌드 로그 요약 test_result = Column(Text, nullable=True) # 테스트 결과 요약 deploy_log = Column(Text, nullable=True) # 배포 로그 요약 error_msg = Column(Text, nullable=True) # 실패 시 오류 요약 sr_request = relationship("SRRequest", foreign_keys=[sr_id]) project = relationship("Project", foreign_keys=[project_id]) class VibeSessionOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int sr_id: Optional[str] project_id: Optional[int] claude_session_id: Optional[str] workspace_path: Optional[str] status: str started_by: Optional[str] started_at: datetime coded_at: Optional[datetime] built_at: Optional[datetime] tested_at: Optional[datetime] deployed_at: Optional[datetime] build_log: Optional[str] test_result: Optional[str] deploy_log: Optional[str] error_msg: Optional[str] class VibeSessionCreate(BaseModel): sr_id: Optional[str] = None project_id: Optional[int] = None claude_session_id: Optional[str] = None workspace_path: Optional[str] = None started_by: Optional[str] = None class VibeSessionUpdate(BaseModel): status: Optional[VibeSessionStatus] = None claude_session_id: Optional[str] = None workspace_path: Optional[str] = None build_log: Optional[str] = None test_result: Optional[str] = None deploy_log: Optional[str] = None error_msg: Optional[str] = None # ═══════════════════════════════════════════════════════════════════════════════ # ── SSL 인증서 이력 (tb_ssl_history) ────────────────────────────────────────── # ═══════════════════════════════════════════════════════════════════════════════ class SslAlertLevel(str, Enum): OK = "OK" # 30일 초과 WARN = "WARN" # 7~30일 URGENT = "URGENT" # 1~7일 EXPIRED = "EXPIRED" # 만료됨 class SslHistory(Base): """SSL 인증서 갱신 이력 — Server.ssl_expire_date 변경 추적.""" __tablename__ = "tb_ssl_history" id = Column(Integer, primary_key=True, index=True) server_id = Column(Integer, ForeignKey("tb_server_info.id"), nullable=False, index=True) cert_domain = Column(String(255), nullable=True) # CN / SAN 주도메인 old_expire = Column(Date, nullable=True) # 이전 만료일 new_expire = Column(Date, nullable=True) # 갱신 후 만료일 issuer = Column(String(255), nullable=True) # 발급기관 cert_path = Column(String(512), nullable=True) # 서버 내 인증서 경로 renewed_by = Column(String(100), nullable=True) # 갱신 처리자 sr_id = Column(String(30), ForeignKey("tb_sr_request.sr_id"), nullable=True) note = Column(Text, nullable=True) created_at = Column(DateTime, default=func.now()) server = relationship("Server", foreign_keys=[server_id]) class SslHistoryOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int server_id: int cert_domain: Optional[str] old_expire: Optional[date] new_expire: Optional[date] issuer: Optional[str] cert_path: Optional[str] renewed_by: Optional[str] sr_id: Optional[str] note: Optional[str] created_at: datetime class SslHistoryCreate(BaseModel): server_id: int cert_domain: Optional[str] = None old_expire: Optional[date] = None new_expire: Optional[date] = None issuer: Optional[str] = None cert_path: Optional[str] = None renewed_by: Optional[str] = None sr_id: Optional[str] = None note: Optional[str] = None class SslExpiryInfo(BaseModel): """SSL 만료 현황 API 응답 — ServerOut + 만료 정보 조합.""" model_config = ConfigDict(from_attributes=True) server_id: int server_name: str inst_name: Optional[str] = None ssl_cert_path: Optional[str] ssl_expire_date: Optional[date] days_left: Optional[int] alert_level: SslAlertLevel last_checked: Optional[datetime] = None # ═══════════════════════════════════════════════════════════════════════════════ # ── 정기 PM 체크리스트 (tb_pm_template / tb_pm_schedule / tb_pm_result) ─────── # ═══════════════════════════════════════════════════════════════════════════════ class PmFrequency(str, Enum): WEEKLY = "WEEKLY" BIWEEKLY = "BIWEEKLY" # 격주 MONTHLY = "MONTHLY" QUARTERLY = "QUARTERLY" # 분기 SEMIANNUAL= "SEMIANNUAL" # 반기 ANNUAL = "ANNUAL" CUSTOM = "CUSTOM" # cron 표현식 class PmItemResult(str, Enum): PASS = "PASS" FAIL = "FAIL" NA = "NA" # 해당 없음 WARNING = "WARNING" class PmTemplate(Base): """PM 체크리스트 템플릿 — 서버 유형별 기본 점검 항목.""" __tablename__ = "tb_pm_template" id = Column(Integer, primary_key=True, index=True) template_name = Column(String(100), nullable=False) server_role = Column(String(20), nullable=True) # WEB/WAS/DB/ALL 등 category = Column(String(50), nullable=True) # OS/WEB/WAS/DB/SECURITY/NETWORK item_order = Column(Integer, default=0) # 항목 순서 item_title = Column(String(200), nullable=False) # 점검 항목명 item_desc = Column(Text, nullable=True) # 점검 방법 설명 check_command = Column(Text, nullable=True) # 확인 명령어 (참고용) expected_value= Column(String(200), nullable=True) # 정상 기준값 is_mandatory = Column(Boolean, default=True) # 필수 항목 여부 is_active = Column(Boolean, default=True) created_at = Column(DateTime, default=func.now()) class PmSchedule(Base): """정기 PM 반복 스케줄 — 자동으로 WorkTimetable 생성.""" __tablename__ = "tb_pm_schedule" id = Column(Integer, primary_key=True, index=True) schedule_name = Column(String(100), nullable=False) inst_id = Column(Integer, ForeignKey("tb_inst_meta.id"), nullable=True) server_id = Column(Integer, ForeignKey("tb_server_info.id"), nullable=True) # 서버 미지정 시 기관 전체 서버 대상 frequency = Column(String(20), nullable=False, default=PmFrequency.MONTHLY) cron_expr = Column(String(50), nullable=True) # CUSTOM 시 cron 표현식 day_of_month = Column(Integer, nullable=True) # MONTHLY: 매월 N일 month_of_year = Column(Integer, nullable=True) # ANNUAL: N월 advance_days = Column(Integer, default=7) # 시작 N일 전 WorkTimetable 생성 template_ids = Column(String(200), nullable=True) # 사용할 PmTemplate ID 목록 (콤마) assignee = Column(String(100), nullable=True) reviewer = Column(String(100), nullable=True) notify_before = Column(Boolean, default=True) # D-7 사전 알림 notify_after = Column(Boolean, default=True) # 완료 후 기관 알림 is_active = Column(Boolean, default=True) last_generated = Column(DateTime, nullable=True) # 마지막 WorkTimetable 생성 시각 next_scheduled = Column(DateTime, nullable=True) # 다음 예정 일시 created_by = Column(String(100), nullable=True) created_at = Column(DateTime, default=func.now()) updated_at = Column(DateTime, default=func.now(), onupdate=func.now()) institution = relationship("Institution", foreign_keys=[inst_id]) server = relationship("Server", foreign_keys=[server_id]) class PmResult(Base): """PM 점검 실행 결과 — 항목별 체크 기록.""" __tablename__ = "tb_pm_result" id = Column(Integer, primary_key=True, index=True) timetable_id = Column(Integer, ForeignKey("tb_work_timetable.id"), nullable=False, index=True) template_id = Column(Integer, ForeignKey("tb_pm_template.id"), nullable=True) item_title = Column(String(200), nullable=False) item_desc = Column(Text, nullable=True) check_command = Column(Text, nullable=True) actual_value = Column(Text, nullable=True) # 실제 확인값 result = Column(String(20), default=PmItemResult.NA) result_note = Column(Text, nullable=True) # 비고 checked_by = Column(String(100), nullable=True) checked_at = Column(DateTime, nullable=True) created_at = Column(DateTime, default=func.now()) timetable = relationship("WorkTimetable", foreign_keys=[timetable_id]) template = relationship("PmTemplate", foreign_keys=[template_id]) # ── Pydantic schemas ───────────────────────────────────────────────────────── class PmTemplateOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int template_name: str server_role: Optional[str] category: Optional[str] item_order: int item_title: str item_desc: Optional[str] check_command: Optional[str] expected_value: Optional[str] is_mandatory: bool is_active: bool class PmTemplateCreate(BaseModel): template_name: str server_role: Optional[str] = None category: Optional[str] = None item_order: int = 0 item_title: str item_desc: Optional[str] = None check_command: Optional[str] = None expected_value: Optional[str] = None is_mandatory: bool = True class PmScheduleOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int schedule_name: str inst_id: Optional[int] server_id: Optional[int] frequency: str cron_expr: Optional[str] day_of_month: Optional[int] advance_days: int template_ids: Optional[str] assignee: Optional[str] reviewer: Optional[str] notify_before: bool notify_after: bool is_active: bool next_scheduled: Optional[datetime] created_by: Optional[str] created_at: datetime class PmScheduleCreate(BaseModel): schedule_name: str inst_id: Optional[int] = None server_id: Optional[int] = None frequency: PmFrequency = PmFrequency.MONTHLY cron_expr: Optional[str] = None day_of_month: Optional[int] = None month_of_year: Optional[int] = None advance_days: int = 7 template_ids: Optional[str] = None assignee: Optional[str] = None reviewer: Optional[str] = None notify_before: bool = True notify_after: bool = True class PmResultOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int timetable_id: int template_id: Optional[int] item_title: str item_desc: Optional[str] check_command: Optional[str] actual_value: Optional[str] result: str result_note: Optional[str] checked_by: Optional[str] checked_at: Optional[datetime] class PmResultUpdate(BaseModel): actual_value: Optional[str] = None result: PmItemResult = PmItemResult.NA result_note: Optional[str] = None # ═══════════════════════════════════════════════════════════════════════════════ # ── 장애 관리 (tb_incident) ──────────────────────────────────────────────────── # ═══════════════════════════════════════════════════════════════════════════════ class IncidentGrade(str, Enum): P1 = "P1" # 전체 서비스 중단 (Critical) P2 = "P2" # 주요 기능 장애 (High) P3 = "P3" # 일부 기능 장애 (Medium) P4 = "P4" # 성능 저하 / 경미 (Low) class IncidentStatus(str, Enum): OPEN = "OPEN" INVESTIGATING = "INVESTIGATING" MITIGATED = "MITIGATED" # 임시 조치 RESOLVED = "RESOLVED" CLOSED = "CLOSED" class Incident(Base): """장애 마스터 — P1~P4 등급, 발생~복구~RCA 전체 추적.""" __tablename__ = "tb_incident" id = Column(Integer, primary_key=True, index=True) incident_id = Column(String(30), unique=True, nullable=False, index=True) # 예: INC-20260525-001234 inst_id = Column(Integer, ForeignKey("tb_inst_meta.id"), nullable=True) title = Column(String(200), nullable=False) description = Column(Text, nullable=True) grade = Column(String(5), nullable=False, default=IncidentGrade.P3) status = Column(String(20), nullable=False, default=IncidentStatus.OPEN) affected_servers= Column(Text, nullable=True) # JSON 배열 (server_id 목록) affected_service= Column(String(200), nullable=True) occurred_at = Column(DateTime, nullable=False, default=func.now()) detected_at = Column(DateTime, nullable=True) mitigated_at = Column(DateTime, nullable=True) resolved_at = Column(DateTime, nullable=True) closed_at = Column(DateTime, nullable=True) rca = Column(Text, nullable=True) # Root Cause Analysis prevention = Column(Text, nullable=True) # 재발 방지 조치 kb_doc_id = Column(String(30), nullable=True) # 연관 KB 문서 reported_by = Column(String(100), nullable=True) assigned_to = Column(String(100), nullable=True) escalated_to = Column(String(100), nullable=True) created_at = Column(DateTime, default=func.now()) updated_at = Column(DateTime, default=func.now(), onupdate=func.now()) institution = relationship("Institution", foreign_keys=[inst_id]) class IncidentSR(Base): """장애 ↔ SR 연결 (1:N — 하나의 장애에 복수 SR 묶기).""" __tablename__ = "tb_incident_sr" id = Column(Integer, primary_key=True, index=True) incident_id = Column(String(30), ForeignKey("tb_incident.incident_id"), nullable=False) sr_id = Column(String(30), ForeignKey("tb_sr_request.sr_id"), nullable=False) created_at = Column(DateTime, default=func.now()) class IncidentOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int incident_id: str inst_id: Optional[int] title: str description: Optional[str] grade: str status: str affected_service: Optional[str] occurred_at: datetime detected_at: Optional[datetime] mitigated_at: Optional[datetime] resolved_at: Optional[datetime] rca: Optional[str] prevention: Optional[str] reported_by: Optional[str] assigned_to: Optional[str] created_at: datetime class IncidentCreate(BaseModel): inst_id: Optional[int] = None title: str description: Optional[str] = None grade: IncidentGrade = IncidentGrade.P3 affected_servers: Optional[str] = None affected_service: Optional[str] = None occurred_at: Optional[datetime] = None reported_by: Optional[str] = None assigned_to: Optional[str] = None class IncidentUpdate(BaseModel): title: Optional[str] = None description: Optional[str] = None grade: Optional[IncidentGrade] = None status: Optional[IncidentStatus] = None affected_service: Optional[str] = None mitigated_at: Optional[datetime] = None resolved_at: Optional[datetime] = None rca: Optional[str] = None prevention: Optional[str] = None kb_doc_id: Optional[str] = None assigned_to: Optional[str] = None escalated_to: Optional[str] = None # ═══════════════════════════════════════════════════════════════════════════════ # ── 온콜/당직 관리 (tb_oncall_schedule) ────────────────────────────────────── # ═══════════════════════════════════════════════════════════════════════════════ class OncallSchedule(Base): """온콜/당직 일정 — 일별 당직자 + 에스컬레이션 대상.""" __tablename__ = "tb_oncall_schedule" id = Column(Integer, primary_key=True, index=True) duty_date = Column(Date, nullable=False, index=True) # 당직 날짜 shift = Column(String(20), default="ALL_DAY") # ALL_DAY / DAYTIME(09-18) / NIGHTTIME(18-09) engineer = Column(String(100), nullable=False, index=True) # 당직 담당자 backup_engineer = Column(String(100), nullable=True) # 백업 담당자 escalation_to = Column(String(100), nullable=True) # 에스컬레이션 note = Column(Text, nullable=True) created_by = Column(String(100), nullable=True) created_at = Column(DateTime, default=func.now()) updated_at = Column(DateTime, default=func.now(), onupdate=func.now()) class OncallScheduleOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int duty_date: date shift: str engineer: str backup_engineer: Optional[str] escalation_to: Optional[str] note: Optional[str] class OncallScheduleCreate(BaseModel): duty_date: date shift: str = "ALL_DAY" engineer: str backup_engineer: Optional[str] = None escalation_to: Optional[str] = None note: Optional[str] = None # ═══════════════════════════════════════════════════════════════════════════════ # ── On-Call 자동 로테이션 설정 (tb_oncall_rotate_config) A-5 ───────────────── # ═══════════════════════════════════════════════════════════════════════════════ class OncallRotateConfig(Base): """ On-Call 자동 로테이션 설정. 엔지니어 목록을 순환하여 매일 당직자를 자동 배정한다. 레코드는 1개(싱글톤)로 운영 — id=1 을 upsert 방식으로 사용. """ __tablename__ = "tb_oncall_rotate_config" id = Column(Integer, primary_key=True, index=True) is_active = Column(Boolean, default=True) # 자동 배정 활성 여부 # 로테이션 대상 엔지니어 목록 (JSON 배열 문자열, 예: '["alice","bob","charlie"]') engineer_list = Column(Text, nullable=False, default="[]") current_index = Column(Integer, default=0) # 현재 순번 rotate_days = Column(Integer, default=1) # 순환 주기(일) default_shift = Column(String(20), default="ALL_DAY") # ALL_DAY/DAYTIME/NIGHTTIME # 에스컬레이션 체인: 당직 미응답 시 순서대로 알림 escalation_chain = Column(Text, nullable=True) # JSON 배열 문자열 notify_on_assign = Column(Boolean, default=True) # 배정 시 알림 발송 advance_days = Column(Integer, default=1) # 며칠 전에 미리 배정 updated_at = Column(DateTime, default=func.now(), onupdate=func.now()) updated_by = Column(String(100), nullable=True) class OncallRotateConfigOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int is_active: bool engineer_list: str # raw JSON string; 클라이언트에서 파싱 current_index: int rotate_days: int default_shift: str escalation_chain: Optional[str] notify_on_assign: bool advance_days: int updated_at: Optional[datetime] updated_by: Optional[str] class OncallRotateConfigUpdate(BaseModel): is_active: Optional[bool] = None engineer_list: Optional[List[str]] = None # 파이썬 리스트로 입력 current_index: Optional[int] = None rotate_days: Optional[int] = None default_shift: Optional[str] = None escalation_chain: Optional[List[str]] = None notify_on_assign: Optional[bool] = None advance_days: Optional[int] = None # ═══════════════════════════════════════════════════════════════════════════════ # ── 배치 작업 관리 (tb_batch_job) ───────────────────────────────────────────── # ═══════════════════════════════════════════════════════════════════════════════ class BatchJobStatus(str, Enum): ACTIVE = "ACTIVE" DISABLED = "DISABLED" class BatchRunResult(str, Enum): SUCCESS = "SUCCESS" FAILED = "FAILED" TIMEOUT = "TIMEOUT" RUNNING = "RUNNING" SKIPPED = "SKIPPED" class BatchJob(Base): """배치 작업 등록 — cron 주기, 실행 서버, 명령어.""" __tablename__ = "tb_batch_job" id = Column(Integer, primary_key=True, index=True) job_name = Column(String(100), nullable=False, unique=True) inst_id = Column(Integer, ForeignKey("tb_inst_meta.id"), nullable=True) server_id = Column(Integer, ForeignKey("tb_server_info.id"), nullable=True) cron_expr = Column(String(50), nullable=False) # "0 2 * * *" command = Column(Text, nullable=False) timeout_sec = Column(Integer, default=3600) status = Column(String(20), default=BatchJobStatus.ACTIVE) alert_on_fail = Column(Boolean, default=True) owner = Column(String(100), nullable=True) last_run_at = Column(DateTime, nullable=True) last_result = Column(String(20), nullable=True) created_at = Column(DateTime, default=func.now()) updated_at = Column(DateTime, default=func.now(), onupdate=func.now()) server = relationship("Server", foreign_keys=[server_id]) class BatchRun(Base): """배치 실행 이력.""" __tablename__ = "tb_batch_run" id = Column(Integer, primary_key=True, index=True) job_id = Column(Integer, ForeignKey("tb_batch_job.id"), nullable=False, index=True) started_at = Column(DateTime, nullable=False, default=func.now()) ended_at = Column(DateTime, nullable=True) result = Column(String(20), default=BatchRunResult.RUNNING) exit_code = Column(Integer, nullable=True) stdout_tail = Column(Text, nullable=True) # 마지막 100줄 error_msg = Column(Text, nullable=True) sr_id = Column(String(30), nullable=True) # 실패 시 자동 생성 SR class BatchJobOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int job_name: str inst_id: Optional[int] server_id: Optional[int] cron_expr: str command: str timeout_sec: int status: str alert_on_fail: bool owner: Optional[str] last_run_at: Optional[datetime] last_result: Optional[str] class BatchJobCreate(BaseModel): job_name: str inst_id: Optional[int] = None server_id: Optional[int] = None cron_expr: str command: str timeout_sec: int = 3600 alert_on_fail: bool = True owner: Optional[str] = None class BatchRunOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int job_id: int started_at: datetime ended_at: Optional[datetime] result: str exit_code: Optional[int] stdout_tail: Optional[str] error_msg: Optional[str] sr_id: Optional[str] # ═══════════════════════════════════════════════════════════════════════════════ # ── SI 프로젝트 관리 — 표준 방법론 (분석→설계→구현→인도) # ═══════════════════════════════════════════════════════════════════════════════ class ProjectPhase(str, Enum): INITIATION = "INITIATION" # 착수 ANALYSIS = "ANALYSIS" # 분석 DESIGN = "DESIGN" # 설계 IMPLEMENTATION = "IMPLEMENTATION" # 구현 DEPLOYMENT = "DEPLOYMENT" # 인도 STABILIZATION = "STABILIZATION" # 안정화 CLOSED = "CLOSED" # 종료 class PhaseActivity(str, Enum): # 분석 단계 ASIS_ANALYSIS = "ASIS_ANALYSIS" # 현행시스템분석 REQ_COLLECTION = "REQ_COLLECTION" # 요구사항수집및정의 PROCESS_REDESIGN = "PROCESS_REDESIGN" # 업무프로세스재설계 DATA_MODELING = "DATA_MODELING" # 데이터모델링 REQ_CONFIRMATION = "REQ_CONFIRMATION" # 요구사항명세서확정 # 설계 단계 ARCH_DESIGN = "ARCH_DESIGN" # 시스템아키텍처설계 UI_UX_DESIGN = "UI_UX_DESIGN" # UI/UX설계 DB_DESIGN = "DB_DESIGN" # 상세데이터설계 INTERFACE_DESIGN = "INTERFACE_DESIGN" # 인터페이스설계 FRAMEWORK_DESIGN = "FRAMEWORK_DESIGN" # 공통모듈/프레임워크설계 # 구현 단계 ENV_SETUP = "ENV_SETUP" # 환경구축 CODE_DEV = "CODE_DEV" # 소스코드개발 UNIT_TEST = "UNIT_TEST" # 단위테스트 INTEGRATION_TEST = "INTEGRATION_TEST" # 통합테스트 DATA_MIGRATION = "DATA_MIGRATION" # 데이터이관 # 인도 단계 UAT = "UAT" # 사용자수용성테스트 PROD_DEPLOY = "PROD_DEPLOY" # 운영환경배포 USER_TRAINING = "USER_TRAINING" # 사용자교육 STABILIZATION_SUPPORT = "STABILIZATION_SUPPORT" # 안정화지원 CLOSEOUT = "CLOSEOUT" # 최종산출물검수및종료 # 단계별 기본 서브 활동 정의 (프로젝트 생성 시 체크리스트 자동 생성용) PHASE_DEFAULT_ACTIVITIES: dict = { ProjectPhase.ANALYSIS: [ (PhaseActivity.ASIS_ANALYSIS, "현행 시스템 분석 (AS-IS)"), (PhaseActivity.REQ_COLLECTION, "요구사항 수집 및 정의"), (PhaseActivity.PROCESS_REDESIGN, "업무 프로세스 재설계 (TO-BE)"), (PhaseActivity.DATA_MODELING, "데이터 모델링 (개념/논리)"), (PhaseActivity.REQ_CONFIRMATION, "요구사항 명세서 확정 (고객 승인)"), ], ProjectPhase.DESIGN: [ (PhaseActivity.ARCH_DESIGN, "시스템 아키텍처 설계"), (PhaseActivity.UI_UX_DESIGN, "UI/UX 설계 (스토리보드·와이어프레임)"), (PhaseActivity.DB_DESIGN, "상세 데이터 설계 (물리 DB)"), (PhaseActivity.INTERFACE_DESIGN, "인터페이스 설계 (API 명세)"), (PhaseActivity.FRAMEWORK_DESIGN, "공통 모듈/프레임워크 설계"), ], ProjectPhase.IMPLEMENTATION: [ (PhaseActivity.ENV_SETUP, "개발/테스트 환경 구축"), (PhaseActivity.CODE_DEV, "소스 코드 개발"), (PhaseActivity.UNIT_TEST, "단위 테스트"), (PhaseActivity.INTEGRATION_TEST, "통합 테스트"), (PhaseActivity.DATA_MIGRATION, "데이터 이관 개발"), ], ProjectPhase.DEPLOYMENT: [ (PhaseActivity.UAT, "사용자 수용성 테스트 (UAT)"), (PhaseActivity.PROD_DEPLOY, "운영 환경 배포"), (PhaseActivity.USER_TRAINING, "사용자 교육 (매뉴얼·교육 실시)"), (PhaseActivity.STABILIZATION_SUPPORT, "안정화 지원"), (PhaseActivity.CLOSEOUT, "최종 산출물 검수 및 프로젝트 종료"), ], } class ReqType(str, Enum): FUNCTIONAL = "FUNCTIONAL" # 기능 요구사항 NON_FUNCTIONAL = "NON_FUNCTIONAL" # 비기능 요구사항 CONSTRAINT = "CONSTRAINT" # 제약 조건 INTERFACE = "INTERFACE" # 인터페이스 요구사항 MIGRATION = "MIGRATION" # 데이터이관 요구사항 class ReqStatus(str, Enum): DRAFT = "DRAFT" CONFIRMED = "CONFIRMED" CHANGED = "CHANGED" CANCELLED = "CANCELLED" class WbsStatus(str, Enum): NOT_STARTED = "NOT_STARTED" IN_PROGRESS = "IN_PROGRESS" COMPLETED = "COMPLETED" DELAYED = "DELAYED" CANCELLED = "CANCELLED" class RiskLevel(str, Enum): LOW = "LOW" MEDIUM = "MEDIUM" HIGH = "HIGH" CRITICAL = "CRITICAL" class RiskStatus(str, Enum): OPEN = "OPEN" MITIGATED = "MITIGATED" ACCEPTED = "ACCEPTED" CLOSED = "CLOSED" class IssueType(str, Enum): BUG = "BUG" REQUIREMENT_CHANGE = "REQUIREMENT_CHANGE" RESOURCE = "RESOURCE" SCHEDULE = "SCHEDULE" TECHNICAL = "TECHNICAL" OTHER = "OTHER" class IssueStatus(str, Enum): OPEN = "OPEN" IN_PROGRESS = "IN_PROGRESS" RESOLVED = "RESOLVED" CLOSED = "CLOSED" DEFERRED = "DEFERRED" class CrType(str, Enum): SCOPE = "SCOPE" SCHEDULE = "SCHEDULE" BUDGET = "BUDGET" SPEC = "SPEC" class CrStatus(str, Enum): DRAFT = "DRAFT" REVIEW = "REVIEW" APPROVED = "APPROVED" REJECTED = "REJECTED" IMPLEMENTED = "IMPLEMENTED" class TestType(str, Enum): UNIT = "UNIT" INTEGRATION = "INTEGRATION" UAT = "UAT" PERFORMANCE = "PERFORMANCE" SECURITY = "SECURITY" class SiTestResult(str, Enum): PASS = "PASS" FAIL = "FAIL" BLOCKED = "BLOCKED" NOT_RUN = "NOT_RUN" class DefectSeverity(str, Enum): CRITICAL = "CRITICAL" HIGH = "HIGH" MEDIUM = "MEDIUM" LOW = "LOW" class DefectStatus(str, Enum): OPEN = "OPEN" ASSIGNED = "ASSIGNED" FIXED = "FIXED" VERIFIED = "VERIFIED" CLOSED = "CLOSED" DEFERRED = "DEFERRED" # ── ORM: SI 프로젝트 마스터 ─────────────────────────────────────────────────── class SiProject(Base): """SI 프로젝트 마스터 — 착수부터 SM 전환까지.""" __tablename__ = "tb_si_project" id = Column(Integer, primary_key=True, index=True) project_code = Column(String(30), unique=True, nullable=False, index=True) project_name = Column(String(200), nullable=False) inst_id = Column(Integer, ForeignKey("tb_inst_meta.id"), nullable=True) phase = Column(String(30), nullable=False, default=ProjectPhase.INITIATION) description = Column(Text, nullable=True) contract_amount = Column(Integer, nullable=True) # 계약금액 (만원) planned_start = Column(Date, nullable=True) planned_end = Column(Date, nullable=True) actual_start = Column(Date, nullable=True) actual_end = Column(Date, nullable=True) pm_user = Column(String(100), nullable=True) # 담당 PM dev_lead = Column(String(100), nullable=True) # 개발 리드 customer_pm = Column(String(100), nullable=True) # 고객사 PM team_members = Column(Text, nullable=True) # JSON array of usernames budget_total = Column(Integer, nullable=True) # 예산 (만원) budget_used = Column(Integer, default=0) overall_progress = Column(Integer, default=0) # 전체 진척률 0-100 health_status = Column(String(10), default="GREEN") # GREEN / YELLOW / RED is_active = Column(Boolean, default=True) converted_to_sm = Column(Boolean, default=False) # SM 전환 여부 note = Column(Text, nullable=True) created_by = Column(String(100), nullable=True) created_at = Column(DateTime, default=func.now()) updated_at = Column(DateTime, default=func.now(), onupdate=func.now()) institution = relationship("Institution", foreign_keys=[inst_id]) requirements = relationship("SiRequirement", back_populates="project", cascade="all, delete-orphan") wbs_items = relationship("WbsItem", back_populates="project", cascade="all, delete-orphan") issues = relationship("ProjectIssue", back_populates="project", cascade="all, delete-orphan") risks = relationship("ProjectRisk", back_populates="project", cascade="all, delete-orphan") milestones = relationship("ProjectMilestone", back_populates="project", cascade="all, delete-orphan") change_requests = relationship("ChangeRequest", back_populates="project", cascade="all, delete-orphan") test_plans = relationship("SiTestPlan", back_populates="project", cascade="all, delete-orphan") phase_checklists = relationship("SiPhaseChecklist", back_populates="project", cascade="all, delete-orphan") deliverables = relationship("Deliverable", back_populates="project", cascade="all, delete-orphan") class SiPhaseChecklist(Base): """단계별 서브 활동 체크리스트 — 표준 방법론 4단계 자동 생성.""" __tablename__ = "tb_si_phase_checklist" id = Column(Integer, primary_key=True, index=True) project_id = Column(Integer, ForeignKey("tb_si_project.id"), nullable=False, index=True) phase = Column(String(30), nullable=False) activity = Column(String(50), nullable=False) activity_name = Column(String(200), nullable=False) is_done = Column(Boolean, default=False) done_by = Column(String(100), nullable=True) done_at = Column(DateTime, nullable=True) note = Column(Text, nullable=True) created_at = Column(DateTime, default=func.now()) project = relationship("SiProject", back_populates="phase_checklists") # ── ORM: WBS ───────────────────────────────────────────────────────────────── class WbsItem(Base): """WBS 항목 — 자기 참조 계층(최대 4레벨), 리프 노드만 진척 직접 입력.""" __tablename__ = "tb_wbs_item" id = Column(Integer, primary_key=True, index=True) project_id = Column(Integer, ForeignKey("tb_si_project.id"), nullable=False, index=True) parent_id = Column(Integer, ForeignKey("tb_wbs_item.id"), nullable=True) wbs_code = Column(String(50), nullable=False) # 예: 1.2.3 title = Column(String(200), nullable=False) phase = Column(String(30), nullable=True) activity = Column(String(50), nullable=True) assignee = Column(String(100), nullable=True) planned_start = Column(Date, nullable=True) planned_end = Column(Date, nullable=True) actual_start = Column(Date, nullable=True) actual_end = Column(Date, nullable=True) completion_pct = Column(Integer, default=0) # 0-100 (리프만 직접 입력) status = Column(String(20), default=WbsStatus.NOT_STARTED) weight = Column(Integer, default=1) # 부모 진척 가중치 level = Column(Integer, default=1) # 1~4 is_leaf = Column(Boolean, default=True) deliverable = Column(String(200), nullable=True) note = Column(Text, nullable=True) created_at = Column(DateTime, default=func.now()) updated_at = Column(DateTime, default=func.now(), onupdate=func.now()) project = relationship("SiProject", back_populates="wbs_items") children = relationship( "WbsItem", backref=backref("parent", remote_side="WbsItem.id"), foreign_keys="WbsItem.parent_id", ) # ── ORM: 요구사항 ───────────────────────────────────────────────────────────── class SiRequirement(Base): """요구사항 관리 — RTM(요구사항→WBS→TC) 추적.""" __tablename__ = "tb_si_requirement" id = Column(Integer, primary_key=True, index=True) req_id = Column(String(30), unique=True, nullable=False, index=True) project_id = Column(Integer, ForeignKey("tb_si_project.id"), nullable=False, index=True) req_type = Column(String(30), nullable=False, default=ReqType.FUNCTIONAL) title = Column(String(200), nullable=False) description = Column(Text, nullable=True) source = Column(String(200), nullable=True) # 출처 (RFP p.XX, 인터뷰 등) priority = Column(String(20), default="MEDIUM") status = Column(String(20), default=ReqStatus.DRAFT) wbs_item_id = Column(Integer, ForeignKey("tb_wbs_item.id"), nullable=True) acceptance_criteria = Column(Text, nullable=True) created_by = Column(String(100), nullable=True) confirmed_by = Column(String(100), nullable=True) confirmed_at = Column(DateTime, nullable=True) created_at = Column(DateTime, default=func.now()) updated_at = Column(DateTime, default=func.now(), onupdate=func.now()) project = relationship("SiProject", back_populates="requirements") wbs_item = relationship("WbsItem", foreign_keys=[wbs_item_id]) test_cases = relationship("SiTestCase", back_populates="requirement") # ── ORM: 이슈 ───────────────────────────────────────────────────────────────── class ProjectIssue(Base): """프로젝트 이슈 관리 — 일정/자원/기술 등 장애 요인 추적.""" __tablename__ = "tb_project_issue" id = Column(Integer, primary_key=True, index=True) issue_id = Column(String(30), unique=True, nullable=False, index=True) project_id = Column(Integer, ForeignKey("tb_si_project.id"), nullable=False, index=True) wbs_item_id = Column(Integer, ForeignKey("tb_wbs_item.id"), nullable=True) issue_type = Column(String(30), nullable=False, default=IssueType.OTHER) title = Column(String(200), nullable=False) description = Column(Text, nullable=True) priority = Column(String(20), default="MEDIUM") status = Column(String(20), default=IssueStatus.OPEN) raised_by = Column(String(100), nullable=True) assigned_to = Column(String(100), nullable=True) due_date = Column(Date, nullable=True) resolved_at = Column(DateTime, nullable=True) resolution = Column(Text, nullable=True) impact = Column(Text, nullable=True) from_risk_id = Column(Integer, nullable=True) # 위험 전환 원본 ID created_at = Column(DateTime, default=func.now()) updated_at = Column(DateTime, default=func.now(), onupdate=func.now()) project = relationship("SiProject", back_populates="issues") # ── ORM: 위험 ───────────────────────────────────────────────────────────────── class ProjectRisk(Base): """위험 관리 — 확률×영향 매트릭스, HIGH×HIGH 시 자동 이슈 전환.""" __tablename__ = "tb_project_risk" id = Column(Integer, primary_key=True, index=True) risk_id = Column(String(30), unique=True, nullable=False, index=True) project_id = Column(Integer, ForeignKey("tb_si_project.id"), nullable=False, index=True) title = Column(String(200), nullable=False) description = Column(Text, nullable=True) category = Column(String(50), nullable=True) # 기술/일정/자원/요구사항/외부 probability_level = Column(Integer, nullable=False, default=2) # 1=LOW 2=MED 3=HIGH impact_level = Column(Integer, nullable=False, default=2) # 1=LOW 2=MED 3=HIGH risk_score = Column(Integer, nullable=False, default=4) # probability × impact risk_level = Column(String(20), default=RiskLevel.MEDIUM) status = Column(String(20), default=RiskStatus.OPEN) mitigation_plan = Column(Text, nullable=True) contingency_plan = Column(Text, nullable=True) owner = Column(String(100), nullable=True) due_date = Column(Date, nullable=True) converted_to_issue = Column(Boolean, default=False) created_at = Column(DateTime, default=func.now()) updated_at = Column(DateTime, default=func.now(), onupdate=func.now()) project = relationship("SiProject", back_populates="risks") # ── ORM: 마일스톤 & 산출물 ──────────────────────────────────────────────────── class ProjectMilestone(Base): """마일스톤 관리 — 단계 완료 기준점.""" __tablename__ = "tb_project_milestone" id = Column(Integer, primary_key=True, index=True) project_id = Column(Integer, ForeignKey("tb_si_project.id"), nullable=False, index=True) phase = Column(String(30), nullable=True) title = Column(String(200), nullable=False) description = Column(Text, nullable=True) planned_date = Column(Date, nullable=False) actual_date = Column(Date, nullable=True) is_completed = Column(Boolean, default=False) completed_by = Column(String(100), nullable=True) note = Column(Text, nullable=True) created_at = Column(DateTime, default=func.now()) project = relationship("SiProject", back_populates="milestones") deliverables = relationship("ProjectDeliverable", back_populates="milestone") class ProjectDeliverable(Base): """산출물 관리 — 단계별 제출 문서/결과물.""" __tablename__ = "tb_project_deliverable" id = Column(Integer, primary_key=True, index=True) project_id = Column(Integer, ForeignKey("tb_si_project.id"), nullable=False, index=True) milestone_id = Column(Integer, ForeignKey("tb_project_milestone.id"), nullable=True) phase = Column(String(30), nullable=True) activity = Column(String(50), nullable=True) title = Column(String(200), nullable=False) doc_type = Column(String(50), nullable=True) # 계획서/명세서/설계서/보고서 due_date = Column(Date, nullable=True) submitted_date = Column(Date, nullable=True) is_submitted = Column(Boolean, default=False) is_approved = Column(Boolean, default=False) approved_by = Column(String(100), nullable=True) file_path = Column(String(512), nullable=True) # 내부 경로 — API 노출 금지 version = Column(String(20), default="1.0") note = Column(Text, nullable=True) created_at = Column(DateTime, default=func.now()) project = relationship("SiProject", foreign_keys=[project_id]) milestone = relationship("ProjectMilestone", back_populates="deliverables") # ── ORM: 산출물 관리 ────────────────────────────────────────────────────────── class DeliverableStatus(str, Enum): PENDING = "PENDING" # 미제출 SUBMITTED = "SUBMITTED" # 제출됨 REVIEWING = "REVIEWING" # 검토중 APPROVED = "APPROVED" # 승인 REJECTED = "REJECTED" # 반려 class DeliverableType(str, Enum): DOCUMENT = "DOCUMENT" # 문서 (분석서, 설계서 등) CODE = "CODE" # 소스코드 TEST_RESULT = "TEST_RESULT" # 테스트 결과 DESIGN = "DESIGN" # 설계도/UI REPORT = "REPORT" # 보고서 MANUAL = "MANUAL" # 매뉴얼 OTHER = "OTHER" class Deliverable(Base): """산출물(Deliverable) — WBS 항목·마일스톤별 제출 추적.""" __tablename__ = "tb_deliverable" id = Column(Integer, primary_key=True, index=True) project_id = Column(Integer, ForeignKey("tb_si_project.id"), nullable=False, index=True) wbs_item_id = Column(Integer, ForeignKey("tb_wbs_item.id"), nullable=True) milestone_id = Column(Integer, ForeignKey("tb_project_milestone.id"), nullable=True) # 식별 name = Column(String(200), nullable=False) description = Column(Text, nullable=True) deliverable_type= Column(String(30), default=DeliverableType.DOCUMENT) # 상태 status = Column(String(20), default=DeliverableStatus.PENDING) version = Column(String(20), default="1.0") # 일정 due_date = Column(Date, nullable=True) submitted_at = Column(DateTime, nullable=True) submitted_by = Column(String(100), nullable=True) # 검토 reviewer = Column(String(100), nullable=True) reviewed_at = Column(DateTime, nullable=True) review_comment = Column(Text, nullable=True) # 파일 file_path = Column(String(500), nullable=True) file_name = Column(String(200), nullable=True) # 이력 created_by = Column(String(100), nullable=True) created_at = Column(DateTime, default=func.now()) updated_at = Column(DateTime, default=func.now(), onupdate=func.now()) project = relationship("SiProject", foreign_keys=[project_id]) class DeliverableOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int project_id: int wbs_item_id: Optional[int] = None milestone_id: Optional[int] = None name: str deliverable_type:str status: str version: str due_date: Optional[date] = None submitted_at: Optional[datetime] = None submitted_by: Optional[str] = None reviewer: Optional[str] = None reviewed_at: Optional[datetime] = None review_comment: Optional[str] = None file_name: Optional[str] = None created_at: datetime class DeliverableCreate(BaseModel): project_id: int wbs_item_id: Optional[int] = None milestone_id: Optional[int] = None name: str description: Optional[str] = None deliverable_type:str = DeliverableType.DOCUMENT version: str = "1.0" due_date: Optional[date] = None reviewer: Optional[str] = None class DeliverableUpdate(BaseModel): name: Optional[str] = None description: Optional[str] = None status: Optional[str] = None version: Optional[str] = None due_date: Optional[date] = None reviewer: Optional[str] = None review_comment: Optional[str] = None # ── ORM: 변경 요청 ──────────────────────────────────────────────────────────── class ChangeRequest(Base): """변경 요청 (CR) — 범위/일정/예산/명세 변경 이력 관리.""" __tablename__ = "tb_change_request" id = Column(Integer, primary_key=True, index=True) cr_id = Column(String(30), unique=True, nullable=False, index=True) project_id = Column(Integer, ForeignKey("tb_si_project.id"), nullable=False, index=True) cr_type = Column(String(20), nullable=False, default=CrType.SCOPE) title = Column(String(200), nullable=False) description = Column(Text, nullable=True) reason = Column(Text, nullable=True) impact_scope = Column(Text, nullable=True) impact_schedule = Column(Text, nullable=True) impact_budget = Column(Text, nullable=True) status = Column(String(20), default=CrStatus.DRAFT) requested_by = Column(String(100), nullable=True) reviewed_by = Column(String(100), nullable=True) approved_by = Column(String(100), nullable=True) approved_at = Column(DateTime, nullable=True) implemented_at = Column(DateTime, nullable=True) created_at = Column(DateTime, default=func.now()) updated_at = Column(DateTime, default=func.now(), onupdate=func.now()) project = relationship("SiProject", back_populates="change_requests") # ── ORM: 테스트 관리 ────────────────────────────────────────────────────────── class SiTestPlan(Base): """테스트 계획 — 단위/통합/UAT 유형별 관리.""" __tablename__ = "tb_si_test_plan" id = Column(Integer, primary_key=True, index=True) project_id = Column(Integer, ForeignKey("tb_si_project.id"), nullable=False, index=True) phase = Column(String(30), nullable=True) test_type = Column(String(30), nullable=False, default=TestType.INTEGRATION) plan_name = Column(String(200), nullable=False) description = Column(Text, nullable=True) planned_start = Column(Date, nullable=True) planned_end = Column(Date, nullable=True) actual_start = Column(Date, nullable=True) actual_end = Column(Date, nullable=True) total_cases = Column(Integer, default=0) pass_count = Column(Integer, default=0) fail_count = Column(Integer, default=0) blocked_count = Column(Integer, default=0) pass_rate = Column(Integer, default=0) is_completed = Column(Boolean, default=False) created_by = Column(String(100), nullable=True) created_at = Column(DateTime, default=func.now()) updated_at = Column(DateTime, default=func.now(), onupdate=func.now()) project = relationship("SiProject", back_populates="test_plans") test_cases = relationship("SiTestCase", back_populates="test_plan", cascade="all, delete-orphan") class SiTestCase(Base): """테스트 케이스 — RTM(TC→요구사항) 연결.""" __tablename__ = "tb_si_test_case" id = Column(Integer, primary_key=True, index=True) tc_id = Column(String(30), unique=True, nullable=False, index=True) plan_id = Column(Integer, ForeignKey("tb_si_test_plan.id"), nullable=False, index=True) req_id = Column(Integer, ForeignKey("tb_si_requirement.id"), nullable=True) title = Column(String(200), nullable=False) precondition = Column(Text, nullable=True) test_steps = Column(Text, nullable=True) # JSON array expected_result = Column(Text, nullable=True) last_result = Column(String(20), default=SiTestResult.NOT_RUN) assigned_to = Column(String(100), nullable=True) priority = Column(String(20), default="MEDIUM") created_at = Column(DateTime, default=func.now()) updated_at = Column(DateTime, default=func.now(), onupdate=func.now()) test_plan = relationship("SiTestPlan", back_populates="test_cases") requirement = relationship("SiRequirement", back_populates="test_cases") executions = relationship("SiTestExecution", back_populates="test_case", cascade="all, delete-orphan") class SiDefect(Base): """결함 관리 — 심각도별 OPEN→FIXED→VERIFIED 흐름.""" __tablename__ = "tb_si_defect" id = Column(Integer, primary_key=True, index=True) defect_id = Column(String(30), unique=True, nullable=False, index=True) project_id = Column(Integer, ForeignKey("tb_si_project.id"), nullable=False, index=True) tc_id = Column(Integer, ForeignKey("tb_si_test_case.id"), nullable=True) req_id = Column(Integer, ForeignKey("tb_si_requirement.id"), nullable=True) title = Column(String(200), nullable=False) description = Column(Text, nullable=True) severity = Column(String(20), nullable=False, default=DefectSeverity.MEDIUM) status = Column(String(20), nullable=False, default=DefectStatus.OPEN) phase = Column(String(30), nullable=True) # 발견 단계 reported_by = Column(String(100), nullable=True) assigned_to = Column(String(100), nullable=True) fixed_by = Column(String(100), nullable=True) fixed_at = Column(DateTime, nullable=True) verified_by = Column(String(100), nullable=True) verified_at = Column(DateTime, nullable=True) build_found = Column(String(50), nullable=True) build_fixed = Column(String(50), nullable=True) note = Column(Text, nullable=True) created_at = Column(DateTime, default=func.now()) updated_at = Column(DateTime, default=func.now(), onupdate=func.now()) class SiTestExecution(Base): """테스트 실행 이력 — 동일 TC 반복 실행 추적.""" __tablename__ = "tb_si_test_execution" id = Column(Integer, primary_key=True, index=True) tc_id = Column(Integer, ForeignKey("tb_si_test_case.id"), nullable=False, index=True) executed_by = Column(String(100), nullable=True) executed_at = Column(DateTime, default=func.now()) result = Column(String(20), nullable=False, default=SiTestResult.NOT_RUN) actual_result = Column(Text, nullable=True) defect_id = Column(Integer, ForeignKey("tb_si_defect.id"), nullable=True) note = Column(Text, nullable=True) build_version = Column(String(50), nullable=True) test_case = relationship("SiTestCase", back_populates="executions") # ── Pydantic Schemas: SiProject ─────────────────────────────────────────────── class SiProjectOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int project_code: str project_name: str inst_id: Optional[int] phase: str description: Optional[str] contract_amount: Optional[int] planned_start: Optional[date] planned_end: Optional[date] actual_start: Optional[date] actual_end: Optional[date] pm_user: Optional[str] dev_lead: Optional[str] customer_pm: Optional[str] team_members: Optional[str] budget_total: Optional[int] budget_used: int overall_progress: int health_status: str is_active: bool converted_to_sm: bool note: Optional[str] created_by: Optional[str] created_at: datetime updated_at: Optional[datetime] class SiProjectCreate(BaseModel): project_name: str inst_id: Optional[int] = None description: Optional[str] = None contract_amount: Optional[int] = None planned_start: Optional[date] = None planned_end: Optional[date] = None pm_user: Optional[str] = None dev_lead: Optional[str] = None customer_pm: Optional[str] = None team_members: Optional[str] = None budget_total: Optional[int] = None note: Optional[str] = None class SiProjectUpdate(BaseModel): project_name: Optional[str] = None inst_id: Optional[int] = None phase: Optional[ProjectPhase] = None description: Optional[str] = None contract_amount: Optional[int] = None planned_start: Optional[date] = None planned_end: Optional[date] = None actual_start: Optional[date] = None actual_end: Optional[date] = None pm_user: Optional[str] = None dev_lead: Optional[str] = None customer_pm: Optional[str] = None team_members: Optional[str] = None budget_total: Optional[int] = None budget_used: Optional[int] = None overall_progress: Optional[int] = None health_status: Optional[str] = None is_active: Optional[bool] = None note: Optional[str] = None class SiPhaseChecklistOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int project_id: int phase: str activity: str activity_name: str is_done: bool done_by: Optional[str] done_at: Optional[datetime] note: Optional[str] class SiPhaseChecklistUpdate(BaseModel): is_done: bool done_by: Optional[str] = None note: Optional[str] = None # ── Pydantic Schemas: WBS ───────────────────────────────────────────────────── class WbsItemOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int project_id: int parent_id: Optional[int] wbs_code: str title: str phase: Optional[str] activity: Optional[str] assignee: Optional[str] planned_start: Optional[date] planned_end: Optional[date] actual_start: Optional[date] actual_end: Optional[date] completion_pct: int status: str weight: int level: int is_leaf: bool deliverable: Optional[str] note: Optional[str] created_at: datetime updated_at: Optional[datetime] class WbsItemCreate(BaseModel): parent_id: Optional[int] = None wbs_code: str title: str phase: Optional[ProjectPhase] = None activity: Optional[PhaseActivity] = None assignee: Optional[str] = None planned_start: Optional[date] = None planned_end: Optional[date] = None weight: int = 1 deliverable: Optional[str] = None note: Optional[str] = None class WbsItemUpdate(BaseModel): title: Optional[str] = None phase: Optional[ProjectPhase] = None activity: Optional[PhaseActivity] = None assignee: Optional[str] = None planned_start: Optional[date] = None planned_end: Optional[date] = None actual_start: Optional[date] = None actual_end: Optional[date] = None completion_pct: Optional[int] = None status: Optional[WbsStatus] = None weight: Optional[int] = None deliverable: Optional[str] = None note: Optional[str] = None # ── Pydantic Schemas: 요구사항 ──────────────────────────────────────────────── class SiRequirementOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int req_id: str project_id: int req_type: str title: str description: Optional[str] source: Optional[str] priority: str status: str wbs_item_id: Optional[int] acceptance_criteria: Optional[str] created_by: Optional[str] confirmed_by: Optional[str] confirmed_at: Optional[datetime] created_at: datetime updated_at: Optional[datetime] class SiRequirementCreate(BaseModel): req_type: ReqType = ReqType.FUNCTIONAL title: str description: Optional[str] = None source: Optional[str] = None priority: str = "MEDIUM" wbs_item_id: Optional[int] = None acceptance_criteria: Optional[str] = None created_by: Optional[str] = None class SiRequirementUpdate(BaseModel): req_type: Optional[ReqType] = None title: Optional[str] = None description: Optional[str] = None source: Optional[str] = None priority: Optional[str] = None status: Optional[ReqStatus] = None wbs_item_id: Optional[int] = None acceptance_criteria: Optional[str] = None confirmed_by: Optional[str] = None # ── Pydantic Schemas: 이슈 ──────────────────────────────────────────────────── class ProjectIssueOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int issue_id: str project_id: int wbs_item_id: Optional[int] issue_type: str title: str description: Optional[str] priority: str status: str raised_by: Optional[str] assigned_to: Optional[str] due_date: Optional[date] resolved_at: Optional[datetime] resolution: Optional[str] impact: Optional[str] from_risk_id: Optional[int] created_at: datetime updated_at: Optional[datetime] class ProjectIssueCreate(BaseModel): wbs_item_id: Optional[int] = None issue_type: IssueType = IssueType.OTHER title: str description: Optional[str] = None priority: str = "MEDIUM" raised_by: Optional[str] = None assigned_to: Optional[str] = None due_date: Optional[date] = None impact: Optional[str] = None class ProjectIssueUpdate(BaseModel): issue_type: Optional[IssueType] = None title: Optional[str] = None description: Optional[str] = None priority: Optional[str] = None status: Optional[IssueStatus] = None assigned_to: Optional[str] = None due_date: Optional[date] = None resolved_at: Optional[datetime] = None resolution: Optional[str] = None impact: Optional[str] = None # ── Pydantic Schemas: 위험 ──────────────────────────────────────────────────── class ProjectRiskOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int risk_id: str project_id: int title: str description: Optional[str] category: Optional[str] probability_level: int impact_level: int risk_score: int risk_level: str status: str mitigation_plan: Optional[str] contingency_plan: Optional[str] owner: Optional[str] due_date: Optional[date] converted_to_issue: bool created_at: datetime updated_at: Optional[datetime] class ProjectRiskCreate(BaseModel): title: str description: Optional[str] = None category: Optional[str] = None probability_level: int = 2 impact_level: int = 2 mitigation_plan: Optional[str] = None contingency_plan: Optional[str] = None owner: Optional[str] = None due_date: Optional[date] = None class ProjectRiskUpdate(BaseModel): title: Optional[str] = None description: Optional[str] = None category: Optional[str] = None probability_level: Optional[int] = None impact_level: Optional[int] = None status: Optional[RiskStatus] = None mitigation_plan: Optional[str] = None contingency_plan: Optional[str] = None owner: Optional[str] = None due_date: Optional[date] = None # ── Pydantic Schemas: 마일스톤/산출물 ──────────────────────────────────────── class ProjectMilestoneOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int project_id: int phase: Optional[str] title: str description: Optional[str] planned_date: date actual_date: Optional[date] is_completed: bool completed_by: Optional[str] note: Optional[str] created_at: datetime class ProjectMilestoneCreate(BaseModel): phase: Optional[ProjectPhase] = None title: str description: Optional[str] = None planned_date: date note: Optional[str] = None class ProjectMilestoneUpdate(BaseModel): title: Optional[str] = None description: Optional[str] = None planned_date: Optional[date] = None actual_date: Optional[date] = None is_completed: Optional[bool] = None completed_by: Optional[str] = None note: Optional[str] = None class ProjectDeliverableOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int project_id: int milestone_id: Optional[int] phase: Optional[str] activity: Optional[str] title: str doc_type: Optional[str] due_date: Optional[date] submitted_date: Optional[date] is_submitted: bool is_approved: bool approved_by: Optional[str] version: str note: Optional[str] created_at: datetime class ProjectDeliverableCreate(BaseModel): milestone_id: Optional[int] = None phase: Optional[ProjectPhase] = None activity: Optional[PhaseActivity] = None title: str doc_type: Optional[str] = None due_date: Optional[date] = None version: str = "1.0" note: Optional[str] = None class ProjectDeliverableUpdate(BaseModel): title: Optional[str] = None doc_type: Optional[str] = None due_date: Optional[date] = None submitted_date: Optional[date] = None is_submitted: Optional[bool] = None is_approved: Optional[bool] = None approved_by: Optional[str] = None version: Optional[str] = None note: Optional[str] = None # ── Pydantic Schemas: 변경 요청 ─────────────────────────────────────────────── class ChangeRequestOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int cr_id: str project_id: int cr_type: str title: str description: Optional[str] reason: Optional[str] impact_scope: Optional[str] impact_schedule: Optional[str] impact_budget: Optional[str] status: str requested_by: Optional[str] reviewed_by: Optional[str] approved_by: Optional[str] approved_at: Optional[datetime] implemented_at: Optional[datetime] created_at: datetime updated_at: Optional[datetime] class ChangeRequestCreate(BaseModel): cr_type: CrType = CrType.SCOPE title: str description: Optional[str] = None reason: Optional[str] = None impact_scope: Optional[str] = None impact_schedule: Optional[str] = None impact_budget: Optional[str] = None requested_by: Optional[str] = None class ChangeRequestUpdate(BaseModel): cr_type: Optional[CrType] = None title: Optional[str] = None description: Optional[str] = None reason: Optional[str] = None impact_scope: Optional[str] = None impact_schedule: Optional[str] = None impact_budget: Optional[str] = None status: Optional[CrStatus] = None reviewed_by: Optional[str] = None approved_by: Optional[str] = None approved_at: Optional[datetime] = None implemented_at: Optional[datetime] = None # ── Pydantic Schemas: 테스트 ────────────────────────────────────────────────── class SiTestPlanOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int project_id: int phase: Optional[str] test_type: str plan_name: str description: Optional[str] planned_start: Optional[date] planned_end: Optional[date] actual_start: Optional[date] actual_end: Optional[date] total_cases: int pass_count: int fail_count: int blocked_count: int pass_rate: int is_completed: bool created_by: Optional[str] created_at: datetime updated_at: Optional[datetime] class SiTestPlanCreate(BaseModel): phase: Optional[ProjectPhase] = None test_type: TestType = TestType.INTEGRATION plan_name: str description: Optional[str] = None planned_start: Optional[date] = None planned_end: Optional[date] = None created_by: Optional[str] = None class SiTestCaseOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int tc_id: str plan_id: int req_id: Optional[int] title: str precondition: Optional[str] test_steps: Optional[str] expected_result: Optional[str] last_result: str assigned_to: Optional[str] priority: str created_at: datetime updated_at: Optional[datetime] class SiTestCaseCreate(BaseModel): req_id: Optional[int] = None title: str precondition: Optional[str] = None test_steps: Optional[str] = None expected_result: Optional[str] = None assigned_to: Optional[str] = None priority: str = "MEDIUM" class SiTestExecutionOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int tc_id: int executed_by: Optional[str] executed_at: datetime result: str actual_result: Optional[str] defect_id: Optional[int] note: Optional[str] build_version: Optional[str] class SiTestExecutionCreate(BaseModel): executed_by: Optional[str] = None result: SiTestResult actual_result: Optional[str] = None note: Optional[str] = None build_version: Optional[str] = None class SiDefectOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int defect_id: str project_id: int tc_id: Optional[int] req_id: Optional[int] title: str description: Optional[str] severity: str status: str phase: Optional[str] reported_by: Optional[str] assigned_to: Optional[str] fixed_by: Optional[str] fixed_at: Optional[datetime] verified_by: Optional[str] verified_at: Optional[datetime] build_found: Optional[str] build_fixed: Optional[str] note: Optional[str] created_at: datetime updated_at: Optional[datetime] class SiDefectCreate(BaseModel): tc_id: Optional[int] = None req_id: Optional[int] = None title: str description: Optional[str] = None severity: DefectSeverity = DefectSeverity.MEDIUM phase: Optional[str] = None reported_by: Optional[str] = None assigned_to: Optional[str] = None build_found: Optional[str] = None note: Optional[str] = None class SiDefectUpdate(BaseModel): title: Optional[str] = None description: Optional[str] = None severity: Optional[DefectSeverity] = None status: Optional[DefectStatus] = None assigned_to: Optional[str] = None fixed_by: Optional[str] = None fixed_at: Optional[datetime] = None verified_by: Optional[str] = None verified_at: Optional[datetime] = None build_fixed: Optional[str] = None note: Optional[str] = None # ══════════════════════════════════════════════════════════════════════════════ # ── AI 에이전트 (Paperclip 스타일 하트비트 기반) ────────────────────────────── # 온프레미스 보안: 모든 LLM 추론은 Ollama(localhost:11434) 전용. # 외부 AI/API 호출 금지 — core/llm_client.py 참조. # ══════════════════════════════════════════════════════════════════════════════ class AgentRole(str, Enum): # ── 개발 조직 에이전트 (Phase 1 Paperclip) ── CEO = "CEO" # 총괄: 목표 분배·보고 CTO = "CTO" # 기술 아키텍처·코드 리뷰 DEVELOPER = "DEVELOPER" # 코드 생성·구현 QA = "QA" # 테스트케이스·품질 PM_AGENT = "PM_AGENT" # 일정·WBS 관리 # ── 운영 자동화 에이전트 (Phase 3 런타임) ── INCIDENT_TRIAGE = "INCIDENT_TRIAGE" # 인시던트 자동 분류·배정 KB_CURATOR = "KB_CURATOR" # 해결 SR → KB 자동 생성 SSL_WATCHER = "SSL_WATCHER" # SSL 만료 → SR 자동 생성 WBS_MONITOR = "WBS_MONITOR" # WBS 지연 위험 분석 PM_SUGGESTER = "PM_SUGGESTER" # PM 스케줄 미등록 서버 권고 class AgentStatus(str, Enum): IDLE = "IDLE" # 대기 (하트비트 사이) ACTIVE = "ACTIVE" # 하트비트로 깨어남 WORKING = "WORKING" # 작업 처리 중 ERROR = "ERROR" # 오류 발생 PAUSED = "PAUSED" # 일시 중지 (관리자) class LLMProvider(str, Enum): OLLAMA = "ollama" # 로컬 Ollama (온프레미스, 기본값·권장) CLAUDE = "claude" # Anthropic Claude API (개발·테스트 전용) OPENAI = "openai" # OpenAI API (개발·테스트 전용) class AgentTaskStatus(str, Enum): PENDING = "PENDING" IN_PROGRESS = "IN_PROGRESS" COMPLETED = "COMPLETED" FAILED = "FAILED" CANCELLED = "CANCELLED" class AgentApprovalStatus(str, Enum): PENDING = "PENDING" # 사람 승인 대기 APPROVED = "APPROVED" # 승인 REJECTED = "REJECTED" # 거부 AUTO_APPROVED = "AUTO_APPROVED" # 저위험 자동 승인 # ── ORM Models ─────────────────────────────────────────────────────────────── class AgentConfig(Base): """에이전트 설정 — 역할·LLM 모델·하트비트 스케줄 정의.""" __tablename__ = "tb_agent_config" id = Column(Integer, primary_key=True, index=True) name = Column(String(100), nullable=False) role = Column(String(30), nullable=False) # AgentRole description = Column(Text) llm_provider = Column(String(20), default="ollama") # LLMProvider llm_model = Column(String(80), default="guardia-agent") # Ollama 모델명 system_prompt = Column(Text) heartbeat_cron = Column(String(50)) # cron 표현식 예: "*/15 * * * *" is_active = Column(Boolean, default=True) status = Column(String(20), default="IDLE") # AgentStatus last_heartbeat = Column(DateTime) last_error = Column(Text) total_tasks = Column(Integer, default=0) total_tokens = Column(Integer, default=0) created_by = Column(Integer, ForeignKey("tb_user.id"), nullable=True) created_at = Column(DateTime, default=func.now()) updated_at = Column(DateTime, onupdate=func.now()) tasks = relationship("AgentTask", back_populates="agent", cascade="all, delete-orphan") approvals = relationship("AgentApproval", back_populates="agent", cascade="all, delete-orphan") class AgentTask(Base): """에이전트 작업 실행 기록 — 입출력·토큰·상태.""" __tablename__ = "tb_agent_task" id = Column(Integer, primary_key=True, index=True) agent_id = Column(Integer, ForeignKey("tb_agent_config.id"), nullable=False) title = Column(String(200), nullable=False) description = Column(Text) status = Column(String(20), default="PENDING") # AgentTaskStatus input_data = Column(JSON) # 에이전트에게 전달된 컨텍스트 output_data = Column(JSON) # 에이전트가 생성한 결과 tokens_used = Column(Integer, default=0) started_at = Column(DateTime) completed_at = Column(DateTime) created_at = Column(DateTime, default=func.now()) agent = relationship("AgentConfig", back_populates="tasks") approval = relationship("AgentApproval", back_populates="task", uselist=False) class AgentApproval(Base): """ 사람 검토·승인 대기 항목. CRITICAL 인시던트 배정, 코드 변경, SR 자동 생성 등 고위험 액션은 사람이 승인·거부해야 실제 적용된다. """ __tablename__ = "tb_agent_approval" id = Column(Integer, primary_key=True, index=True) agent_id = Column(Integer, ForeignKey("tb_agent_config.id"), nullable=False) task_id = Column(Integer, ForeignKey("tb_agent_task.id"), nullable=True) action_type = Column(String(50), nullable=False) # 예: TRIAGE_INCIDENT / CREATE_SR / CODE_CHANGE / CREATE_RISK / ASSIGN_ENGINEER action_data = Column(JSON) # 적용 예정 데이터 status = Column(String(20), default="PENDING") # AgentApprovalStatus notes = Column(Text) # 승인·거부 사유 requested_at = Column(DateTime, default=func.now()) reviewed_by = Column(Integer, ForeignKey("tb_user.id"), nullable=True) reviewed_at = Column(DateTime, nullable=True) agent = relationship("AgentConfig", back_populates="approvals") task = relationship("AgentTask", back_populates="approval") # ── Pydantic Schemas ────────────────────────────────────────────────────────── class AgentConfigOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int name: str role: str description: Optional[str] llm_provider: str llm_model: str heartbeat_cron: Optional[str] is_active: bool status: str last_heartbeat: Optional[datetime] total_tasks: int total_tokens: int created_at: datetime class AgentConfigCreate(BaseModel): name: str role: AgentRole description: Optional[str] = None llm_provider: LLMProvider = LLMProvider.OLLAMA llm_model: str = "guardia-agent" system_prompt: Optional[str] = None heartbeat_cron: Optional[str] = None # None = 수동 실행만 is_active: bool = True class AgentConfigUpdate(BaseModel): name: Optional[str] = None description: Optional[str] = None llm_model: Optional[str] = None system_prompt: Optional[str] = None heartbeat_cron: Optional[str] = None is_active: Optional[bool] = None class AgentTaskOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int agent_id: int title: str status: str input_data: Optional[dict] output_data: Optional[dict] tokens_used: int started_at: Optional[datetime] completed_at: Optional[datetime] created_at: datetime class AgentTaskCreate(BaseModel): agent_id: int title: str description: Optional[str] = None input_data: Optional[dict] = None class AgentApprovalOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int agent_id: int task_id: Optional[int] action_type: str action_data: Optional[dict] status: str notes: Optional[str] requested_at: datetime reviewed_by: Optional[int] reviewed_at: Optional[datetime] class AgentApprovalReview(BaseModel): approved: bool notes: Optional[str] = None class AgentOrgNode(BaseModel): """에이전트 조직도 노드 (Phase 4 대시보드용).""" id: int name: str role: str status: str children: List["AgentOrgNode"] = [] model_config = ConfigDict(from_attributes=True) # ══════════════════════════════════════════════════════════════════════════════ # B-3: 코드 리뷰 (Code Review Agent) # ══════════════════════════════════════════════════════════════════════════════ class ReviewSeverity(str, Enum): CRITICAL = "CRITICAL" HIGH = "HIGH" MEDIUM = "MEDIUM" LOW = "LOW" INFO = "INFO" class ReviewCategory(str, Enum): SECURITY = "SECURITY" # 보안 취약점 PERFORMANCE = "PERFORMANCE" # 성능 문제 CODE_QUALITY = "CODE_QUALITY" # 코드 품질 ARCHITECTURE = "ARCHITECTURE" # 아키텍처 문제 NAMING = "NAMING" # 네이밍 컨벤션 TESTING = "TESTING" # 테스트 부족 DOCUMENTATION= "DOCUMENTATION" # 문서화 부족 class CodeReview(Base): """코드 리뷰 결과 저장.""" __tablename__ = "tb_code_review" id = Column(Integer, primary_key=True, index=True) project_id = Column(Integer, ForeignKey("tb_project.id"), nullable=True, index=True) vibe_session_id= Column(Integer, ForeignKey("tb_vibe_session.id"), nullable=True, index=True) sr_id = Column(String(50), nullable=True, index=True) reviewer = Column(String(100), default="code-review-agent") # 리뷰어 (에이전트명) target_path = Column(String(512), nullable=True) # 리뷰 대상 경로 tech_stack = Column(String(50), nullable=True) # 리뷰 결과 score = Column(Integer, default=0) # 0-100 종합 점수 summary = Column(Text, nullable=True) # LLM 요약 findings_json = Column(Text, nullable=True) # JSON: List[ReviewFinding] # 메타 file_count = Column(Integer, default=0) line_count = Column(Integer, default=0) model_used = Column(String(100), nullable=True) # 사용된 LLM 모델 duration_sec = Column(Integer, nullable=True) status = Column(String(20), default="PENDING") # PENDING/RUNNING/DONE/FAILED error_msg = Column(Text, nullable=True) requested_by = Column(String(100), nullable=True) created_at = Column(DateTime, default=func.now()) completed_at = Column(DateTime, nullable=True) class CodeReviewOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int project_id: Optional[int] vibe_session_id: Optional[int] sr_id: Optional[str] reviewer: str target_path: Optional[str] tech_stack: Optional[str] score: int summary: Optional[str] findings_json: Optional[str] file_count: int line_count: int model_used: Optional[str] duration_sec: Optional[int] status: str error_msg: Optional[str] requested_by: Optional[str] created_at: datetime completed_at: Optional[datetime] class CodeReviewRequest(BaseModel): """코드 리뷰 요청.""" project_id: Optional[int] = None vibe_session_id: Optional[int] = None sr_id: Optional[str] = None target_path: Optional[str] = None # 특정 파일/디렉토리만 리뷰 (None=project_dir 전체) focus: Optional[str] = None # 리뷰 집중 포인트 (예: "security,performance") model: str = "codellama" # Ollama 모델명 class AgentStatsOut(BaseModel): """에이전트 엔진 전체 통계.""" total_agents: int active_agents: int total_tasks_today: int total_tokens_today: int pending_approvals: int llm_online: bool # ── 에이전트 간 메시지 (Priority 3 — 에이전트 메시지 전달) ──────────────────── class AgentMessageType(str, Enum): TASK_DELEGATION = "TASK_DELEGATION" # CEO → CTO: 개발 태스크 위임 STATUS_UPDATE = "STATUS_UPDATE" # CTO/Dev → CEO: 완료 보고 ESCALATION = "ESCALATION" # 하위 에이전트 → 상위 에이전트 class AgentMessage(Base): """에이전트 간 메시지 전달 테이블.""" __tablename__ = "tb_agent_message" id = Column(Integer, primary_key=True, index=True) from_agent_id = Column(Integer, ForeignKey("tb_agent_config.id"), nullable=True, index=True) to_agent_id = Column(Integer, ForeignKey("tb_agent_config.id"), nullable=False, index=True) message_type = Column(String(30), nullable=False) subject = Column(String(200), nullable=False) body = Column(Text, nullable=True) metadata_json = Column(Text, nullable=True) # JSON 문자열 저장 is_read = Column(Boolean, default=False) created_at = Column(DateTime, default=func.now()) read_at = Column(DateTime, nullable=True) from_agent = relationship("AgentConfig", foreign_keys=[from_agent_id]) to_agent = relationship("AgentConfig", foreign_keys=[to_agent_id]) class AgentMessageOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int from_agent_id: Optional[int] to_agent_id: int message_type: str subject: str body: Optional[str] metadata_json: Optional[str] is_read: bool created_at: datetime read_at: Optional[datetime] class AgentMessageCreate(BaseModel): from_agent_id: Optional[int] = None to_agent_id: int message_type: AgentMessageType subject: str body: Optional[str] = None metadata_json: Optional[str] = None # ══════════════════════════════════════════════════════════════════════════════ # B-1: AI 이상 탐지 (Anomaly Detection) 모델 # ══════════════════════════════════════════════════════════════════════════════ class AnomalySeverity(str, Enum): INFO = "INFO" WARNING = "WARNING" CRITICAL = "CRITICAL" class AnomalyStatus(str, Enum): OPEN = "OPEN" ACKNOWLEDGED = "ACKNOWLEDGED" RESOLVED = "RESOLVED" FALSE_POSITIVE = "FALSE_POSITIVE" class MetricType(str, Enum): CPU_USAGE = "CPU_USAGE" # % MEMORY_USAGE = "MEMORY_USAGE" # % DISK_USAGE = "DISK_USAGE" # % RESPONSE_TIME = "RESPONSE_TIME" # ms ERROR_RATE = "ERROR_RATE" # per minute NETWORK_RX = "NETWORK_RX" # KB/s NETWORK_TX = "NETWORK_TX" # KB/s ACTIVE_CONNECTIONS = "ACTIVE_CONNECTIONS" # count HEAP_USAGE = "HEAP_USAGE" # % (JVM) THROUGHPUT = "THROUGHPUT" # req/s CUSTOM = "CUSTOM" class MetricSnapshot(Base): """시계열 메트릭 스냅샷 (이상 탐지 학습/검출 데이터).""" __tablename__ = "tb_metric_snapshot" id = Column(Integer, primary_key=True, index=True) server_id = Column(Integer, ForeignKey("tb_server_info.id"), nullable=True) source = Column(String(100), nullable=False) # 서버명 또는 앱명 metric_type = Column(String(50), nullable=False) value = Column(Float, nullable=False) unit = Column(String(20), nullable=True) # %, ms, KB/s, req/s 등 tags = Column(Text, nullable=True) # JSON 태그 (inst_code 등) recorded_at = Column(DateTime, default=func.now(), index=True) class AnomalyEvent(Base): """이상 탐지 이벤트.""" __tablename__ = "tb_anomaly_event" id = Column(Integer, primary_key=True, index=True) server_id = Column(Integer, ForeignKey("tb_server_info.id"), nullable=True) source = Column(String(100), nullable=False) metric_type = Column(String(50), nullable=False) severity = Column(String(20), default="WARNING") status = Column(String(20), default="OPEN") title = Column(String(200), nullable=False) description = Column(Text, nullable=True) # 탐지 수치 current_value = Column(Float, nullable=False) baseline_mean = Column(Float, nullable=True) baseline_std = Column(Float, nullable=True) z_score = Column(Float, nullable=True) threshold = Column(Float, nullable=True) detect_method = Column(String(30), nullable=True) # ZSCORE/IQR/THRESHOLD/TREND # Ollama LLM 분석 (선택적) llm_analysis = Column(Text, nullable=True) # 연결 sr_id = Column(String(20), nullable=True) # 자동 생성 SR rule_name = Column(String(100), nullable=True) # 적용 룰명 # 타임스탬프 detected_at = Column(DateTime, default=func.now(), index=True) acknowledged_at = Column(DateTime, nullable=True) resolved_at = Column(DateTime, nullable=True) acknowledged_by = Column(String(100), nullable=True) server = relationship("Server", foreign_keys=[server_id]) class AnomalyRule(Base): """이상 탐지 룰 설정 (임계값, 알림 조건).""" __tablename__ = "tb_anomaly_rule" id = Column(Integer, primary_key=True, index=True) name = Column(String(100), nullable=False, unique=True) description = Column(String(300), nullable=True) source_pattern = Column(String(100), nullable=True) # 적용 소스 패턴 (None=전체) metric_type = Column(String(50), nullable=False) method = Column(String(20), default="ZSCORE") # ZSCORE/IQR/THRESHOLD/TREND threshold = Column(Float, nullable=True) # THRESHOLD 방식: 정적 임계값 z_threshold = Column(Float, default=3.0) # ZSCORE 방식: Z-score 임계 iqr_factor = Column(Float, default=1.5) # IQR 방식: IQR 배율 window_size = Column(Integer, default=60) # 기준 윈도우 샘플 수 min_samples = Column(Integer, default=10) # 최소 필요 샘플 수 severity = Column(String(20), default="WARNING") auto_create_sr = Column(Boolean, default=False) # CRITICAL 시 SR 자동 생성 is_active = Column(Boolean, default=True) created_at = Column(DateTime, default=func.now()) # ── B-1 Pydantic schemas ──────────────────────────────────────────────────── class MetricSnapshotIn(BaseModel): """메트릭 수집 요청.""" source: str metric_type: str # MetricType value value: float unit: Optional[str] = None tags: Optional[dict] = None server_id: Optional[int] = None class MetricSnapshotOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int source: str metric_type: str value: float unit: Optional[str] tags: Optional[str] recorded_at: datetime class AnomalyEventOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int source: str metric_type: str severity: str status: str title: str description: Optional[str] current_value: float baseline_mean: Optional[float] baseline_std: Optional[float] z_score: Optional[float] threshold: Optional[float] detect_method: Optional[str] llm_analysis: Optional[str] sr_id: Optional[str] rule_name: Optional[str] detected_at: datetime acknowledged_at: Optional[datetime] resolved_at: Optional[datetime] acknowledged_by: Optional[str] class AnomalyRuleOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int name: str description: Optional[str] source_pattern: Optional[str] metric_type: str method: str threshold: Optional[float] z_threshold: float iqr_factor: float window_size: int min_samples: int severity: str auto_create_sr: bool is_active: bool class AnomalyRuleCreate(BaseModel): name: str description: Optional[str] = None source_pattern: Optional[str] = None metric_type: str method: str = "ZSCORE" threshold: Optional[float] = None z_threshold: float = 3.0 iqr_factor: float = 1.5 window_size: int = 60 min_samples: int = 10 severity: str = "WARNING" auto_create_sr: bool = False class SimulateMetricIn(BaseModel): """이상 탐지 시뮬레이션 (테스트용).""" source: str = "test-server" metric_type: str = "CPU_USAGE" normal_count: int = 50 # 정상 데이터 포인트 수 anomaly_value: float = 95.0 # 이상 값 baseline_mean: float = 40.0 baseline_std: float = 10.0 include_llm: bool = False # Ollama LLM 분석 포함 여부 # ============================================================================== # B-2: 자연어 SR 접수 챗봇 모델 # ============================================================================== class ChatSessionStatus(str, Enum): ACTIVE = "ACTIVE" RESOLVED = "RESOLVED" ABANDONED = "ABANDONED" class ChatIntentType(str, Enum): SR_CREATE = "SR_CREATE" # SR 생성 요청 SR_QUERY = "SR_QUERY" # SR 조회 DEPLOY_REQUEST = "DEPLOY_REQUEST" # 배포 요청 INCIDENT_REPORT = "INCIDENT_REPORT" # 인시던트 보고 GENERAL_INQUIRY = "GENERAL_INQUIRY" # 일반 문의 CLARIFICATION = "CLARIFICATION" # 추가 정보 응답 UNKNOWN = "UNKNOWN" class ChatSession(Base): """챗봇 대화 세션.""" __tablename__ = "tb_chat_session" id = Column(Integer, primary_key=True, index=True) session_key = Column(String(64), unique=True, nullable=False, index=True) user_id = Column(Integer, ForeignKey("tb_user.id"), nullable=True) username = Column(String(50), nullable=True) status = Column(String(20), default="ACTIVE") context_json = Column(Text, nullable=True) # 수집된 컨텍스트 JSON created_sr_id = Column(String(20), nullable=True) # 자동 생성된 SR ID created_at = Column(DateTime, default=func.now()) updated_at = Column(DateTime, default=func.now(), onupdate=func.now()) messages = relationship("ChatMessage", back_populates="session", cascade="all, delete-orphan", order_by="ChatMessage.created_at") class ChatMessage(Base): """챗봇 대화 메시지.""" __tablename__ = "tb_chat_message" id = Column(Integer, primary_key=True, index=True) session_id = Column(Integer, ForeignKey("tb_chat_session.id"), nullable=False) role = Column(String(10), nullable=False) # user | assistant | system content = Column(Text, nullable=False) intent = Column(String(30), nullable=True) # 감지된 인텐트 entities_json = Column(Text, nullable=True) # 추출된 엔티티 JSON confidence = Column(Float, nullable=True) # 인텐트 확신도 0.0~1.0 created_at = Column(DateTime, default=func.now()) session = relationship("ChatSession", back_populates="messages") # ── B-2 Pydantic schemas ──────────────────────────────────────────────────── class ChatMessageRequest(BaseModel): """챗봇 메시지 전송 요청.""" message: str session_key: Optional[str] = None # None이면 새 세션 생성 username: Optional[str] = None class ChatMessageOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int role: str content: str intent: Optional[str] entities_json: Optional[str] confidence: Optional[float] created_at: datetime class ChatSessionOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int session_key: str username: Optional[str] status: str context_json: Optional[str] created_sr_id: Optional[str] created_at: datetime updated_at: datetime messages: Optional[List[ChatMessageOut]] = None class ChatResponse(BaseModel): """챗봇 응답 전체.""" session_key: str reply: str intent: str entities: dict action_taken: bool created_sr_id: Optional[str] = None needs_clarification: bool = False clarification_prompt: Optional[str] = None suggestions: Optional[List[str]] = None # ============================================================================== # B-5: 멀티 에이전트 협업 오케스트레이션 모델 # ============================================================================== class WorkflowStatus(str, Enum): PENDING = "PENDING" RUNNING = "RUNNING" WAITING = "WAITING" # 승인 대기 등 외부 이벤트 대기 COMPLETED = "COMPLETED" FAILED = "FAILED" CANCELLED = "CANCELLED" class WorkflowType(str, Enum): SR_TO_DEPLOY = "SR_TO_DEPLOY" # SR 접수 → 코드 리뷰 → 배포 INCIDENT_RESP = "INCIDENT_RESP" # 인시던트 대응 → RCA → 복구 CODE_REVIEW = "CODE_REVIEW" # 코드 리뷰 → 취약점 스캔 → 보고 MAINTENANCE = "MAINTENANCE" # 유지보수 → 변경 관리 CUSTOM = "CUSTOM" class WorkflowInstance(Base): """워크플로우 실행 인스턴스.""" __tablename__ = "tb_workflow_instance" id = Column(Integer, primary_key=True, index=True) workflow_type = Column(String(30), nullable=False) status = Column(String(20), default="PENDING") title = Column(String(200), nullable=False) sr_id = Column(String(20), nullable=True) project_id = Column(Integer, ForeignKey("tb_project.id"), nullable=True) triggered_by = Column(String(100), nullable=True) context_json = Column(Text, nullable=True) # 워크플로우 컨텍스트 JSON current_step = Column(Integer, default=0) total_steps = Column(Integer, default=0) progress_pct = Column(Integer, default=0) # 0~100 result_json = Column(Text, nullable=True) # 최종 결과 error_msg = Column(Text, nullable=True) started_at = Column(DateTime, nullable=True) completed_at = Column(DateTime, nullable=True) created_at = Column(DateTime, default=func.now()) steps = relationship("WorkflowStep", back_populates="instance", cascade="all, delete-orphan", order_by="WorkflowStep.step_order") class WorkflowStep(Base): """워크플로우 단계.""" __tablename__ = "tb_workflow_step" id = Column(Integer, primary_key=True, index=True) instance_id = Column(Integer, ForeignKey("tb_workflow_instance.id"), nullable=False) step_order = Column(Integer, nullable=False) agent_name = Column(String(50), nullable=False) # sr-manager, code-reviewer 등 action = Column(String(100), nullable=False) # 수행할 액션 status = Column(String(20), default="PENDING") input_json = Column(Text, nullable=True) # 입력 파라미터 output_json = Column(Text, nullable=True) # 출력 결과 error_msg = Column(Text, nullable=True) retry_count = Column(Integer, default=0) started_at = Column(DateTime, nullable=True) completed_at = Column(DateTime, nullable=True) duration_sec = Column(Integer, nullable=True) instance = relationship("WorkflowInstance", back_populates="steps") # ── B-5 Pydantic schemas ──────────────────────────────────────────────────── class WorkflowStepOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int step_order: int agent_name: str action: str status: str input_json: Optional[str] output_json: Optional[str] error_msg: Optional[str] retry_count: int started_at: Optional[datetime] completed_at: Optional[datetime] duration_sec: Optional[int] class WorkflowInstanceOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int workflow_type: str status: str title: str sr_id: Optional[str] project_id: Optional[int] triggered_by: Optional[str] current_step: int total_steps: int progress_pct: int result_json: Optional[str] error_msg: Optional[str] started_at: Optional[datetime] completed_at: Optional[datetime] created_at: datetime steps: Optional[List[WorkflowStepOut]] = None class WorkflowCreateRequest(BaseModel): workflow_type: str = "SR_TO_DEPLOY" title: str sr_id: Optional[str] = None project_id: Optional[int] = None context: Optional[dict] = None # 초기 컨텍스트 # ════════════════════════════════════════════════════════════════════════════════ # C-1: CMDB 확장 — CI 관리, 관계, 변경 이력 # ════════════════════════════════════════════════════════════════════════════════ class CIStatus(str, Enum): PLANNED = "PLANNED" ACTIVE = "ACTIVE" INACTIVE = "INACTIVE" RETIRED = "RETIRED" DISPOSED = "DISPOSED" class CIType(str, Enum): SERVER = "SERVER" NETWORK = "NETWORK" STORAGE = "STORAGE" SOFTWARE = "SOFTWARE" SERVICE = "SERVICE" DATABASE = "DATABASE" MIDDLEWARE = "MIDDLEWARE" SECURITY = "SECURITY" OTHER = "OTHER" class CIRelationType(str, Enum): DEPENDS_ON = "DEPENDS_ON" # A → B: A는 B에 의존 PART_OF = "PART_OF" # A는 B의 구성 요소 HOSTED_ON = "HOSTED_ON" # A는 B 위에서 실행 CONNECTS_TO = "CONNECTS_TO" # A↔B: 네트워크 연결 BACKS_UP = "BACKS_UP" # A가 B를 백업 MONITORS = "MONITORS" # A가 B를 모니터링 REPLACED_BY = "REPLACED_BY" # A는 B로 교체됨 class CIChangeType(str, Enum): CREATE = "CREATE" UPDATE = "UPDATE" STATUS_CHANGE = "STATUS_CHANGE" RETIRE = "RETIRE" RELATION_ADD = "RELATION_ADD" RELATION_DEL = "RELATION_DEL" # ── ORM ────────────────────────────────────────────────────────────────────── class ConfigItem(Base): """형상 관리 항목 (CI).""" __tablename__ = "tb_ci" id = Column(Integer, primary_key=True, index=True) ci_id = Column(String(30), unique=True, nullable=False, index=True) # CI-YYYYMMDD-NNNN name = Column(String(200), nullable=False) ci_type = Column(String(20), nullable=False, default="SERVER") category = Column(String(50)) # 세분류 (예: WEB서버, WAS서버) status = Column(String(20), default="ACTIVE") version = Column(String(50)) # 소프트웨어 버전 등 owner = Column(String(100)) # 담당자 사번/이름 location = Column(String(200)) # 물리적 위치 / 데이터센터 랙 vendor = Column(String(100)) model = Column(String(100)) serial_number = Column(String(100)) install_date = Column(Date) retire_date = Column(Date) linked_server_id = Column(Integer, ForeignKey("tb_server_info.id"), nullable=True) inst_id = Column(Integer, ForeignKey("tb_inst_meta.id"), nullable=True) sr_id = Column(String(20)) # 등록/변경 관련 SR ID attributes_json = Column(Text) # 유연한 추가 속성 (JSON) description = Column(Text) created_by = Column(String(50)) created_at = Column(DateTime, default=func.now()) updated_at = Column(DateTime, default=func.now(), onupdate=func.now()) # relationships relations_from = relationship( "CIRelation", foreign_keys="CIRelation.from_ci_id", back_populates="from_ci", lazy="select", ) relations_to = relationship( "CIRelation", foreign_keys="CIRelation.to_ci_id", back_populates="to_ci", lazy="select", ) change_logs = relationship("CIChangeLog", back_populates="ci", lazy="select") class CIRelation(Base): """CI 간 관계.""" __tablename__ = "tb_ci_relation" id = Column(Integer, primary_key=True, index=True) from_ci_id = Column(Integer, ForeignKey("tb_ci.id"), nullable=False, index=True) to_ci_id = Column(Integer, ForeignKey("tb_ci.id"), nullable=False, index=True) relation_type = Column(String(30), nullable=False) # CIRelationType description = Column(String(500)) created_by = Column(String(50)) created_at = Column(DateTime, default=func.now()) from_ci = relationship("ConfigItem", foreign_keys=[from_ci_id], back_populates="relations_from") to_ci = relationship("ConfigItem", foreign_keys=[to_ci_id], back_populates="relations_to") class CIChangeLog(Base): """CI 변경 이력 (불변 감사 로그).""" __tablename__ = "tb_ci_change_log" id = Column(Integer, primary_key=True, index=True) ci_id_fk = Column(Integer, ForeignKey("tb_ci.id"), nullable=False, index=True) ci_id_str = Column(String(30)) # ci_id 문자열 (CI 삭제 후에도 조회 가능) change_type = Column(String(30), nullable=False) # CIChangeType field_name = Column(String(100)) # 변경된 필드명 (UPDATE 시) old_value = Column(Text) new_value = Column(Text) changed_by = Column(String(50)) changed_at = Column(DateTime, default=func.now()) sr_id = Column(String(20)) # 변경 관련 SR note = Column(String(500)) ci = relationship("ConfigItem", back_populates="change_logs") # ── Pydantic Schemas ───────────────────────────────────────────────────────── class CIRelationOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int from_ci_id: int to_ci_id: int relation_type: str description: Optional[str] created_by: Optional[str] created_at: datetime class CIChangeLogOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int ci_id_str: Optional[str] change_type: str field_name: Optional[str] old_value: Optional[str] new_value: Optional[str] changed_by: Optional[str] changed_at: datetime sr_id: Optional[str] note: Optional[str] class ConfigItemOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int ci_id: str name: str ci_type: str category: Optional[str] status: str version: Optional[str] owner: Optional[str] location: Optional[str] vendor: Optional[str] model: Optional[str] serial_number: Optional[str] install_date: Optional[date] retire_date: Optional[date] linked_server_id: Optional[int] inst_id: Optional[int] sr_id: Optional[str] description: Optional[str] created_by: Optional[str] created_at: datetime updated_at: datetime class ConfigItemCreate(BaseModel): name: str ci_type: str = "SERVER" category: Optional[str] = None status: str = "ACTIVE" version: Optional[str] = None owner: Optional[str] = None location: Optional[str] = None vendor: Optional[str] = None model: Optional[str] = None serial_number: Optional[str] = None install_date: Optional[date] = None retire_date: Optional[date] = None linked_server_id: Optional[int] = None inst_id: Optional[int] = None sr_id: Optional[str] = None attributes: Optional[dict] = None description: Optional[str] = None class ConfigItemUpdate(BaseModel): name: Optional[str] = None ci_type: Optional[str] = None category: Optional[str] = None status: Optional[str] = None version: Optional[str] = None owner: Optional[str] = None location: Optional[str] = None vendor: Optional[str] = None model: Optional[str] = None serial_number: Optional[str] = None install_date: Optional[date] = None retire_date: Optional[date] = None linked_server_id: Optional[int] = None sr_id: Optional[str] = None attributes: Optional[dict] = None description: Optional[str] = None class CIRelationCreate(BaseModel): from_ci_id: int to_ci_id: int relation_type: str description: Optional[str] = None # ════════════════════════════════════════════════════════════════════════════════ # C-2: 변경 관리 CAB (Change Advisory Board) # ════════════════════════════════════════════════════════════════════════════════ class RFCStatus(str, Enum): DRAFT = "DRAFT" # 초안 SUBMITTED = "SUBMITTED" # 제출됨 (CAB 검토 대기) IN_REVIEW = "IN_REVIEW" # CAB 검토 중 APPROVED = "APPROVED" # CAB 승인 REJECTED = "REJECTED" # CAB 거부 SCHEDULED = "SCHEDULED" # 일정 확정 IN_PROGRESS = "IN_PROGRESS" # 변경 진행 중 COMPLETED = "COMPLETED" # 변경 완료 FAILED = "FAILED" # 변경 실패/롤백 WITHDRAWN = "WITHDRAWN" # 철회 class ChangeType(str, Enum): STANDARD = "STANDARD" # 표준 변경 (사전 승인) NORMAL = "NORMAL" # 일반 변경 (CAB 필요) EMERGENCY = "EMERGENCY" # 긴급 변경 (e-CAB) MAJOR = "MAJOR" # 대규모 변경 class ChangeRisk(str, Enum): LOW = "LOW" MEDIUM = "MEDIUM" HIGH = "HIGH" CRITICAL = "CRITICAL" class CABVoteResult(str, Enum): APPROVE = "APPROVE" REJECT = "REJECT" ABSTAIN = "ABSTAIN" DEFER = "DEFER" # ── ORM ────────────────────────────────────────────────────────────────────── class RFChange(Base): """변경 요청서 (Request for Change).""" __tablename__ = "tb_rfc" id = Column(Integer, primary_key=True, index=True) rfc_id = Column(String(30), unique=True, nullable=False, index=True) # RFC-YYYYMMDD-NNNN title = Column(String(300), nullable=False) description = Column(Text) change_type = Column(String(20), default="NORMAL") # ChangeType risk_level = Column(String(20), default="MEDIUM") # ChangeRisk status = Column(String(20), default="DRAFT") # RFCStatus priority = Column(String(20), default="MEDIUM") # 요청자 / 담당자 requester_id = Column(Integer, ForeignKey("tb_user.id"), nullable=True) assignee_id = Column(Integer, ForeignKey("tb_user.id"), nullable=True) # 일정 planned_start = Column(DateTime) planned_end = Column(DateTime) actual_start = Column(DateTime) actual_end = Column(DateTime) freeze_exempt = Column(Boolean, default=False) # 동결 기간 예외 # 변경 내용 change_plan = Column(Text) # 변경 계획 상세 rollback_plan = Column(Text) # 롤백 계획 test_plan = Column(Text) # 테스트 계획 impact_analysis = Column(Text) # 영향도 분석 # 연관 정보 sr_id = Column(String(20)) # 관련 SR ci_ids_json = Column(Text) # 영향받는 CI ID 목록 (JSON) # 결과 result_summary = Column(Text) error_msg = Column(Text) # 감사 created_by = Column(String(50)) created_at = Column(DateTime, default=func.now()) updated_at = Column(DateTime, default=func.now(), onupdate=func.now()) submitted_at = Column(DateTime) approved_at = Column(DateTime) completed_at = Column(DateTime) # relationships cab_votes = relationship("CABVote", back_populates="rfc", lazy="select") freeze_checks = relationship("FreezeWindow", back_populates="rfc", lazy="select", uselist=False) class CABVote(Base): """CAB 위원 투표 기록.""" __tablename__ = "tb_cab_vote" id = Column(Integer, primary_key=True, index=True) rfc_id_fk = Column(Integer, ForeignKey("tb_rfc.id"), nullable=False, index=True) voter_id = Column(Integer, ForeignKey("tb_user.id"), nullable=True) voter_name = Column(String(100)) vote = Column(String(20), nullable=False) # CABVoteResult comment = Column(Text) voted_at = Column(DateTime, default=func.now()) is_final = Column(Boolean, default=False) # 최종 결정권자 여부 rfc = relationship("RFChange", back_populates="cab_votes") class FreezeWindow(Base): """변경 동결 기간 (Freeze Window).""" __tablename__ = "tb_freeze_window" id = Column(Integer, primary_key=True, index=True) rfc_id_fk = Column(Integer, ForeignKey("tb_rfc.id"), nullable=True) name = Column(String(200), nullable=False) # 동결 사유 (예: 연말 결산) start_dt = Column(DateTime, nullable=False) end_dt = Column(DateTime, nullable=False) scope = Column(String(50), default="ALL") # ALL / CRITICAL / NORMAL reason = Column(Text) created_by = Column(String(50)) created_at = Column(DateTime, default=func.now()) is_active = Column(Boolean, default=True) rfc = relationship("RFChange", back_populates="freeze_checks") # ── Pydantic Schemas ───────────────────────────────────────────────────────── class CABVoteOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int rfc_id_fk: int voter_name: Optional[str] vote: str comment: Optional[str] voted_at: datetime is_final: bool class RFChangeOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int rfc_id: str title: str change_type: str risk_level: str status: str priority: str planned_start: Optional[datetime] planned_end: Optional[datetime] actual_start: Optional[datetime] actual_end: Optional[datetime] freeze_exempt: bool sr_id: Optional[str] created_by: Optional[str] created_at: datetime updated_at: datetime submitted_at: Optional[datetime] approved_at: Optional[datetime] completed_at: Optional[datetime] class RFChangeCreate(BaseModel): title: str description: Optional[str] = None change_type: str = "NORMAL" risk_level: str = "MEDIUM" priority: str = "MEDIUM" planned_start: Optional[datetime] = None planned_end: Optional[datetime] = None freeze_exempt: bool = False change_plan: Optional[str] = None rollback_plan: Optional[str] = None test_plan: Optional[str] = None impact_analysis: Optional[str] = None sr_id: Optional[str] = None ci_ids: Optional[List[int]] = None class RFChangeUpdate(BaseModel): title: Optional[str] = None description: Optional[str] = None change_type: Optional[str] = None risk_level: Optional[str] = None priority: Optional[str] = None planned_start: Optional[datetime] = None planned_end: Optional[datetime] = None freeze_exempt: Optional[bool] = None change_plan: Optional[str] = None rollback_plan: Optional[str] = None test_plan: Optional[str] = None impact_analysis: Optional[str] = None ci_ids: Optional[List[int]] = None class CABVoteCreate(BaseModel): vote: str # APPROVE / REJECT / ABSTAIN / DEFER comment: Optional[str] = None is_final: bool = False class FreezeWindowOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int name: str start_dt: datetime end_dt: datetime scope: str reason: Optional[str] created_by: Optional[str] is_active: bool class FreezeWindowCreate(BaseModel): name: str start_dt: datetime end_dt: datetime scope: str = "ALL" reason: Optional[str] = None # ════════════════════════════════════════════════════════════════════════════════ # C-3: Problem Management # ════════════════════════════════════════════════════════════════════════════════ class ProblemStatus(str, Enum): OPEN = "OPEN" INVESTIGATING = "INVESTIGATING" RCA_DONE = "RCA_DONE" # 근본 원인 분석 완료 WORKAROUND = "WORKAROUND" # 임시 해결책 적용 RFC_RAISED = "RFC_RAISED" # 영구 수정 RFC 등록됨 RESOLVED = "RESOLVED" CLOSED = "CLOSED" class ProblemCategory(str, Enum): HARDWARE = "HARDWARE" SOFTWARE = "SOFTWARE" NETWORK = "NETWORK" PROCESS = "PROCESS" HUMAN = "HUMAN" VENDOR = "VENDOR" UNKNOWN = "UNKNOWN" class ProblemRecord(Base): """문제 레코드 (Problem Record).""" __tablename__ = "tb_problem" id = Column(Integer, primary_key=True, index=True) problem_id = Column(String(30), unique=True, nullable=False, index=True) # PRB-YYYYMMDD-NNNN title = Column(String(300), nullable=False) description = Column(Text) status = Column(String(30), default="OPEN") priority = Column(String(20), default="MEDIUM") category = Column(String(30), default="UNKNOWN") # 근본 원인 분석 root_cause = Column(Text) # RCA 결과 workaround = Column(Text) # 임시 해결 방법 permanent_fix = Column(Text) # 영구 수정 계획 known_error = Column(Boolean, default=False) # Known Error DB 등재 여부 # 연관 정보 related_sr_ids = Column(Text) # 관련 SR ID 목록 (JSON) rfc_id = Column(String(30)) # 영구 수정 RFC ci_id = Column(String(30)) # 영향 CI # 담당자 assignee_id = Column(Integer, ForeignKey("tb_user.id"), nullable=True) created_by = Column(String(50)) # 이력 created_at = Column(DateTime, default=func.now()) updated_at = Column(DateTime, default=func.now(), onupdate=func.now()) rca_completed_at = Column(DateTime) resolved_at = Column(DateTime) closed_at = Column(DateTime) # 영향도 affected_users = Column(Integer, default=0) incident_count = Column(Integer, default=0) # 관련 인시던트 수 total_downtime_min = Column(Integer, default=0) # 누적 다운타임 (분) class ProblemNote(Base): """문제 레코드 활동 노트.""" __tablename__ = "tb_problem_note" id = Column(Integer, primary_key=True, index=True) problem_id_fk = Column(Integer, ForeignKey("tb_problem.id"), nullable=False, index=True) note_type = Column(String(30), default="UPDATE") # UPDATE / RCA / WORKAROUND / RESOLUTION content = Column(Text, nullable=False) author = Column(String(100)) created_at = Column(DateTime, default=func.now()) # ── Pydantic ───────────────────────────────────────────────────────────────── class ProblemRecordOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int problem_id: str title: str status: str priority: str category: str root_cause: Optional[str] workaround: Optional[str] known_error: bool rfc_id: Optional[str] ci_id: Optional[str] created_by: Optional[str] created_at: datetime updated_at: datetime resolved_at: Optional[datetime] affected_users: int incident_count: int total_downtime_min: int class ProblemRecordCreate(BaseModel): title: str description: Optional[str] = None priority: str = "MEDIUM" category: str = "UNKNOWN" related_sr_ids: Optional[List[str]] = None ci_id: Optional[str] = None affected_users: int = 0 incident_count: int = 0 total_downtime_min: int = 0 class ProblemNoteCreate(BaseModel): note_type: str = "UPDATE" # UPDATE / RCA / WORKAROUND / RESOLUTION content: str # ════════════════════════════════════════════════════════════════════════════════ # C-4: 용량 관리 대시보드 # ════════════════════════════════════════════════════════════════════════════════ class CapacityStatus(str, Enum): NORMAL = "NORMAL" WARNING = "WARNING" CRITICAL = "CRITICAL" OVERLOAD = "OVERLOAD" class CapacityPlan(Base): """용량 계획 레코드.""" __tablename__ = "tb_capacity_plan" id = Column(Integer, primary_key=True, index=True) source = Column(String(100), nullable=False) # 서버/서비스명 metric_type = Column(String(30), nullable=False) # CPU_USAGE / MEMORY_USAGE / DISK_USAGE current_value = Column(Float) capacity_max = Column(Float) # 최대 용량 threshold_warn = Column(Float) threshold_crit = Column(Float) status = Column(String(20), default="NORMAL") growth_rate = Column(Float) # 월 증가율 (%) forecast_3m = Column(Float) # 3개월 후 예측값 forecast_6m = Column(Float) # 6개월 후 예측값 forecast_12m = Column(Float) # 12개월 후 예측값 expansion_needed_at = Column(DateTime) # 확장 필요 예상 시점 note = Column(Text) inst_id = Column(Integer, ForeignKey("tb_inst_meta.id"), nullable=True) updated_by = Column(String(50)) updated_at = Column(DateTime, default=func.now(), onupdate=func.now()) created_at = Column(DateTime, default=func.now()) class CapacityPlanOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int source: str metric_type: str current_value: Optional[float] capacity_max: Optional[float] threshold_warn: Optional[float] threshold_crit: Optional[float] status: str growth_rate: Optional[float] forecast_3m: Optional[float] forecast_6m: Optional[float] forecast_12m: Optional[float] expansion_needed_at: Optional[datetime] note: Optional[str] updated_at: datetime class CapacityPlanCreate(BaseModel): source: str metric_type: str current_value: Optional[float] = None capacity_max: Optional[float] = None threshold_warn: Optional[float] = None threshold_crit: Optional[float] = None growth_rate: Optional[float] = None note: Optional[str] = None inst_id: Optional[int] = None # ════════════════════════════════════════════════════════════════════════════════ # C-5: 서비스 카탈로그 # ════════════════════════════════════════════════════════════════════════════════ class ServiceStatus(str, Enum): ACTIVE = "ACTIVE" DRAFT = "DRAFT" DEPRECATED = "DEPRECATED" RETIRED = "RETIRED" class ServiceItem(Base): """서비스 카탈로그 항목.""" __tablename__ = "tb_service_catalog" id = Column(Integer, primary_key=True, index=True) service_id = Column(String(30), unique=True, nullable=False, index=True) # SVC-NNNN name = Column(String(200), nullable=False) category = Column(String(50)) # 카테고리 (배포, DB관리, 보안 등) description = Column(Text) status = Column(String(20), default="ACTIVE") # SLA 정보 sla_response_h = Column(Float, default=8.0) # 응답 SLA (시간) sla_resolve_h = Column(Float, default=24.0) # 해결 SLA (시간) sla_availability = Column(Float, default=99.5) # 가용성 SLA (%) # 요청 안내 request_template = Column(Text) # SR 요청 템플릿 approval_required = Column(Boolean, default=False) # 승인 필요 여부 auto_assignee = Column(String(50)) # 자동 배정 담당자 사번 estimated_hours = Column(Float) # 예상 처리 시간 # 비용 charge_per_request = Column(Float, default=0.0) # 요청당 비용 (0=무료) currency = Column(String(10), default="KRW") # 통계 request_count = Column(Integer, default=0) avg_resolve_h = Column(Float) # 실제 평균 해결 시간 satisfaction = Column(Float) # 평균 만족도 (1-5) # 메타 owner = Column(String(100)) version = Column(String(20), default="1.0") tags = Column(String(500)) created_by = Column(String(50)) created_at = Column(DateTime, default=func.now()) updated_at = Column(DateTime, default=func.now(), onupdate=func.now()) class ServiceItemOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int service_id: str name: str category: Optional[str] description: Optional[str] status: str sla_response_h: float sla_resolve_h: float sla_availability: float approval_required: bool estimated_hours: Optional[float] charge_per_request: float request_count: int avg_resolve_h: Optional[float] satisfaction: Optional[float] owner: Optional[str] version: str tags: Optional[str] created_at: datetime class ServiceItemCreate(BaseModel): name: str category: Optional[str] = None description: Optional[str] = None status: str = "ACTIVE" sla_response_h: float = 8.0 sla_resolve_h: float = 24.0 sla_availability: float = 99.5 request_template: Optional[str] = None approval_required: bool = False auto_assignee: Optional[str] = None estimated_hours: Optional[float] = None charge_per_request: float = 0.0 owner: Optional[str] = None tags: Optional[str] = None class ServiceItemUpdate(BaseModel): name: Optional[str] = None category: Optional[str] = None description: Optional[str] = None status: Optional[str] = None sla_response_h: Optional[float] = None sla_resolve_h: Optional[float] = None approval_required: Optional[bool] = None estimated_hours: Optional[float] = None owner: Optional[str] = None tags: Optional[str] = None # ── 자율 운영 자동처리 ──────────────────────────────────────────────────────── class AutoActionStatus(str, Enum): AUTO_DONE = "AUTO_DONE" # 자동 완료 PENDING_APPROVAL = "PENDING_APPROVAL" # 승인 대기 APPROVED = "APPROVED" # 승인됨 REJECTED = "REJECTED" # 거부됨 EXPIRED = "EXPIRED" # 시간 만료 FAILED = "FAILED" # 실행 실패 class AutoAction(Base): """자율 운영 작업 이력 및 승인 큐.""" __tablename__ = "tb_auto_action" id = Column(Integer, primary_key=True, index=True) action_id = Column(String(30), unique=True, nullable=False, index=True) # ACT-XXXXXXXX action_type = Column(String(50), nullable=False) # sr_classify | kb_answer | server_restart | deploy_prd | dr_failover 등 description = Column(Text) target = Column(String(200)) # 대상 서버/SR/서비스 risk_level = Column(String(20), default="HIGH") # LOW | MEDIUM | HIGH | CRITICAL status = Column(String(30), default=AutoActionStatus.PENDING_APPROVAL) payload = Column(JSON, default=dict) # 실행에 필요한 파라미터 result = Column(JSON, default=dict) # 실행 결과 requested_by = Column(String(100)) approved_by = Column(String(100)) approved_at = Column(DateTime) comment = Column(Text) processed_at = Column(DateTime) expires_at = Column(DateTime) # 승인 만료 시각 created_at = Column(DateTime, default=func.now()) # ── DR 자동화 ────────────────────────────────────────────────────────────────── class DRScenario(Base): """DR 시나리오 정의.""" __tablename__ = "tb_dr_scenario" id = Column(Integer, primary_key=True, index=True) name = Column(String(100), nullable=False) scenario_type = Column(String(30), default="SERVER_FAILURE") # SITE_FAILURE | SERVER_FAILURE | DATA_CORRUPTION primary_server_id = Column(Integer, ForeignKey("tb_server_info.id"), nullable=True) standby_server_id = Column(Integer, ForeignKey("tb_server_info.id"), nullable=True) rto_minutes = Column(Integer, default=240) # 목표 복구 시간 (분) rpo_minutes = Column(Integer, default=60) # 목표 복구 시점 (분) failover_steps = Column(JSON, default=list) # 실행 단계 목록 healthcheck_url = Column(String(255)) last_test_at = Column(DateTime) last_test_result = Column(String(20)) # PASS | FAIL | PARTIAL is_active = Column(Boolean, default=True) created_at = Column(DateTime, default=func.now()) tests = relationship("DRTest", back_populates="scenario", cascade="all, delete-orphan") class DRTest(Base): """DR 복구 테스트 실행 기록.""" __tablename__ = "tb_dr_test" id = Column(Integer, primary_key=True, index=True) scenario_id = Column(Integer, ForeignKey("tb_dr_scenario.id"), nullable=False) test_type = Column(String(20), default="RECOVERY") # BACKUP_VERIFY | RECOVERY | FAILOVER_SIM status = Column(String(20), default="RUNNING") # RUNNING | PASS | FAIL | PARTIAL rto_actual = Column(Integer) # 실제 복구 시간 (분) rpo_actual = Column(Integer) # 실제 복구 시점 (분) result_detail = Column(JSON, default=dict) started_at = Column(DateTime, default=func.now()) completed_at = Column(DateTime) triggered_by = Column(String(100)) scenario = relationship("DRScenario", back_populates="tests") # ── 네트워크 장비 관리 ───────────────────────────────────────────────────────── class NetworkDevice(Base): """네트워크 장비 (스위치/라우터/방화벽/LB).""" __tablename__ = "tb_network_device" id = Column(Integer, primary_key=True, index=True) device_name = Column(String(100), nullable=False) device_type = Column(String(30)) # SWITCH | ROUTER | FIREWALL | LOAD_BALANCER vendor = Column(String(30)) # CISCO | HUAWEI | JUNIPER | PIOLINK | SECUI | RADWARE model = Column(String(100)) os_type = Column(String(30)) # cisco_ios | huawei_vrp | junos | linux ip_addr = Column(String(45)) # NOT exposed in API ssh_user = Column(String(50)) # NOT exposed ssh_pw_enc = Column(Text) # AES-256-GCM, NEVER exposed ssh_port = Column(Integer, default=22) location = Column(String(200)) inst_id = Column(Integer, ForeignKey("tb_inst_meta.id"), nullable=True) is_active = Column(Boolean, default=True) last_backup_at = Column(DateTime) created_at = Column(DateTime, default=func.now()) backups = relationship("NetworkConfigBackup", back_populates="device", cascade="all, delete-orphan") class NetworkConfigBackup(Base): """네트워크 장비 설정 백업.""" __tablename__ = "tb_network_backup" id = Column(Integer, primary_key=True, index=True) device_id = Column(Integer, ForeignKey("tb_network_device.id"), nullable=False) config_text = Column(Text) # 설정 전문 config_hash = Column(String(64)) # SHA-256 backup_type = Column(String(20), default="MANUAL") # SCHEDULED | MANUAL | PRE_CHANGE backed_up_at = Column(DateTime, default=func.now()) backed_up_by = Column(String(100)) device = relationship("NetworkDevice", back_populates="backups") # ── CSAP 공공기관 보안 점검 ──────────────────────────────────────────────────── class CSAPCheckResult(Base): """CSAP/ISMS-P 점검 결과.""" __tablename__ = "tb_csap_result" id = Column(Integer, primary_key=True, index=True) scan_id = Column(String(50), nullable=False, index=True) # 배치 ID: CSAP-YYYYMMDD-HHMMSS inst_id = Column(Integer, ForeignKey("tb_inst_meta.id"), nullable=True) item_id = Column(String(20), nullable=False) # M-01, T-03 등 category = Column(String(20)) # 관리적 | 기술적 | 물리적 | 운영 item_name = Column(String(200)) severity = Column(String(20)) # HIGH | MEDIUM | LOW status = Column(String(20)) # PASS | FAIL | PARTIAL | MANUAL_REQUIRED | N_A finding = Column(Text) evidence = Column(JSON, default=dict) # 자동 수집 증적 (마스킹) recommendation = Column(Text) scanned_at = Column(DateTime, default=func.now()) # ── 개방망 API Key ───────────────────────────────────────────────────────────── # ── 스크랩핑 ──────────────────────────────────────────────────────────────── class ScrapingTarget(Base): """스크랩 대상 URL 등록 테이블.""" __tablename__ = "tb_scraping_target" id = Column(Integer, primary_key=True, index=True) name = Column(String(100), nullable=False) url = Column(String(500), nullable=False) selector = Column(String(200)) # CSS 셀렉터 (옵션) schedule = Column(String(50)) # cron (옵션) is_active = Column(Boolean, default=True) last_scraped = Column(DateTime, nullable=True) note = Column(Text) created_by = Column(String(50)) created_at = Column(DateTime, default=func.now()) updated_at = Column(DateTime, default=func.now(), onupdate=func.now()) results = relationship("ScrapingResult", back_populates="target", cascade="all, delete-orphan") class ScrapingResult(Base): """스크랩 결과 저장 테이블.""" __tablename__ = "tb_scraping_result" id = Column(Integer, primary_key=True, index=True) target_id = Column(Integer, ForeignKey("tb_scraping_target.id"), nullable=True) title = Column(String(300)) content = Column(Text) # 정제된 본문 (HTML) plain_text = Column(Text) # 텍스트 본문 (검색용) url = Column(String(500), nullable=False) source_html = Column(Text) # 원본 HTML (원복용) status = Column(String(20), default="DRAFT") # DRAFT/PUBLISHED/DELETED/FAILED scraped_at = Column(DateTime, default=func.now()) published_at = Column(DateTime, nullable=True) deleted_at = Column(DateTime, nullable=True) published_by = Column(String(50), nullable=True) messenger_room = Column(String(50), default="ops") meta = Column(JSON, default=dict) error_msg = Column(Text, nullable=True) scraped_by = Column(String(50), default="system") created_at = Column(DateTime, default=func.now()) target = relationship("ScrapingTarget", back_populates="results") class ScrapingTargetOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int name: str url: str selector: Optional[str] schedule: Optional[str] is_active: bool last_scraped: Optional[datetime] note: Optional[str] created_by: Optional[str] created_at: datetime class ScrapingTargetCreate(BaseModel): name: str url: str selector: Optional[str] = None schedule: Optional[str] = None is_active: bool = True note: Optional[str] = None class ScrapingResultOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int target_id: Optional[int] title: Optional[str] plain_text: Optional[str] url: str status: str scraped_at: datetime published_at: Optional[datetime] deleted_at: Optional[datetime] published_by: Optional[str] messenger_room: Optional[str] meta: Optional[dict] error_msg: Optional[str] scraped_by: Optional[str] created_at: datetime class APIKey(Base): """외부 시스템 연동용 API Key (개방망 전용).""" __tablename__ = "tb_api_key" id = Column(Integer, primary_key=True, index=True) name = Column(String(100), nullable=False) # 키 이름 (ex: "카카오워크 봇") key_hash = Column(String(64), unique=True, nullable=False, index=True) # SHA-256 scopes = Column(String(200), default="read") # read,write,admin,webhook allowed_ips = Column(String(500), default="") # "1.2.3.4,5.6.7.8" 빈칸=전체허용 is_active = Column(Boolean, default=True) use_count = Column(Integer, default=0) last_used_at = Column(DateTime, nullable=True) expires_at = Column(DateTime, nullable=True) created_by = Column(String(50), nullable=True) created_at = Column(DateTime, default=func.now())