실습 6 — 배포할 폴더 만들기

import shutil
from pathlib import Path

DEPLOY = Path("out_ch07_deploy")
# parents=True: 중간 폴더까지 만든다. exist_ok=True: 이미 있으면 그대로 둔다
(DEPLOY / ".streamlit").mkdir(parents=True, exist_ok=True)

APP_CODE = r'''
import os
import re
import zlib
from pathlib import Path

import streamlit as st
from dotenv import find_dotenv, load_dotenv
from langchain_classic.chains import RetrievalQA
from langchain_community.document_loaders import PyPDFLoader
from langchain_community.vectorstores import FAISS
from langchain_core.embeddings import Embeddings
from langchain_core.language_models.fake_chat_models import FakeListChatModel
from langchain_openai import ChatOpenAI
from langchain_text_splitters import RecursiveCharacterTextSplitter

# 앱 파일과 같은 폴더의 약관 PDF. 클라우드에서도 저장소 안의 이 파일을 찾는다
PDF = Path(__file__).parent / "travel_terms.pdf"
HEADER = "구름다리여행 여행 약관 (연습용 가상 문서)"
PLACEHOLDER = "발급받은_인증키"


class BigramEmbeddings(Embeddings):
    """글자 두 개 묶음(바이그램)을 세어 숫자 목록을 만드는 연습용 임베딩"""

    def __init__(self, size=512):
        self.size = size

    def _vector(self, text):
        text = text.replace(" ", "").replace("\n", "")
        vec = [0.0] * self.size
        for i in range(len(text) - 1):
            # 글자 두 개를 crc32 로 0~511 사이의 칸 번호로 바꿔 그 칸을 1 올린다
            vec[zlib.crc32(text[i:i + 2].encode("utf-8")) % self.size] += 1.0
        norm = sum(v * v for v in vec) ** 0.5 or 1.0
        return [v / norm for v in vec]

    def embed_documents(self, texts):
        return [self._vector(t) for t in texts]

    def embed_query(self, text):
        return self._vector(text)


def get_api_key():
    """클라우드에서는 Secrets 칸, 내 PC 에서는 .env 에서 키를 읽는다"""
    try:
        return st.secrets["OPENAI_API_KEY"]
    except (FileNotFoundError, KeyError):
        load_dotenv(find_dotenv(usecwd=True))
        return os.getenv("OPENAI_API_KEY")


@st.cache_resource(show_spinner="약관으로 색인을 만드는 중입니다...")
def load_retriever():
    """PDF 읽기 → 머리글 지우기 → 조각 내기 → FAISS 색인. 서버에서 한 번만 한다"""
    pages = PyPDFLoader(str(PDF)).load()
    for page in pages:
        page.page_content = page.page_content.replace(HEADER, "").strip()
    chunks = RecursiveCharacterTextSplitter(chunk_size=200, chunk_overlap=30).split_documents(pages)
    db = FAISS.from_documents(chunks, BigramEmbeddings())
    return db.as_retriever(search_kwargs={"k": 2}), len(chunks)


@st.cache_resource
def load_qa(api_key):
    """키가 있으면 진짜 모델, 없거나 자리표시면 연습 모드의 가짜 모델로 RetrievalQA 를 만든다"""
    retriever, _ = load_retriever()
    if api_key and api_key != PLACEHOLDER:
        llm = ChatOpenAI(model=os.getenv("OPENAI_MODEL", "gpt-4.1-mini"), api_key=api_key, temperature=0)
        mode = "OpenAI"
    else:
        llm = FakeListChatModel(responses=["(연습 모드) 진짜 모델 없이 약관만 찾았습니다. 아래 근거 조항을 확인해 주세요."])
        mode = "연습 모드"
    qa = RetrievalQA.from_chain_type(llm=llm, retriever=retriever, return_source_documents=True)
    return qa, mode


def source_label(doc):
    """근거 조각 하나를 '2쪽 · 제5조, 제6조' 꼴로 줄인다"""
    names = list(dict.fromkeys(re.findall(r"제\d+조", doc.page_content)))
    return f"{doc.metadata['page'] + 1}쪽 · {', '.join(names)}"


def show(message):
    with st.chat_message(message["role"]):
        st.markdown(message["content"])
        if message.get("sources"):
            with st.expander("근거 조항 보기"):
                for s in message["sources"]:
                    st.caption(s)


st.set_page_config(page_title="구름다리여행 약관 챗봇")
st.title("구름다리여행 약관 챗봇")
_, n_chunks = load_retriever()
qa, mode = load_qa(get_api_key())
with st.sidebar:
    st.header("문서")
    st.caption(f"약관 조각: {n_chunks}개")
    st.caption(f"답하는 모델: {mode}")
    st.caption("연습용으로 지어낸 가상의 약관입니다.")

if "doc_messages" not in st.session_state:
    st.session_state.doc_messages = []
for m in st.session_state.doc_messages:
    show(m)

if question := st.chat_input("약관에 대해 물어보세요"):
    user = {"role": "user", "content": question}
    show(user)
    with st.spinner("약관에서 찾는 중입니다..."):
        result = qa.invoke({"query": question})
    bot = {"role": "assistant", "content": result["result"],
           "sources": [source_label(d) for d in result["source_documents"]]}
    show(bot)
    st.session_state.doc_messages.extend([user, bot])
'''

(DEPLOY / "app.py").write_text(APP_CODE.lstrip(), encoding="utf-8")
# 앱이 읽는 PDF 도 같은 폴더에 복사한다. 저장소에 함께 올라가야 클라우드에서 찾는다
shutil.copy("travel_terms.pdf", DEPLOY / "travel_terms.pdf")
print("저장:", DEPLOY / "app.py", "/", len(APP_CODE.strip().splitlines()), "줄")
print("복사:", DEPLOY / "travel_terms.pdf", "/", (DEPLOY / "travel_terms.pdf").stat().st_size, "바이트")

from importlib.metadata import version

# 앱이 import 하는 패키지와, import 에는 안 보이지만 안쪽에서 쓰는 패키지(pypdf, faiss-cpu)
PACKAGES = [
    "streamlit", "python-dotenv",
    "langchain-core", "langchain-community", "langchain-classic", "langchain-openai", "langchain-text-splitters",
    "pypdf", "faiss-cpu",
]
# version(이름): 이 PC 에 설치된 판. "이름==판" 으로 적어 클라우드에도 같은 판을 설치하게 한다
lines = [f"{name}=={version(name)}" for name in PACKAGES]
(DEPLOY / "requirements.txt").write_text("\n".join(lines) + "\n", encoding="utf-8")
print("저장:", DEPLOY / "requirements.txt")
print("\n".join(lines))

# 비밀 값 견본: 값은 자리표시. 진짜 키는 클라우드의 Secrets 칸에 넣는다
SECRETS_EXAMPLE = '''# 이 파일을 복사해 이름을 secrets.toml 로 바꾸고 값을 넣는다. secrets.toml 은 올리지 않는다
OPENAI_API_KEY = "발급받은_인증키"
OPENAI_MODEL = "gpt-4.1-mini"
'''
(DEPLOY / ".streamlit" / "secrets.toml.example").write_text(SECRETS_EXAMPLE, encoding="utf-8")

# .gitignore: git 이 올리지 않을 파일. 키가 든 두 파일과 파이썬이 만드는 찌꺼기 폴더
GITIGNORE = """.env
.streamlit/secrets.toml
__pycache__/
.venv/
"""
(DEPLOY / ".gitignore").write_text(GITIGNORE, encoding="utf-8")

# 폴더 안의 파일을 모두 찍어 본다. rglob("*"): 아래 폴더까지 모두 찾는다
for f in sorted(p for p in DEPLOY.rglob("*") if p.is_file()):
    print(f.relative_to(DEPLOY).as_posix(), f.stat().st_size, "바이트")

import streamlit as st
from streamlit.testing.v1 import AppTest

st.cache_resource.clear()

at = AppTest.from_file("out_ch07_deploy/app.py", default_timeout=30)
# 자리표시를 Secrets 에 넣어 연습 모드로 시험한다(진짜 모델을 부르지 않으려고)
at.secrets["OPENAI_API_KEY"] = "발급받은_인증키"
at.run()
print("제목:", at.title[0].value)
print("사이드바:", [c.value for c in at.sidebar.caption])
print("오류:", len(at.exception), "개")