MCP OpenDART

import logging, datetime
from typing import Any, Optional, Annotated
from pydantic import Field
from mcp_opendart.server import mcp
from mcp.types import TextContent
from mcp_opendart.utils.ctx_helper import with_context, with_context_async, as_json_text
from mcp_opendart.registry.initialize_registry import initialize_registry

logger = logging.getLogger("mcp-opendart")
tool_registry = initialize_registry()

@mcp.tool(
    name="get_opendart_tool_info",
    description="기업의 심층 분석을 위해 mcp 도구의 이름을 정확히 입력하여 도구의 목적, 활용 흐름, 필수 파라미터, 연관 도구 등을 확인 후 진행, 예: get_opendart_tool_info(tool_name='get_corporation_info')",
    tags={"심층분석", "기업분석", "도구흐름"}
)
async def get_opendart_tool_info(
    tool_name: str,
) -> TextContent:
    tool = tool_registry.get_tool(tool_name)
    if not tool:
        return TextContent(type="text", text=f"❌ '{tool_name}'이라는 이름의 도구를 찾을 수 없습니다.")

    # 파라미터 설명 추출
    param_lines = []
    props = tool.parameters.get("properties", {})
    for param, meta in props.items():
        desc = meta.get("description", "")
        param_lines.append(f"- {param}: {desc}")

    # linked 도구 리스트
    linked = tool.linked_tools or []
    linked_str = ", ".join(linked) if linked else "(연관 도구 없음)"

    # 현재 날짜
    today = datetime.date.today().strftime("%Y-%m-%d")

    # 응답 생성
    text = f"""
        📌 {tool.name} ({tool.korean_name or '도구명 미정'})

        이 도구는 특정 기업의 "{tool.name}"을(를) 다음과 같은 파라미터를 이용하여 조회할 수 있습니다:
        {chr(10).join(param_lines)}

        날짜가 필요한 경우, 반드시 시스템 기준일({today})을 기준으로 조회를 시작해야 합니다.

        이 정보를 바탕으로, 사용자는 다음과 같은 흐름으로 도구를 활용할 수 있습니다:

        1. 먼저 get_corporation_code_by_name 도구를 통해 기업명을 고유번호로 변환합니다.
        2. {tool.name}을 사용해 관련 정보를 조회합니다.
        3. 연관된 다음 도구를 반드시 확인해야 합니다:
        {linked_str}

        이 도구는 정보 흐름의 중심 축으로 활용되며, 다른 도구들과 결합되어 기업 리스크 분석에 핵심적인 역할을 합니다.
        """

    return TextContent(type="text", text=text.strip())

@mcp.tool(
    name="get_corporation_code_by_name",
    description="기업명을 이용하여 기업 고유번호 조회, 공시조회를 위해 가장 먼저 실행하여 고유번호를 얻어야 함",
    tags={"기업검색", "고유번호", "기업기초정보", "기업식별"}
)
async def get_corporation_code_by_name(
    corp_name: Annotated[str, "기업명 (예: 삼성전자, 현대자동차)"],
) -> TextContent:
    result = await with_context_async(None, "get_corporation_code_by_name", lambda context: context.ds001.get_corporation_code_by_name(corp_name))
    return as_json_text(result)

@mcp.tool(
    name="get_disclosure_list",
    description="기업의 전체 공시 이력을 날짜별로 조회하여 경영활동, 재무현황, 지배구조 변화를 신속하게 파악. 여러 페이지가 있을 경우 페이지 번호를 바꿔 조회",
    tags={"공시", "목록", "접수내역", "이벤트탐지"}
)
def get_disclosure_list(
    corp_code: Annotated[str, "공시대상회사의 고유번호 (8자리)"],
    bgn_de: Annotated[str, "조회 시작일 (YYYYMMDD)"],
    end_de: Annotated[str, "조회 종료일 (YYYYMMDD)"],
    page_no: Annotated[int, "페이지 번호 (기본값: 1)"] = 1,
    pblntf_ty: Annotated[str, "공시유형 (A:정기공시, B:주요사항, C:발행, D:지분, E:기타 등)"] = "",
) -> TextContent:
    """
    공시 목록 조회 - 기업의 전체 공시 이력을 날짜별로 조회하여 경영활동, 재무현황, 지배구조 변화를 신속하게 파악
    
    응답에 total_count와 total_page 정보가 포함됩니다. 
    여러 페이지가 있을 경우 page_no 파라미터를 사용하여 다른 페이지를 조회하세요.
    예: page_no=2, page_no=3 등으로 순차적으로 조회

    Args:
        corp_code (str): 공시대상회사의 고유번호 (8자리)
        bgn_de (str): 조회 시작일 (YYYYMMDD)
        end_de (str): 조회 종료일 (YYYYMMDD)
        page_no (int): 페이지 번호 (기본값: 1) - total_page를 확인하여 필요한 페이지를 조회
        pblntf_ty (str): 공시유형 필터 (기본값: "" - 모든 공시유형)
            - "": 모든 공시유형 (기본값)
            - "A": 정기공시 (사업보고서, 반기보고서, 분기보고서, 반기보고서 등)
            - "B": 주요사항보고 (영업정지, 감자, 합병, 분할, 영업양수도 등)
            - "C": 발행공시 (신주인수권부사채, 전환사채, 신주발행 등)
            - "D": 지분공시 (대량보유상황보고서, 임원변경 등)
            - "E": 기타공시 (기업설명회, 투자설명서 등)
            - "F": 외부감사관련 (감사보고서, 회계법인 변경 등)
            - "G": 펀드공시 (펀드 관련 공시)
            - "H": 자산유동화 (자산유동화 관련 공시)
            - "I": 거래소공시 (거래소 관련 공시)
            - "J": 공정위공시 (공정거래위원회 관련 공시)

    공시유형 선택 가이드:
        - 사업보고서, 반기보고서, 분기보고서 조회 시: "A" 사용
        - 경영진 변경, 사업 구조 변경 등 중요 사항: "B" 사용  
        - 신주발행, 사채발행 등 자금조달: "C" 사용
        - 주주변동, 대주주 변경: "D" 사용
        - IR 활동, 투자설명회: "E" 사용

    참고: https://opendart.fss.or.kr/guide/detail.do?apiGrpCd=DS001&apiId=2019001
    """
    result = with_context(None, "get_disclosure_list", lambda context: context.ds001.get_disclosure_list(
        corp_code=corp_code,
        bgn_de=bgn_de,
        end_de=end_de,
        page_no=page_no,
        pblntf_ty=pblntf_ty
    ))
    return as_json_text(result)


@mcp.tool(
    name="get_corporation_info",
    description="대표자, 결산월, 상장상태 등 기업 기본 정보 기반 지배구조 및 공시 일정 분석",
    tags={"기업기초정보", "지배구조", "공시일정", "대표자분석", "가족경영"}
)
def get_corporation_info(
    corp_code: Annotated[str, "공시대상회사의 고유번호 (8자리)"],
) -> TextContent:
    """
    기업 개황정보 조회

    Args:
        corp_code (str): 공시대상회사의 고유번호(8자리)

    참고: https://opendart.fss.or.kr/guide/detail.do?apiGrpCd=DS001&apiId=2019002
    """
    result = with_context(None, "get_corporation_info", lambda context: context.ds001.get_corporation_info(corp_code))
    return as_json_text(result)


@mcp.tool(
    name="get_disclosure_document",
    description="공시서류 원본파일을 다운로드하고 자동으로 재무제표 주석을 추출하여 캐싱합니다. 재무제표 주석이 없으면 다른 보고서를 찾아보라는 안내를 제공합니다.",
    tags={"공시서류", "원본파일", "재무제표주석", "자동추출", "캐싱"}
)
def get_disclosure_document(
    rcp_no: Annotated[str, "공시서류의 접수번호 (14자리)"],
) -> TextContent:
    """
    공시서류 원본파일 다운로드 및 자동 재무제표 주석 추출
    
    Args:
        rcp_no (str): 공시서류의 접수번호 (14자리)
        
    Returns:
        TextContent: 공시서류 데이터 및 재무제표 주석 추출 결과
        
    Features:
        - XML 파일 다운로드 및 로컬 저장
        - 자동 재무제표 주석 추출 (연결재무제표 주석, 재무제표 주석)
        - 재무제표 주석이 없으면 다른 보고서를 찾아보라는 안내 제공
        - 구조화된 데이터 캐싱 (disclosure_cache 디렉토리)
        
    참고: https://opendart.fss.or.kr/guide/detail.do?apiGrpCd=DS001&apiId=2019003
    """
    # 1. 공시서류 다운로드
    result = with_context(None, "get_disclosure_document", lambda context: context.ds001.get_disclosure_document(rcp_no))
    
    # 2. 재무제표 주석 자동 추출 (다운로드 성공 시)
    if result.get("status_code") == "000" or result.get("status") == "000":
        try:
            financial_notes_result = with_context(None, "extract_financial_notes_document", 
                lambda context: context.ds001.extract_financial_notes_document(rcp_no))
            
            if financial_notes_result.get("status") == "000":
                result["financial_notes_extraction"] = {
                    "status": "success",
                    "message": "재무제표 주석이 성공적으로 추출되었습니다",
                    "summary": financial_notes_result.get("summary", {}),
                    "sections": financial_notes_result.get("sections", {}),
                    "save_path": financial_notes_result.get("save_path", ""),
                    "extraction_date": financial_notes_result.get("extraction_date", ""),
                    "final_rcp_no": financial_notes_result.get("rcp_no", rcp_no)
                }
            else:
                result["financial_notes_extraction"] = {
                    "status": "error",
                    "message": financial_notes_result.get("message", "재무제표 주석 추출 실패"),
                    "error": financial_notes_result.get("error", "알 수 없는 오류")
                }
        except Exception as e:
            result["financial_notes_extraction"] = {
                "status": "error",
                "message": "재무제표 주석 추출 중 오류 발생",
                "error": str(e)
            }
    
    return as_json_text(result)


@mcp.tool(
    name="search_financial_notes",
    description="추출된 재무제표 주석, 사업의 내용, 회사의 개요 데이터에서 테이블과 문단을 검색합니다. 연결재무제표 주석, 재무제표 주석, 사업의 내용, 회사의 개요에서 특정 키워드를 검색할 수 있습니다.",
    tags={"재무제표주석", "검색", "테이블검색", "문단검색", "키워드검색", "사업의내용", "회사의개요"}
)
def search_financial_notes(
    rcp_no: Annotated[str, Field(description="공시 접수번호 (예: 20241231000420). get_disclosure_list에서 조회 가능")],
    search_term: Annotated[str, Field(description="검색할 키워드 (예: CSM, 계약서비스마진, 보험계약자산)")],
    section_type: Annotated[str, Field(description="검색 섹션 (all: 전체, consolidated_notes: 연결재무제표 주석, separate_notes: 재무제표 주석, business_content: 사업의 내용, company_overview: 회사의 개요)")] = "all",
    search_in: Annotated[str, Field(description="검색 범위 (both: 테이블과 문단 모두, tables: 테이블만, paragraphs: 문단만)")] = "both",
    case_sensitive: Annotated[bool, Field(description="대소문자 구분 여부 (기본값: False)")] = False,
    max_results: Annotated[int, Field(description="최대 결과 수 (기본값: 50)")] = 50,
) -> TextContent:
    """
    재무제표 주석 데이터 검색
    
    Args:
        rcp_no (str): 공시서류의 접수번호 (14자리)
        search_term (str): 검색할 키워드
        section_type (str): 검색할 섹션 ("all", "consolidated_notes", "separate_notes", "business_content", "company_overview")
        search_in (str): 검색 대상 ("both", "tables", "paragraphs")
        case_sensitive (bool): 대소문자 구분 여부 (기본값: False)
        max_results (int): 최대 결과 수 (기본값: 50)
        
    Returns:
        TextContent: 검색 결과 (테이블 및 문단)
    """
    from pathlib import Path
    import json
    import re
    from difflib import SequenceMatcher
    
    def is_similar_match(search_term: str, content: str) -> bool:
        """유사도 20% 이상 매칭 확인"""
        # 정확한 매칭 먼저 확인
        if case_sensitive:
            if search_term in content:
                return True
        else:
            if search_term.lower() in content.lower():
                return True
        
        # 유사도 매칭 확인 (20% 이상)
        similarity = SequenceMatcher(None, search_term.lower(), content.lower()).ratio()
        return similarity >= 0.2
    
    try:
        # 캐시 디렉토리 확인
        project_root = Path(__file__).parent.parent.parent
        cache_path = project_root / 'mcp_opendart' / 'utils' / 'data' / 'disclosure_cache' / f'financial_notes_{rcp_no}'
        
        if not cache_path.exists():
            return as_json_text({
                "status": "404",
                "message": "재무제표 주석 데이터를 찾을 수 없습니다. 먼저 get_disclosure_document를 실행하여 데이터를 추출하세요.",
                "error": f"캐시 경로가 존재하지 않습니다: {cache_path}"
            })
        
        # 검색 결과 저장
        search_results = {
            "status": "000",
            "message": "정상",
            "rcp_no": rcp_no,
            "search_term": search_term,
            "section_type": section_type,
            "search_in": search_in,
            "results": {
                "tables": [],
                "paragraphs": []
            }
        }
        
        # 섹션별 검색 (자동 확장 포함)
        sections_to_search = []
        if section_type == "all":
            sections_to_search = ["consolidated_notes", "separate_notes", "business_content", "company_overview"]
        elif section_type in ["consolidated_notes", "separate_notes", "business_content", "company_overview"]:
            sections_to_search = [section_type]
        
        # 검색 결과 풍부화를 위한 자동 확장 로직
        enriched_sections = set(sections_to_search)
        
        # 매칭된 섹션의 상위 디렉토리 자동 포함
        for section in sections_to_search:
            if section == "business_content":
                # business_content의 모든 하위 섹션 자동 포함
                enriched_sections.add("business_content")
            elif section == "company_overview":
                # company_overview의 모든 하위 섹션 자동 포함
                enriched_sections.add("company_overview")
        
        sections_to_search = list(enriched_sections)
        
        for section_name in sections_to_search:
            section_path = cache_path / section_name
            if not section_path.exists():
                continue
            
            # 사업의 내용과 회사의 개요 섹션은 하위 구조가 다름
            if section_name in ["business_content", "company_overview"]:
                # 하위 섹션들 검색
                subsections_path = section_path / 'subsections'
                if subsections_path.exists():
                    for subsection_dir in subsections_path.iterdir():
                        if subsection_dir.is_dir():
                            # 하위 섹션 메타데이터 읽기
                            try:
                                with open(subsection_dir / 'metadata.json', 'r', encoding='utf-8') as f:
                                    subsection_meta = json.load(f)
                                subsection_name = subsection_meta.get('original_name', subsection_dir.name)
                            except:
                                subsection_name = subsection_dir.name
                            
                            # 🎯 섹션명 자체도 검색 대상에 포함
                            section_name_match = False
                            if case_sensitive:
                                section_name_match = search_term in subsection_name
                            else:
                                section_name_match = search_term.lower() in subsection_name.lower()
                            
                            # 🎯 섹션명이 매칭되면 해당 섹션의 모든 내용을 포함
                            if section_name_match:
                                # 섹션명 매칭 시 모든 테이블과 문단을 결과에 포함
                                if search_in in ["tables", "both"]:
                                    tables_path = subsection_dir / 'tables'
                                    if tables_path.exists():
                                        for table_file in tables_path.glob('*.json'):
                                            try:
                                                with open(table_file, 'r', encoding='utf-8') as f:
                                                    table_data = json.load(f)
                                                
                                                search_text = ' '.join(table_data.get('headers', []) + 
                                                                     [cell for row in table_data.get('rows', []) for cell in row])
                                                
                                                search_results["results"]["tables"].append({
                                                    "section": f"{section_name}/{subsection_name}",
                                                    "table_id": table_data.get('table_id'),
                                                    "file_path": str(table_file),
                                                    "headers": table_data.get('headers', []),
                                                    "row_count": len(table_data.get('rows', [])),
                                                    "matched_content": search_text,
                                                    "context_type": "사업정보" if section_name == "business_content" else "기본정보" if section_name == "company_overview" else "재무정보",
                                                    "match_type": "section_name"
                                                })
                                            except Exception as e:
                                                continue
                                
                                if search_in in ["paragraphs", "both"]:
                                    paragraphs_path = subsection_dir / 'paragraphs'
                                    if paragraphs_path.exists():
                                        for para_file in paragraphs_path.glob('*.json'):
                                            try:
                                                with open(para_file, 'r', encoding='utf-8') as f:
                                                    para_data = json.load(f)
                                                
                                                content = para_data.get('content', '')
                                                
                                                search_results["results"]["paragraphs"].append({
                                                    "section": f"{section_name}/{subsection_name}",
                                                    "para_id": para_data.get('para_id'),
                                                    "file_path": str(para_file),
                                                    "line_number": para_data.get('line_number'),
                                                    "content": content,
                                                    "context_type": "사업정보" if section_name == "business_content" else "기본정보" if section_name == "company_overview" else "재무정보",
                                                    "match_type": "section_name"
                                                })
                                            except Exception as e:
                                                continue
                            else:
                                # 섹션명이 매칭되지 않으면 내용에서만 검색
                                # 테이블 검색
                                if search_in in ["tables", "both"]:
                                    tables_path = subsection_dir / 'tables'
                                    if tables_path.exists():
                                        for table_file in tables_path.glob('*.json'):
                                            try:
                                                with open(table_file, 'r', encoding='utf-8') as f:
                                                    table_data = json.load(f)
                                                
                                                # 테이블 내용에서 검색
                                                search_text = ' '.join(table_data.get('headers', []) + 
                                                                     [cell for row in table_data.get('rows', []) for cell in row])
                                                
                                                # 내용 매칭 확인 (유사도 80% 포함)
                                                content_match = is_similar_match(search_term, search_text)
                                                
                                                if content_match:
                                                    search_results["results"]["tables"].append({
                                                        "section": f"{section_name}/{subsection_name}",
                                                        "table_id": table_data.get('table_id'),
                                                        "file_path": str(table_file),
                                                        "headers": table_data.get('headers', []),
                                                        "row_count": len(table_data.get('rows', [])),
                                                        "matched_content": search_text,
                                                        "context_type": "사업정보" if section_name == "business_content" else "기본정보" if section_name == "company_overview" else "재무정보",
                                                        "match_type": "content"
                                                    })
                                            except Exception as e:
                                                continue
                                
                                # 문단 검색
                                if search_in in ["paragraphs", "both"]:
                                    paragraphs_path = subsection_dir / 'paragraphs'
                                    if paragraphs_path.exists():
                                        for para_file in paragraphs_path.glob('*.json'):
                                            try:
                                                with open(para_file, 'r', encoding='utf-8') as f:
                                                    para_data = json.load(f)
                                                
                                                content = para_data.get('content', '')
                                                
                                                # 내용 매칭 확인 (유사도 80% 포함)
                                                content_match = is_similar_match(search_term, content)
                                                
                                                if content_match:
                                                    search_results["results"]["paragraphs"].append({
                                                        "section": f"{section_name}/{subsection_name}",
                                                        "para_id": para_data.get('para_id'),
                                                        "file_path": str(para_file),
                                                        "line_number": para_data.get('line_number'),
                                                        "content": content,
                                                        "context_type": "사업정보" if section_name == "business_content" else "기본정보" if section_name == "company_overview" else "재무정보",
                                                        "match_type": "content"
                                                    })
                                            except Exception as e:
                                                continue
            else:
                # 기존 재무제표 주석 섹션 검색
                # 테이블 검색
                if search_in in ["tables", "both"]:
                    tables_path = section_path / 'tables'
                    if tables_path.exists():
                        for table_file in tables_path.glob('*.json'):
                            try:
                                with open(table_file, 'r', encoding='utf-8') as f:
                                    table_data = json.load(f)
                                
                                # 테이블 내용에서 검색
                                search_text = ' '.join(table_data.get('headers', []) + 
                                                     [cell for row in table_data.get('rows', []) for cell in row])
                                
                                if is_similar_match(search_term, search_text):
                                    search_results["results"]["tables"].append({
                                        "section": section_name,
                                        "table_id": table_data.get('table_id'),
                                        "file_path": str(table_file),
                                        "headers": table_data.get('headers', []),
                                        "row_count": len(table_data.get('rows', [])),
                                        "matched_content": search_text,
                                        "context_type": "재무정보"
                                    })
                            except Exception as e:
                                continue
                
                # 문단 검색
                if search_in in ["paragraphs", "both"]:
                    paragraphs_path = section_path / 'paragraphs'
                    if paragraphs_path.exists():
                        for para_file in paragraphs_path.glob('*.json'):
                            try:
                                with open(para_file, 'r', encoding='utf-8') as f:
                                    para_data = json.load(f)
                                
                                content = para_data.get('content', '')
                                
                                if is_similar_match(search_term, content):
                                    search_results["results"]["paragraphs"].append({
                                        "section": section_name,
                                        "para_id": para_data.get('para_id'),
                                        "file_path": str(para_file),
                                        "line_number": para_data.get('line_number'),
                                        "content": content,
                                        "context_type": "재무정보"
                                    })
                            except Exception as e:
                                continue
        
        # 결과 수 제한
        search_results["results"]["tables"] = search_results["results"]["tables"][:max_results]
        search_results["results"]["paragraphs"] = search_results["results"]["paragraphs"][:max_results]
        
        # 컨텍스트별 그룹핑 정보 추가
        context_groups = {}
        for result in search_results["results"]["paragraphs"]:
            context_type = result.get("context_type", "기타")
            if context_type not in context_groups:
                context_groups[context_type] = 0
            context_groups[context_type] += 1
        
        for result in search_results["results"]["tables"]:
            context_type = result.get("context_type", "기타")
            if context_type not in context_groups:
                context_groups[context_type] = 0
            context_groups[context_type] += 1
        
        # 통계 추가
        search_results["summary"] = {
            "total_tables_found": len(search_results["results"]["tables"]),
            "total_paragraphs_found": len(search_results["results"]["paragraphs"]),
            "sections_searched": sections_to_search,
            "context_groups": context_groups,
            "enrichment_applied": True
        }
        
        return as_json_text(search_results)
        
    except Exception as e:
        return as_json_text({
            "status": "500",
            "message": "검색 중 오류 발생",
            "error": str(e)
        })


@mcp.tool(
    name="get_corporation_code",
    description="OpenDART에서 제공하는 모든 공시대상 회사의 고유번호 전체 목록(XML 파일)을 조회합니다. 기업명 검색 또는 고유번호 매핑에 사용됩니다.",
    tags={"기업전체목록", "고유번호전체", "기업식별", "코드매핑"}
)
def get_corporation_code() -> TextContent:
    """
    고유번호 목록 조회

    Returns:
        Dict[str, Any]: 고유번호 목록 (기업명, 고유번호, 종목코드 등 포함)

    참고: https://opendart.fss.or.kr/guide/detail.do?apiGrpCd=DS001&apiId=2019018
    """
    result = with_context(None, "get_corporation_code", lambda context: context.ds001.get_corporation_code())
    return as_json_text(result)