app.py

"""
Ginigen Blog / Streamlit App
────────────────────────────────────────────────────────────────────
- 2025-04-23  : Brave Search API 버전
- SerpHouse 전면 제거, Brave Search API 적용
- API Key     : 환경변수  SERPHOUSE_API_KEY  (이름만 그대로 사용)
────────────────────────────────────────────────────────────────────
"""

import os
import streamlit as st
import json
import anthropic
import requests
import logging
from gradio_client import Client
import markdown
import re
from datetime import datetime
# BeautifulSoup는 더 이상 사용하지 않지만, 필요 시 유지
# from bs4 import BeautifulSoup

# ───────────────────────────── 1) 로깅 ─────────────────────────────────────────
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)

# ───────────────────────────── 2) 전역 상수 / API 키 ───────────────────────────
ANTHROPIC_KEY  = os.getenv("API_KEY", "")
BRAVE_KEY      = os.getenv("SERPHOUSE_API_KEY", "")     # Brave Search API 키
BRAVE_ENDPOINT = "https://api.search.brave.com/res/v1/web/search"
IMAGE_API_URL  = "http://211.233.58.201:7896"
MAX_TOKENS     = 7_999

# ───────────────────────────── 3) 클라이언트 ──────────────────────────────────
client = anthropic.Anthropic(api_key=ANTHROPIC_KEY)

# ───────────────────────────── 4) 시스템 프롬프트 ─────────────────────────────
def get_system_prompt() -> str:
    return """
당신은 전문 블로그 작성 전문가입니다. 모든 블로그 글 작성 요청에 대해 다음의 8단계 프레임워크를 철저히 따르되, 자연스럽고 매력적인 글이 되도록 작성해야 합니다:

독자 연결 단계
1.1. 공감대 형성을 위한 친근한 인사
1.2. 독자의 실제 고민을 반영한 도입 질문
1.3. 주제에 대한 즉각적 관심 유도

문제 정의 단계
2.1. 독자의 페인포인트 구체화
2.2. 문제의 시급성과 영향도 분석
2.3. 해결 필요성에 대한 공감대 형성

전문성 입증 단계
3.1. 객관적 데이터 기반 분석
3.2. 전문가 견해와 연구 결과 인용
3.3. 실제 사례를 통한 문제 구체화

솔루션 제공 단계
4.1. 단계별 실천 가이드라인 제시
4.2. 즉시 적용 가능한 구체적 팁
4.3. 예상 장애물과 극복 방안 포함

신뢰도 강화 단계
5.1. 실제 성공 사례 제시
5.2. 구체적 사용자 후기 인용
5.3. 객관적 데이터로 효과 입증

행동 유도 단계
6.1. 명확한 첫 실천 단계 제시
6.2. 시급성을 강조한 행동 촉구
6.3. 실천 동기 부여 요소 포함

진정성 강화 단계
7.1. 솔루션의 한계 투명하게 공개
7.2. 개인별 차이 존재 인정
7.3. 필요 조건과 주의사항 명시

관계 지속 단계
8.1. 진정성 있는 감사 인사
8.2. 다음 컨텐츠 예고로 기대감 조성
8.3. 소통 채널 안내

작성 시 준수사항
9.1. 글자 수: 1500-2000자 내외
9.2. 문단 길이: 3-4문장 이내
9.3. 시각적 구분: 소제목, 구분선, 번호 목록 활용
9.4. 톤앤매너: 친근하고 전문적인 대화체
9.5. 데이터: 모든 정보의 출처 명시
9.6. 가독성: 명확한 단락 구분과 강조점 사용
"""

# ───────────────────────────── 5) Brave Search 함수 ───────────────────────────
def brave_search(query: str, count: int = 5):
    """
    Brave Web Search API 호출 → list[dict] 반환
    반환 항목: title, link, snippet, displayed_link, index
    """
    if not BRAVE_KEY:
        raise RuntimeError("환경변수 SERPHOUSE_API_KEY(=Brave API key)가 설정되어 있지 않습니다.")

    headers = {
        "Accept": "application/json",
        "Accept-Encoding": "gzip",
        "X-Subscription-Token": BRAVE_KEY
    }
    params = {"q": query, "count": str(count)}
    resp = requests.get(BRAVE_ENDPOINT, headers=headers, params=params, timeout=15)
    resp.raise_for_status()
    data = resp.json()

    web_results = (
        data.get("web", {}).get("results") or
        data.get("results", [])
    )

    articles = []
    for idx, r in enumerate(web_results[:count], 1):
        url = r.get("url", r.get("link", ""))
        host = re.sub(r"https?://(www\\.)?", "", url).split("/")[0]
        articles.append({
            "index": idx,
            "title": r.get("title", "제목 없음"),
            "link": url,
            "snippet": r.get("description", r.get("text", "내용 없음")),
            "displayed_link": host
        })
    return articles

# ───────────────────────────── 6) 검색 → 마크다운 ─────────────────────────────
def generate_mock_search_results(query: str) -> str:
    ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    mock = [{
        "title": f"{query} 관련 가상 결과",
        "link": "https://example.com",
        "snippet": "API 호출 실패로 생성된 예시 결과입니다.",
        "displayed_link": "example.com"
    }]
    body = "\n".join(
        f"### Result {i+1}: {m['title']}\n\n{m['snippet']}\n\n"
        f"**출처**: [{m['displayed_link']}]({m['link']})\n\n---\n"
        for i, m in enumerate(mock)
    )
    return f"# 가상 검색 결과 (생성: {ts})\n\n{body}"

def do_web_search(query: str) -> str:
    """
    Brave Search 전용 검색 함수.
    실패하거나 쿼터 초과 시 mock 결과 반환.
    """
    try:
        articles = brave_search(query, count=5)
    except Exception as e:
        logging.error(f"Brave 검색 실패: {e}")
        return generate_mock_search_results(query)

    if not articles:
        return generate_mock_search_results(query)

    md_lines = []
    for a in articles:
        md_lines.append(
            f"### Result {a['index']}: {a['title']}\n\n"
            f"{a['snippet']}\n\n"
            f"**출처**: [{a['displayed_link']}]({a['link']})\n\n---\n"
        )
    header = (
        "# 웹 검색 결과\n"
        "아래 정보를 답변에 활용하세요: 출처 인용·링크 포함·다수 출처 종합\n\n"
    )
    return header + "".join(md_lines)

# ───────────────────────────── 7) 이미지·MD 변환 등 유틸 ───────────────────────
def test_image_api_connection():
    try:
        Client(IMAGE_API_URL)
        return "이미지 API 연결 성공"
    except Exception as e:
        logging.error(e)
        return f"이미지 API 연결 실패: {e}"

def generate_image(prompt, width=768, height=768, guidance=3.5,
                   inference_steps=30, seed=3):
    if not prompt:
        return None, "프롬프트 부족"
    try:
        c = Client(IMAGE_API_URL)
        res = c.predict(
            prompt=prompt, width=width, height=height,
            guidance=guidance, inference_steps=inference_steps,
            seed=seed, do_img2img=False, init_image=None,
            image2image_strength=0.8, resize_img=True,
            api_name="/generate_image"
        )
        return res[0], f"Seed: {res[1]}"
    except Exception as e:
        logging.error(e)
        return None, str(e)

def extract_image_prompt(blog_content, blog_topic):
    system = f"다음 글을 바탕으로 적절한 이미지 프롬프트를 영어로 한 줄만 써줘:\n{blog_topic}"
    try:
        res = client.messages.create(
            model="claude-3-7-sonnet-20250219",
            max_tokens=80,
            system=system,
            messages=[{"role": "user", "content": blog_content}]
        )
        return res.content[0].text.strip()
    except Exception:
        return f"A professional photo related to {blog_topic}, high quality"

def convert_md_to_html(md_text, title="Ginigen Blog"):
    body = markdown.markdown(md_text)
    return f"""<!DOCTYPE html><html><head>
<title>{title}</title><meta charset="utf-8"></head><body>{body}</body></html>"""

def extract_keywords(text: str, k: int = 5) -> str:
    txt = re.sub(r"[^가-힣a-zA-Z0-9\\s]", "", text)
    return " ".join(txt.split()[:k])

# ───────────────────────────── 8) Streamlit UI ────────────────────────────────
def chatbot_interface():
    st.title("Ginigen Blog")

    # 세션 상태 초기화
    defaults = {
        "ai_model": "claude-3-7-sonnet-20250219",
        "messages": [],
        "auto_save": True,
        "generate_image": False,
        "use_web_search": False,
        "image_api_status": test_image_api_connection()
    }
    for k, v in defaults.items():
        if k not in st.session_state:
            st.session_state[k] = v

    sb = st.sidebar
    sb.title("대화 기록 관리")
    sb.toggle("자동 저장", key="auto_save")
    sb.toggle("블로그 글 작성 후 이미지 자동 생성", key="generate_image")
    sb.toggle("주제 웹 검색 및 분석", key="use_web_search")
    sb.text(st.session_state.image_api_status)

    # 기존 메시지 렌더링
    for m in st.session_state.messages:
        with st.chat_message(m["role"]):
            st.markdown(m["content"])
            if "image" in m:
                st.image(m["image"], caption=m.get("image_caption", ""))

    # 사용자 입력
    if prompt := st.chat_input("무엇을 도와드릴까요?"):
        st.session_state.messages.append({"role": "user", "content": prompt})
        with st.chat_message("user"):
            st.markdown(prompt)

        with st.chat_message("assistant"):
            placeholder = st.empty()
            full_resp = ""
            sys_prompt = get_system_prompt()

            # (선택) Brave 검색
            if st.session_state.use_web_search:
                with st.spinner("웹 검색 중…"):
                    q = extract_keywords(prompt)
                    sb.info(f"검색어: {q}")
                    search_md = do_web_search(q)
                    if "가상 검색 결과" in search_md:
                        sb.warning("실제 검색 결과를 가져오지 못했습니다.")
                    sys_prompt += f"\n\n검색 결과:\n{search_md}\n"

            # Claude 스트리밍
            with client.messages.stream(
                model=st.session_state.ai_model,
                max_tokens=MAX_TOKENS,
                system=sys_prompt,
                messages=[{"role": m["role"], "content": m["content"]}
                          for m in st.session_state.messages]
            ) as stream:
                for t in stream.text_stream:
                    full_resp += t or ""
                    placeholder.markdown(full_resp + "▌")
            placeholder.markdown(full_resp)

            # (선택) 이미지 생성
            if st.session_state.generate_image:
                with st.spinner("이미지 생성 중…"):
                    img_prompt = extract_image_prompt(full_resp, prompt)
                    img, caption = generate_image(img_prompt)
                    if img:
                        st.image(img, caption=caption)
                        st.session_state.messages.append(
                            {"role": "assistant", "content": full_resp,
                             "image": img, "image_caption": caption}
                        )
                    else:
                        st.error(f"이미지 생성 실패: {caption}")
                        st.session_state.messages.append(
                            {"role": "assistant", "content": full_resp}
                        )
            else:
                st.session_state.messages.append(
                    {"role": "assistant", "content": full_resp}
                )

            # 다운로드 버튼
            st.subheader("이 블로그 다운로드:")
            c1, c2 = st.columns(2)
            c1.download_button("마크다운", full_resp,
                               file_name=f"{prompt[:30]}.md", mime="text/markdown")
            html = convert_md_to_html(full_resp, prompt[:30])
            c2.download_button("HTML", html,
                               file_name=f"{prompt[:30]}.html", mime="text/html")

    # 자동 저장
    if st.session_state.auto_save and st.session_state.messages:
        try:
            fname = f"chat_history_{datetime.now():%Y%m%d_%H%M%S}.json"
            with open(fname, "w", encoding="utf-8") as f:
                json.dump(st.session_state.messages, f, ensure_ascii=False, indent=2)
        except Exception as e:
            sb.error(f"자동 저장 오류: {e}")

# ───────────────────────────── 9) main ────────────────────────────────────────
def main():
    chatbot_interface()

if __name__ == "__main__":
    # requirements.txt 생성
    with open("requirements.txt", "w") as f:
        f.write("\n".join([
            "streamlit>=1.31.0",
            "anthropic>=0.18.1",
            "gradio-client>=1.8.0",
            "requests>=2.32.3",
            "markdown>=3.5.1",
            "pillow>=10.1.0"
        ]))
    main()