from typing import Dict, Any, Optional, List
from pathlib import Path
from ..apis.client import OpenDartClient
class DisclosureAPI:
"""DS001 - 공시정보 API"""
def __init__(self, client: OpenDartClient):
self.client = client
def get_corporation_code_by_name(self, corp_name: str) -> Dict[str, Any]:
"""
기업명으로 고유번호 검색
"""
from ..utils.corp_code_search import read_local_xml, parse_corp_code_xml, search_corporations
try:
xml_content = read_local_xml()
corporations = parse_corp_code_xml(xml_content)
results = search_corporations(corporations, corp_name)
return {
"status": "000",
"message": "정상",
"items": results
}
except FileNotFoundError:
return {
"status": "400",
"message": "CORPCODE.xml 파일이 없습니다. get_corporation_code를 먼저 실행해주세요."
}
except Exception as e:
return {
"status": "500",
"message": f"오류가 발생했습니다: {str(e)}"
}
def get_disclosure_list(
self,
corp_code: Optional[str] = None,
bgn_de: Optional[str] = None,
end_de: Optional[str] = None,
last_report_at: Optional[str] = None,
pblntf_ty: Optional[str] = None,
pblntf_detail_ty: Optional[str] = None,
corp_cls: Optional[str] = None,
sort: Optional[str] = None,
sort_mth: Optional[str] = None,
page_no: int = 1
) -> Dict[str, Any]:
"""
공시검색
https://opendart.fss.or.kr/guide/detail.do?apiGrpCd=DS001&apiId=2019001
"""
endpoint = "list.json"
params = {
"corp_code": corp_code,
"bgn_de": bgn_de,
"end_de": end_de,
"last_reprt_at": last_report_at,
"pblntf_ty": pblntf_ty,
"pblntf_detail_ty": pblntf_detail_ty,
"corp_cls": corp_cls,
"sort": sort,
"sort_mth": sort_mth,
"page_no": page_no,
"page_count": 20 # 고정값으로 설정
}
# None 값과 빈 문자열 제거
params = {k: v for k, v in params.items() if v is not None and v != ""}
return self.client.get(endpoint, params)
def get_corporation_info(self, corp_code: str) -> Dict[str, Any]:
"""
기업개황 조회
https://opendart.fss.or.kr/guide/detail.do?apiGrpCd=DS001&apiId=2019002
"""
endpoint = "company.json"
params = {"corp_code": corp_code}
return self.client.get(endpoint, params)
def get_disclosure_document(self, rcp_no: str) -> Dict[str, Any]:
"""
공시서류원본파일 조회
https://opendart.fss.or.kr/guide/detail.do?apiGrpCd=DS001&apiId=2019003
"""
import zipfile
import os
import shutil
endpoint = "document.xml"
params = {"rcept_no": rcp_no}
response = self.client.get(endpoint, params)
# Save the response to the specified directory if successful
if response.get("status") == "000" and isinstance(response.get("content"), bytes):
# Get the absolute path of the project root
project_root = Path(__file__).parent.parent.parent
data_dir = project_root / 'mcp_opendart' / 'utils' / 'data'
data_dir.mkdir(parents=True, exist_ok=True)
# Save the zip content to a temporary file
zip_path = data_dir / f'disclosure_{rcp_no}.zip'
with open(zip_path, 'wb') as f:
f.write(response["content"])
try:
# Extract the XML file from the zip
with zipfile.ZipFile(zip_path, 'r') as zip_ref:
# ZIP 파일 내용 확인
file_list = zip_ref.namelist()
print(f"ZIP 파일 내용: {file_list}")
# XML 파일 찾기
xml_files = [f for f in file_list if f.endswith('.xml')]
if not xml_files:
raise Exception("ZIP 파일에서 XML 파일을 찾을 수 없습니다")
# 🎯 개선된 XML 파일 선택 로직
xml_filename = self._select_best_xml_file(xml_files, zip_ref)
if not xml_filename:
raise Exception("선택할 수 있는 XML 파일이 없습니다")
zip_ref.extract(xml_filename, data_dir)
# 추출된 XML 파일의 실제 경로
extracted_xml_path = data_dir / xml_filename
# disclosure_ 접두사가 있는 이름으로 복사
xml_path = data_dir / f'disclosure_{rcp_no}.xml'
if extracted_xml_path != xml_path:
shutil.copy2(extracted_xml_path, xml_path)
# 원본 파일 삭제
os.remove(extracted_xml_path)
# Remove the temporary zip file
os.remove(zip_path)
# 응답에 추가 정보 포함 (content는 제거하여 JSON 직렬화 문제 방지)
response.pop("content", None) # bytes 타입 content 제거
response.update({
"saved_path": str(xml_path),
"file_size": xml_path.stat().st_size if xml_path.exists() else 0,
"content_info": f"XML 파일이 다운로드되어 {str(xml_path)}에 저장됨"
})
except Exception as e:
print(f"Failed to extract zip file: {e}")
if os.path.exists(zip_path):
os.remove(zip_path)
response["error"] = f"파일 처리 실패: {str(e)}"
return response
def get_corporation_code(self) -> Dict[str, Any]:
"""
고유번호 조회 및 저장
https://opendart.fss.or.kr/guide/detail.do?apiGrpCd=DS001&apiId=2019018
"""
import zipfile
import io
endpoint = "corpCode.xml"
response = self.client.get(endpoint)
# Save the response to the specified file
if response.get("status") == "000" and isinstance(response.get("content"), bytes):
import os
from pathlib import Path
# Get the absolute path of the project root
project_root = Path(__file__).parent.parent.parent
data_dir = project_root / 'mcp_opendart' / 'utils' / 'data'
data_dir.mkdir(parents=True, exist_ok=True)
# Save the zip content to a temporary file
zip_path = data_dir / 'CORPCODE.zip'
with open(zip_path, 'wb') as f:
f.write(response["content"])
try:
# Extract the XML file from the zip
with zipfile.ZipFile(zip_path, 'r') as zip_ref:
zip_ref.extractall(data_dir)
# Remove the temporary zip file
os.remove(zip_path)
except Exception as e:
print(f"Failed to extract zip file: {e}")
if os.path.exists(zip_path):
os.remove(zip_path)
return response
def extract_financial_notes_document(self, rcp_no: str) -> Dict[str, Any]:
"""
공시서류에서 연결재무제표 주석과 재무제표 주석만 추출
"""
from ..utils.financial_notes_extractor import extract_financial_notes, save_extracted_data, should_regenerate_cache
import shutil
try:
# XML 파일 경로 확인
project_root = Path(__file__).parent.parent.parent
xml_path = project_root / 'mcp_opendart' / 'utils' / 'data' / f'disclosure_{rcp_no}.xml'
if not xml_path.exists():
return {
"status": "404",
"message": "XML 파일을 찾을 수 없습니다",
"error": f"파일이 존재하지 않습니다: {xml_path}"
}
# XML 파일 읽기
with open(xml_path, 'r', encoding='utf-8') as f:
xml_content = f.read()
# 기업명 추출
import re
corp_name_match = re.search(r'<COMPANY-NAME[^>]*>([^<]+)</COMPANY-NAME>', xml_content)
corp_name = corp_name_match.group(1) if corp_name_match else "Unknown"
# 캐시 경로 설정
cache_path = project_root / 'mcp_opendart' / 'utils' / 'data' / 'disclosure_cache' / f'financial_notes_{rcp_no}'
# 재무제표 주석 추출
extracted_data = extract_financial_notes(xml_content, corp_name, rcp_no)
# 재무제표 주석이 없으면 LLM에게 다시 호출하라고 지시
if extracted_data.get('status') == 'error' or len(extracted_data.get('sections', {})) == 0:
# 기업코드 추출 (XML에서)
corp_code_match = re.search(r'<CORP_CODE[^>]*>([^<]+)</CORP_CODE>', xml_content)
corp_code = corp_code_match.group(1) if corp_code_match else "Unknown"
return {
"status": "404",
"message": "현재 공시서류에 재무제표 주석이 없습니다. 다른 보고서로 다시 호출하세요.",
"error": "연결재무제표 주석과 재무제표 주석을 찾을 수 없습니다",
"action_required": "get_disclosure_document를 다른 rcp_no로 다시 호출하세요",
"corp_code": corp_code,
"corp_name": corp_name,
"recommended_reports": [
"반기보고서",
"사업보고서",
"1분기보고서",
"3분기보고서"
]
}
# 저장 경로 설정 (disclosure_cache 디렉토리 사용)
save_path = project_root / 'mcp_opendart' / 'utils' / 'data' / 'disclosure_cache' / f'financial_notes_{rcp_no}'
saved_path = save_extracted_data(extracted_data, str(save_path))
# 요약 정보 생성
summary = {
"sections_found": len(extracted_data['sections']),
"total_tables": sum(section['table_count'] for section in extracted_data['sections'].values()),
"total_paragraphs": sum(section['paragraph_count'] for section in extracted_data['sections'].values())
}
return {
"status": "000",
"message": "정상",
"corp_name": corp_name,
"rcp_no": rcp_no,
"summary": summary,
"sections": {name: {
"title": data['title'],
"table_count": data['table_count'],
"paragraph_count": data['paragraph_count'],
"line_count": data['line_count']
} for name, data in extracted_data['sections'].items()},
"save_path": saved_path,
"extraction_date": extracted_data['metadata']['extraction_date']
}
except Exception as e:
return {
"status": "500",
"message": "추출 중 오류 발생",
"error": str(e)
}
def _select_best_xml_file(self, xml_files: List[str], zip_ref) -> Optional[str]:
"""
ZIP 파일에서 최적의 XML 파일을 선택
감사보고서를 제외하고 비즈니스 보고서를 우선 선택
"""
try:
best_file = None
best_score = -1
for xml_file in xml_files:
try:
# XML 내용 읽기
xml_content = zip_ref.read(xml_file).decode('utf-8')
# 점수 계산 시스템
score = 0
# 비즈니스 보고서 키워드 (높은 가중치)
business_keywords = ['연결재무제표', '재무제표', '사업보고서', '반기보고서', '분기보고서']
for keyword in business_keywords:
if keyword in xml_content:
score += 2 # 비즈니스 키워드는 +2점
# 감사보고서 키워드 (낮은 가중치, 하지만 제외하지는 않음)
audit_keywords = ['감사보고서', '연결감사보고서']
for keyword in audit_keywords:
if keyword in xml_content:
score += 1 # 감사 키워드는 +1점
# 파일 크기 고려 (큰 파일 우선)
size_bonus = min(len(xml_content) // 100000, 5) # 최대 5점 보너스
score += size_bonus
print(f"파일 {xml_file}: 점수={score} (크기={len(xml_content):,} bytes)")
# 최고 점수 파일 선택
if score > best_score:
best_score = score
best_file = xml_file
except Exception as e:
print(f"XML 파일 분석 실패 {xml_file}: {e}")
continue
if best_file:
print(f"최적 파일 선택: {best_file} (점수: {best_score})")
return best_file
else:
# 적절한 파일을 찾지 못하면 첫 번째 선택 (fallback)
print(f"적절한 파일을 찾지 못해 첫 번째 선택: {xml_files[0]}")
return xml_files[0]
except Exception as e:
print(f"XML 파일 선택 중 오류: {e}")
return xml_files[0] if xml_files else None
