import requests from lxml import etree, html import copy from fastmcp import FastMCP import os from datetime import datetime, timedelta import json import re CONFLUENCE_URL = os.getenv("CONFLUENCE_URL") CONFLUENCE_USERNAME = os.getenv("CONFLUENCE_USERNAME") CONFLUENCE_API_TOKEN = os.getenv("CONFLUENCE_API_TOKEN") # 환경 변수를 성공적으로 불러왔는지 확인 if not all([CONFLUENCE_URL, CONFLUENCE_USERNAME, CONFLUENCE_API_TOKEN]): raise ValueError(f"'파일에서 필수 환경 변수(CONFLUENCE_URL, CONFLUENCE_USERNAME, CONFLUENCE_API_TOKEN)를 찾을 수 없습니다.") # requests 라이브러리에서 사용할 인증 정보 튜플 생성 AUTH = (CONFLUENCE_USERNAME, CONFLUENCE_API_TOKEN) print("[INFO] MCP Tool 기반 요약 시스템 초기화 완료") def update_confluence_cell( page_id: str, week_label: str, team_name: str, column_header: str, new_content_html: str ) -> dict: print(f"\n[DEBUG] --- 함수 시작: update_confluence_cell ---") print(f"[DEBUG] Page ID: {page_id}, 탭: {week_label}, 팀: {team_name}, 컬럼: {column_header}") try: # 1. 페이지 정보 가져오기 get_url = f"{CONFLUENCE_URL}rest/api/content/{page_id}?expand=body.storage,version,title" print(f"[DEBUG] 1. 페이지 정보 가져오기 (GET 요청): {get_url}") response = requests.get(get_url, auth=AUTH, headers={"Accept": "application/json"}, timeout=10) response.raise_for_status() data = response.json() storage_xml = data['body']['storage']['value'] current_version = data['version']['number'] page_title = data['title'] print(f"[DEBUG] - 페이지 제목: '{page_title}', 현재 버전: {current_version}") # 2. XML 파싱 (루트에 네임스페이스 선언 추가) parser = etree.XMLParser(recover=True, strip_cdata=False) root = etree.fromstring( f"{storage_xml}", parser=parser ) ns = {'ac': 'http://www.atlassian.com/confluence/macros/4/ac'} # 3. 카드(탭) 찾기 print(f"\n[DEBUG] 3-1. '{week_label}' 카드(탭) 찾기") target_cards = root.xpath( ".//ac:structured-macro[@ac:name='card'][ac:parameter[@ac:name='label' and normalize-space(text())=$week_label]]", week_label=week_label, namespaces=ns ) if not target_cards: print(f"[오류] '{week_label}' 카드(탭)을 찾을 수 없습니다.") all_labels = root.xpath( ".//ac:structured-macro[@ac:name='card']/ac:parameter[@ac:name='label']/text()", namespaces=ns ) print(f"[DEBUG] - 페이지에서 찾은 모든 카드 label: {all_labels}") return {"status": "fail", "message": f"오류: Storage Format 내에서 이름이 '{week_label}'인 카드(탭)를 찾을 수 없습니다."} target_card = target_cards[0] print(f"[DEBUG] - '{week_label}' 카드(탭) 찾기 성공!") # 4. 카드 내부 테이블 찾기 print(f"\n[DEBUG] 3-2. 카드 내 테이블 찾기") table_list = target_card.xpath('.//table') if not table_list: print(f"[오류] 테이블 찾기 실패!") return {"status": "fail", "message": f"오류: 카드(탭) '{week_label}' 내에서 테이블을 찾을 수 없습니다."} table = table_list[0] print(f"[DEBUG] - 테이블 찾기 성공!") # 5. 헤더에서 컬럼 인덱스 robust하게 찾기 print(f"\n[DEBUG] 3-3. '{column_header}' 컬럼 헤더 찾기") headers = table.xpath('.//thead/tr/th') if not headers: headers = table.xpath('.//tr[1]/th') target_col_index = -1 header_texts = [] for i, th in enumerate(headers): header_text = ''.join(th.itertext()).strip() header_texts.append(header_text) if column_header in header_text: target_col_index = i break if target_col_index == -1: print(f"[오류] 컬럼 헤더 찾기 실패!") print(f"[DEBUG] - 테이블에서 찾은 전체 헤더: {header_texts}") return {"status": "fail", "message": f"오류: 테이블에서 '{column_header}'을(를) 포함하는 열 헤더를 찾을 수 없습니다."} print(f"[DEBUG] - 찾음! 컬럼 인덱스: {target_col_index}") # 6. 행 찾기 print(f"\n[DEBUG] 3-4. '{team_name}' 행 찾기 (첫 번째 셀 기준)") rows = table.xpath('.//tbody/tr') if not rows: rows = table.xpath('.//tr') target_row = None for row in rows: first_cell_text = row.xpath("normalize-space(./td[1])") if first_cell_text == team_name: target_row = row break if target_row is None: print(f"[오류] 행 찾기 실패!") return {"status": "fail", "message": f"오류: 첫 번째 열의 내용이 '{team_name}'인 행을 찾을 수 없습니다."} print(f"[DEBUG] - 찾음! 행을 찾았습니다.") # 7. 셀(td) 찾기 print(f"\n[DEBUG] 3-5. 목표 셀 찾기 (행: '{team_name}', 열 인덱스: {target_col_index})") target_cells = target_row.xpath(f'./td[{target_col_index + 1}]') if not target_cells: print(f"[오류] 셀 찾기 실패!") return {"status": "fail", "message": f"오류: '{team_name}' 행에서 {target_col_index + 1}번째 열(셀)을 찾을 수 없습니다."} target_cell = target_cells[0] print(f"[DEBUG] - 최종 목표 셀 찾기 성공!") # --- 8. 셀 내용 업데이트 --- print(f"\n[DEBUG] 4. 셀 내용 업데이트 시작") # 기존 셀의 모든 자식 노드와 텍스트를 깨끗이 지웁니다. target_cell.clear() if not new_content_html or not new_content_html.strip(): # 내용이 비어있으면 셀이 깨지지 않도록
태그 하나만 추가합니다. etree.SubElement(target_cell, 'br') else: try: # 1. HTML 문자열을 파싱하기 위해 div로 감싸줍니다. (안정성을 위해) # 2. etree.fromstring 대신 html.fromstring을 사용해 너그럽게 파싱합니다. fragment_wrapper = html.fromstring(f"
{new_content_html}
") # 3. 파싱된 div의 자식 요소들을 목표 셀(td)에 그대로 복사합니다. # div의 첫 텍스트가 있다면 target_cell의 text로 먼저 할당합니다. if fragment_wrapper.text: target_cell.text = fragment_wrapper.text # div의 모든 자식 요소(br, span 등)를 순회하며 target_cell에 추가합니다. # 요소의 tail 텍스트(예: 뒤에 오는 텍스트)까지 모두 복사됩니다. for child in fragment_wrapper: target_cell.append(copy.deepcopy(child)) # ---------------------------------------------------- except Exception as e: # 파싱에 실패할 경우, 안전하게 텍스트로라도 넣습니다. (기존 문제 상황) print(f"[경고] HTML 파싱에 실패하여 내용을 순수 텍스트로 삽입합니다. 오류: {e}") target_cell.text = new_content_html print(f"[DEBUG] - 셀 내용 업데이트 완료") # 9. 수정된 XML 반환 및 페이지 업데이트 print(f"\n[DEBUG] 5. 업데이트된 페이지 XML 생성") updated_storage_xml_bytes = etree.tostring(root, encoding="unicode", method="xml") # 루트 네임스페이스 선언까지 제거 prefix = '' updated_storage_xml = updated_storage_xml_bytes[len(prefix):-len("")] print(f"\n[DEBUG] 6. Confluence 페이지 업데이트 (PUT 요청)") api_put_url = f"{CONFLUENCE_URL}rest/api/content/{page_id}" payload = { "version": {"number": current_version + 1}, "title": page_title, "type": "page", "body": { "storage": { "value": updated_storage_xml, "representation": "storage" } } } put_response = requests.put(api_put_url, auth=AUTH, json=payload, timeout=20) put_response.raise_for_status() print(f"[DEBUG] - PUT 요청 성공 (HTTP Status: {put_response.status_code})") page_link = f"{CONFLUENCE_URL}pages/viewpage.action?pageId={page_id}" return {"status": "success", "message": f"페이지 업데이트 완료! 확인: {page_link}"} except requests.exceptions.RequestException as e: print(f"[오류] Confluence API 요청 오류: {e}") return {"status": "fail", "message": f"Confluence API 요청 오류: {e}"} except etree.XMLSyntaxError as e: print(f"[오류] XML/HTML 파싱 오류: {e}") return {"status": "fail", "message": f"XML/HTML 파싱 오류: {e}"} except Exception as e: import traceback traceback.print_exc() return {"status": "fail", "message": f"알 수 없는 오류 발생: {e}"} def get_space_key_from_page_id(page_id: str) -> str: """페이지 ID로부터 스페이스 키를 가져옵니다.""" try: get_url = f"{CONFLUENCE_URL}rest/api/content/{page_id}?expand=space" response = requests.get(get_url, auth=AUTH, headers={"Accept": "application/json"}, timeout=10) response.raise_for_status() data = response.json() return data.get('space', {}).get('key', '') except Exception as e: print(f"[오류] 페이지 {page_id}의 스페이스 키 조회 실패: {e}") return '' def get_recent_pages_in_space(space_key: str, days: int = 7) -> dict: """특정 스페이스에서 최근 지정된 일수 동안 생성되거나 수정된 페이지들을 찾습니다. 댓글보다 실제 페이지 생성/수정에 집중합니다.""" print(f"\n[DEBUG] --- 함수 시작: get_recent_pages_in_space ---") print(f"[DEBUG] Space: {space_key}, Days: {days}") try: # 날짜 계산 (지정된 일수 전) start_date = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d') print(f"[DEBUG] 검색 시작 날짜: {start_date}") # 페이지와 댓글을 모두 검색 content_cql = f'type in (page, comment) AND space = "{space_key}" AND created >= "{start_date}"' content_url = f"{CONFLUENCE_URL}rest/api/content/search" content_params = { 'cql': content_cql, 'expand': 'version,space,history.lastUpdated,body.storage,container', 'limit': 200 } print(f"[DEBUG] 페이지+댓글 검색 CQL: {content_cql}") content_response = requests.get(content_url, auth=AUTH, params=content_params, timeout=10) if content_response.status_code != 200: print(f"[오류] API 응답 오류: {content_response.status_code} - {content_response.text}") content_response.raise_for_status() content_data = content_response.json() # 수정된 페이지 검색 modified_cql = f'type = page AND space = "{space_key}" AND lastModified >= "{start_date}"' modified_params = { 'cql': modified_cql, 'expand': 'version,space,history.lastUpdated,body.storage', 'limit': 100 } print(f"[DEBUG] 수정된 페이지 검색 CQL: {modified_cql}") modified_response = requests.get(content_url, auth=AUTH, params=modified_params, timeout=10) if modified_response.status_code != 200: print(f"[오류] API 응답 오류: {modified_response.status_code} - {modified_response.text}") modified_response.raise_for_status() modified_data = modified_response.json() # 결과 정리 및 분류 created_pages = [] created_comments = [] for item in content_data.get('results', []): if item['type'] == 'page': content = item.get('body', {}).get('storage', {}).get('value', '') if len(content.strip()) > 50: # 최소 50자 이상의 내용이 있는 페이지만 created_pages.append({ 'id': item['id'], 'title': item['title'], 'url': f"{CONFLUENCE_URL}pages/viewpage.action?pageId={item['id']}", 'created': item.get('history', {}).get('createdDate', ''), 'author': item.get('history', {}).get('createdBy', {}).get('displayName', ''), 'content_length': len(content.strip()) }) elif item['type'] == 'comment': content = item.get('body', {}).get('storage', {}).get('value', '') parent_page = item.get('container', {}) created_comments.append({ 'id': item['id'], 'content': content[:300], # 댓글 내용 (300자 제한) 'author': item.get('history', {}).get('createdBy', {}).get('displayName', ''), 'created': item.get('history', {}).get('createdDate', ''), 'parent_page_id': parent_page.get('id', ''), 'parent_page_title': parent_page.get('title', '') }) modified_pages = [] created_page_ids = set(p['id'] for p in created_pages) for page in modified_data.get('results', []): if page['id'] not in created_page_ids: content = page.get('body', {}).get('storage', {}).get('value', '') if len(content.strip()) > 50: modified_pages.append({ 'id': page['id'], 'title': page['title'], 'url': f"{CONFLUENCE_URL}pages/viewpage.action?pageId={page['id']}", 'modified': page.get('version', {}).get('when', ''), 'modifier': page.get('version', {}).get('by', {}).get('displayName', ''), 'content_length': len(content.strip()) }) # 우선순위로 정렬 created_pages.sort(key=lambda x: x['content_length'], reverse=True) modified_pages.sort(key=lambda x: x['content_length'], reverse=True) created_comments.sort(key=lambda x: x['created'], reverse=True) # 통계 계산 all_page_ids = set([p['id'] for p in created_pages] + [p['id'] for p in modified_pages]) print(f"[DEBUG] 결과 - 생성된 페이지: {len(created_pages)}개, 수정된 페이지: {len(modified_pages)}개, 댓글: {len(created_comments)}개") print(f"[DEBUG] 전체 고유 페이지: {len(all_page_ids)}개") # 디버깅 정보 출력 if created_pages: print(f"[DEBUG] 생성된 페이지 목록:") for page in created_pages[:3]: print(f" - {page['title']} (내용 길이: {page['content_length']}자)") if created_comments: print(f"[DEBUG] 최근 댓글 목록:") for comment in created_comments[:3]: print(f" - {comment['parent_page_title']}에 댓글 by {comment['author']}") return { "status": "success", "space_key": space_key, "period": f"최근 {days}일", "start_date": start_date, "created_pages": created_pages, "modified_pages": modified_pages, "created_comments": created_comments, "total_unique_pages": len(all_page_ids), "created_count": len(created_pages), "modified_count": len(modified_pages), "comment_count": len(created_comments) } except requests.exceptions.RequestException as e: print(f"[오류] Confluence API 요청 오류: {e}") return {"status": "fail", "message": f"Confluence API 요청 오류: {e}"} except Exception as e: import traceback traceback.print_exc() return {"status": "fail", "message": f"알 수 없는 오류 발생: {e}"} # MCP 서버 설정 mcp = FastMCP( name="Confluence Weekly Report Editor", instructions="""컨플루언스 주간 보고서 편집 도구입니다. 페이지 내용을 분석하여 다음 두 카테고리로 요약합니다: **1. 시너지 협업 방안:** 팀 간 협업, 공동 프로젝트, 기술 공유, 회의/발표, 외부 부서와의 협력, 교류회, 워크샵, 업체 미팅 등 **2. 주요 이슈 및 의사결정 사항:** 업체 미팅 결과, 솔루션 변경, 서비스 종료/시작, 개발 완료, 정책 변경, 일정 조정, 기술적 이슈 등 제목 없이 바로 내용만 출력하세요.""", ) @mcp.tool() async def edit_weekly_report_cell( page_id: str, week_label: str, team_name: str, column_header: str, new_content_html: str ) -> dict: """ confluence 페이지의 주간 보고서의 특정 셀 내용을 수정합니다. """ return update_confluence_cell( page_id=page_id, week_label=week_label, team_name=team_name, column_header=column_header, new_content_html=new_content_html ) @mcp.tool() async def get_recent_space_pages( space_key: str, days: int = 7 ) -> dict: """ 특정 Confluence 스페이스에서 최근 지정된 일수 동안 생성되거나 수정된 페이지들을 찾습니다. """ return get_recent_pages_in_space(space_key=space_key, days=days) @mcp.tool() async def summarize_recent_pages_for_report( space_key: str = None, page_id: str = None, days: int = 7 ) -> dict: """ Confluence 스페이스의 최근 페이지 변경사항을 WRM(Weekly Report Meeting) 주간보고서 형식으로 요약합니다. 담당 급에게 보고하는 실장간담회용 '시너지 협업 방안'과 '주요 이슈 및 의사결정 사항' 두 카테고리로 분류하여 요약합니다. **구체적인 사람 이름은 언급하지 말아줘** Args: space_key: Confluence 스페이스 키 (직접 지정) page_id: 페이지 ID (해당 페이지의 스페이스를 자동으로 찾음) days: 분석할 최근 일수 (기본값: 7일) Returns: WRM 주간보고서 형식의 요약 결과: **구체적인 사람 이름은 언급하지 말아줘** **시너지 협업 방안 (조직 간 협업 중심):** - **구체적인 사람 이름은 언급하지 말아줘** - 팀 안의 '파트 간 기술 교류 활동'은 제외! - 다른 팀/부서와의 협업 및 교류회 (예: 컨텐츠서비스개발팀, 광고플랫폼개발팀, 클라우드운영개발팀 등) - 외부 업체와의 미팅 및 기술 협력 (업체명, 미팅 목적, 협력 내용 명시) - 타 본부와의 기술 교류 및 공동 프로젝트 (HS본부, CTO, VS본부 등) - 전사 차원의 활동 및 성과 공유회 (REINVENT, 커미티, 워크샵 등) - 외부 지원 활동 (영업 지원, 글로벌 법인 지원, 고객 지원 등) - 산학 협력 및 연구 과제 (대학교, 연구소와의 협업) **주요 이슈 및 의사결정 사항 (담당 급 보고 수준):** - **구체적인 사람 이름은 언급하지 말아줘** - 팀 안의 '파트 간 주요 이슈 및 의사결정'은 제외! - 신규 서비스/시스템 개발 완료 및 론칭 (출시 일정, 주요 기능, 영향도) - 중요한 시스템 이관 및 아키텍처 변경 (webOS, 서버 시스템 등) - 업체 계약 체결 및 기술 도입 (계약 금액, 기술명, 적용 계획) - 보안 검토 및 정책 변경 (보안 이슈, 대응 방안) - 예산 관련 의사결정 (비용 절감, 투자 승인 등) - 전략적 방향성 변경 및 중요한 정책 결정 - 품질 관리 체계 및 프로세스 개선 (테스트 체계, 운영 프로세스) - 글로벌 전개 및 신규 시장 진출 관련 의사결정 작성 형식: - HTML 형식으로 출력하세요 -