26Y27a1

Confluence 상단의 문서 속성(Page Properties)에서 사람이 작성한 1차 메타데이터(Human Input)를 먼저 파싱하고, 비어있는 메타데이터 필드만 사내 LLM(vLLM)이 2차로 보충(Fallback Extraction)하여 YAML Front-Matter로 자동 합성하는 방식 B 개선 파이프라인 전체 소스 코드입니다.

---

1. 프로젝트 파일 구조

/app
├── main.py               # 파이프라인 오케스트레이션 메인 스크립트
├── confluence_client.py   # Confluence CQL 증분 조회
├── converter.py           # HTML 표(Page Properties) 파싱 + LLM 메타데이터 합성 + Clean MD 변환
├── git_sync.py            # Git 저장 및 Commit Hash 반환
├── indexer.py             # Header Chunking & OpenSearch / Qdrant Dual Indexing
├── requirements.txt       # Python 패키지 의존성
└── Dockerfile            # Container 이미지 빌드 명세서

---

2. 모듈별 소스 코드

① requirements.txt

beautifulsoup4==4.12.3
html2text==2024.2.26
GitPython==3.1.43
langchain-text-splitters==0.2.2
opensearch-py==2.6.0
qdrant-client==1.9.1
requests==2.32.3
pyyaml==6.0.1

---

② converter.py (HTML 표 파싱 + Local LLM 보충 + Clean MD 변환)

Confluence 상단에 표 형태로 존재하는 doc_type, target_solutions, status 등을 우선 수집하고, 누락된 error_codes, tags 등은 사내 LLM이 본문을 분석하여 보충합니다.

import re
import json
import yaml
import requests
from bs4 import BeautifulSoup
import html2text

LOCAL_LLM_URL = "http://vllm.internal:8000/v1/chat/completions"

def extract_human_metadata_from_table(soup: BeautifulSoup) -> dict:
    """Confluence 상단 'Page Properties(문서 속성)' 표에서 사람이 입력한 메타데이터 추출"""
    human_meta = {}
    
    # 문서 내 첫 번째 table 검색 (보통 상단 매크로 표)
    table = soup.find('table')
    if not table:
        return human_meta

    # 키워드 매핑 테이블
    key_mapping = {
        "문서 유형": "doc_type", "doc_type": "doc_type",
        "대상 솔루션": "target_solutions", "target_solutions": "target_solutions",
        "적용 환경": "environment", "environment": "environment",
        "문서 상태": "status", "status": "status",
        "장애 코드": "error_codes", "에러 코드": "error_codes", "error_codes": "error_codes",
        "담당 팀": "owner", "owner": "owner"
    }

    for row in table.find_all('tr'):
        cols = row.find_all(['th', 'td'])
        if len(cols) >= 2:
            raw_key = cols[0].get_text(strip=True).lower()
            raw_val = cols[1].get_text(strip=True)

            for target_key, std_key in key_mapping.items():
                if target_key in raw_key:
                    if std_key == "target_solutions":
                        # 콤마, 공백 기준 리스트 변환
                        human_meta[std_key] = [item.strip().lower() for item in re.split(r'[,/ ]+', raw_val) if item.strip()]
                    else:
                        human_meta[std_key] = raw_val.lower()
                    break

    # 표 정제 후 DOM에서 제거하여 Markdown 본문 노이즈 방지
    table.decompose()
    return human_meta

def extract_llm_fallback_metadata(doc_text: str) -> dict:
    """누락된 메타데이터를 사내 Local LLM(Qwen/Llama)을 통해 보충 파싱"""
    prompt = f"""
    너는 Cloud-Native 인프라 문서의 메타데이터 분석기다.
    아래 문서 본문을 읽고, 규격에 맞는 메타데이터 JSON만 출력해라.

    [규칙]
    1. doc_type: [sop, incident_report, architecture, work_plan, weekly_report, general] 중 하나
    2. target_solutions: 본문에 관련된 솔루션 목록 (예: k8s, cilium, minio, vault, keycloak, harbor, jenkins, argocd 등)
    3. environment: [prd, stg, dev, shared] 중 하나
    4. status: [active, deprecated] 중 하나
    5. error_codes: 본문에 언급된 장애/에러 코드 리스트 (예: OOMKilled, ImagePullBackOff 등)
    6. tags: 핵심 키워드 3~5개 (영문 소문자)

    [문서 본문]
    {doc_text[:2500]}

    JSON Format:
    """
    
    payload = {
        "model": "Qwen2.5-Coder-32B",
        "messages": [{"role": "user", "content": prompt}],
        "temperature": 0.1,
        "response_format": {"type": "json_object"}
    }

    try:
        res = requests.post(LOCAL_LLM_URL, json=payload, timeout=20)
        return json.loads(res.json()["choices"][0]["message"]["content"])
    except Exception:
        return {
            "doc_type": "general", "target_solutions": [], "environment": "shared",
            "status": "active", "error_codes": [], "tags": []
        }

def convert_confluence_html_to_md(page_data: dict, raw_html: str) -> str:
    soup = BeautifulSoup(raw_html, 'html.parser')

    # 불필요 태그 제거
    for tag in soup.find_all(['script', 'style', 'nav', 'header', 'footer']):
        tag.decompose()

    # 1. 사람이 작성한 1차 메타데이터 추출 (표 파싱)
    human_meta = extract_human_metadata_from_table(soup)

    # 2. html2text 이용한 Clean Markdown 생성
    h2t = html2text.HTML2Text()
    h2t.ignore_links = False
    h2t.ignore_images = False
    h2t.ignore_tables = False
    h2t.body_width = 0
    clean_md_body = h2t.handle(str(soup))

    # 3. 누락된 항목이 있으면 Local LLM으로 2차 보충 (Fallback)
    llm_meta = {}
    if not human_meta.get("doc_type") or not human_meta.get("target_solutions"):
        llm_meta = extract_llm_fallback_metadata(clean_md_body)

    # 4. 메타데이터 최종 병합 (사람 입력 우선)
    final_metadata = {
        "id": page_data["id"],
        "title": page_data["title"],
        "space": page_data["space"]["key"],
        "version": page_data["version"]["number"],
        "doc_type": human_meta.get("doc_type") or llm_meta.get("doc_type", "general"),
        "target_solutions": human_meta.get("target_solutions") or llm_meta.get("target_solutions", []),
        "environment": human_meta.get("environment") or llm_meta.get("environment", "prd"),
        "status": human_meta.get("status") or llm_meta.get("status", "active"),
        "error_codes": human_meta.get("error_codes") or llm_meta.get("error_codes", []),
        "owner": human_meta.get("owner", page_data["history"]["createdBy"]["displayName"]),
        "tags": llm_meta.get("tags", []),
        "last_modified": page_data["history"]["lastUpdated"]["when"]
    }

    yaml_frontmatter = f"---\n{yaml.dump(final_metadata, allow_unicode=True, sort_keys=False)}---\n\n"
    return yaml_frontmatter + clean_md_body

---

③ git_sync.py (Bitbucket/GitLab Sync)

import os
import re
from git import Repo

GIT_REPO_PATH = "/workspace/ops-knowledge-repo"

def sync_md_to_git(doc_id: str, title: str, space: str, md_content: str) -> str:
    repo = Repo(GIT_REPO_PATH)
    repo.remotes.origin.pull()

    safe_title = re.sub(r'[\/:*?"<>|]', '_', title).replace(' ', '_')
    dir_path = os.path.join(GIT_REPO_PATH, "ops-knowledge", space)
    os.makedirs(dir_path, exist_ok=True)
    
    file_path = os.path.join(dir_path, f"{doc_id}_{safe_title}.md")

    with open(file_path, "w", encoding="utf-8") as f:
        f.write(md_content)

    repo.git.add(file_path)
    if repo.is_dirty(untracked_files=True):
        commit_message = f"docs(sync): update CONF-{doc_id} with hybrid metadata"
        repo.index.commit(commit_message)
        repo.remotes.origin.push()

    return repo.head.commit.hexsha

---

④ indexer.py (YAML Front-Matter 기반 Dual Indexing)

YAML Front-Matter 내 메타데이터를 파싱하여 OpenSearch 필드 및 Qdrant Payload에 직접 주입함으로써 검색 필터링을 가능하게 합니다.

import yaml
import requests
from langchain_text_splitters import MarkdownHeaderTextSplitter
from opensearchpy import OpenSearch
from qdrant_client import QdrantClient
from qdrant_client.models import PointStruct

opensearch_cli = OpenSearch(hosts=[{'host': 'opensearch.internal', 'port': 9200}])
qdrant_cli = QdrantClient(host="qdrant.internal", port=6333)
TEI_EMBEDDING_URL = "http://tei-embedding.internal/embed"

def generate_embedding(text: str) -> list[float]:
    response = requests.post(TEI_EMBEDDING_URL, json={"inputs": text})
    return response.json()[0]

def chunk_and_dual_index(doc_id: str, md_content: str, commit_hash: str):
    # 1. Front-Matter 메타데이터와 본문 분리
    parts = md_content.split("---\n")
    parsed_metadata = {}
    if len(parts) >= 3:
        parsed_metadata = yaml.safe_load(parts[1])
        body_content = "---\n".join(parts[2:])
    else:
        body_content = md_content

    # 2. Header 기반 Chunking
    headers_to_split = [("#", "Header 1"), ("##", "Header 2"), ("###", "Header 3")]
    splitter = MarkdownHeaderTextSplitter(headers_to_split_on=headers_to_split)
    chunks = splitter.split_text(body_content)

    # 3. 기존 인덱스 삭제 (동기화 정합성)
    opensearch_cli.delete_by_query(
        index="ops-knowledge-sparse",
        body={"query": {"term": {"doc_id.keyword": doc_id}}}
    )
    qdrant_cli.delete(
        collection_name="ops-knowledge-dense",
        points_selector={"filter": {"must": [{"key": "doc_id", "match": {"value": doc_id}}]}}
    )

    # 4. OpenSearch & Qdrant Bulk Upsert
    qdrant_points = []
    for idx, chunk in enumerate(chunks):
        chunk_id = f"{doc_id}#{idx}"
        chunk_text = chunk.page_content
        
        chunk_meta = {
            "doc_id": doc_id,
            "chunk_id": chunk_id,
            "commit_hash": commit_hash,
            **parsed_metadata,
            **chunk.metadata
        }

        # A. OpenSearch (BM25)
        opensearch_cli.index(
            index="ops-knowledge-sparse",
            id=chunk_id,
            body={"content": chunk_text, **chunk_meta}
        )

        # B. Qdrant (Vector)
        vector = generate_embedding(chunk_text)
        qdrant_points.append(
            PointStruct(
                id=chunk_id,
                vector=vector,
                payload={"content": chunk_text, **chunk_meta}
            )
        )

    if qdrant_points:
        qdrant_cli.upsert(collection_name="ops-knowledge-dense", points=qdrant_points)

---

⑤ confluence_client.py (CQL API 수집)

import os
import requests

CONFLUENCE_URL = os.getenv("CONFLUENCE_URL", "http://confluence.internal")
CONFLUENCE_TOKEN = os.getenv("CONFLUENCE_TOKEN", "")

def get_updated_pages(last_sync_time: str) -> list:
    headers = {"Authorization": f"Bearer {CONFLUENCE_TOKEN}", "Accept": "application/json"}
    cql = f'space="OPS" AND lastModified >= "{last_sync_time}" ORDER BY lastModified ASC'
    
    url = f"{CONFLUENCE_URL}/wiki/rest/api/content/search?cql={cql}&expand=body.storage,version,history,metadata.labels"
    
    res = requests.get(url, headers=headers)
    if res.status_code == 200:
        return res.json().get("results", [])
    return []

---

⑥ main.py (전체 흐름 연결)

import os
import json
import logging
from datetime import datetime, timezone
from confluence_client import get_updated_pages
from converter import convert_confluence_html_to_md
from git_sync import sync_md_to_git
from indexer import chunk_and_dual_index

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
STATE_FILE = "/workspace/ops-knowledge-repo/.sync_state.json"

def get_last_sync_time():
    if os.path.exists(STATE_FILE):
        with open(STATE_FILE, "r") as f:
            return json.load(f).get("last_sync_time")
    return "2026-07-25T00:00:00Z"

def run_pipeline():
    last_sync = get_last_sync_time()
    current_time = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
    
    logging.info(f"Pipeline started. Syncing since: {last_sync}")
    pages = get_updated_pages(last_sync)
    logging.info(f"Target documents count: {len(pages)}")

    for page in pages:
        p_id, title, space = page["id"], page["title"], page["space"]["key"]
        raw_html = page["body"]["storage"]["value"]

        # Step 2: Hybrid Metadata Parsing + Clean MD
        md_content = convert_confluence_html_to_md(page, raw_html)
        
        # Step 3: Git Commit & Push
        commit_hash = sync_md_to_git(p_id, title, space, md_content)
        
        # Step 4: Metadata-Aware Dual Indexing
        chunk_and_dual_index(p_id, md_content, commit_hash)

    with open(STATE_FILE, "w") as f:
        json.dump({"last_sync_time": current_time}, f)
    logging.info("Pipeline executed successfully.")

if __name__ == "__main__":
    run_pipeline()

---

⑦ Dockerfile

FROM python:3.11-slim

RUN apt-get update && apt-get install -y --no-install-recommends \
    git \
    && rm -rf /var/lib/apt/lists/*

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

RUN useradd -m appuser && chown -R appuser:appuser /app
USER appuser

ENTRYPOINT ["python", "main.py"]