# ──────────────────────────────── Imports ────────────────────────────────
import os, json, re, logging, requests, markdown, time
from datetime import datetime
import streamlit as st
import anthropic
from gradio_client import Client
# from bs4 import BeautifulSoup # 필요 시 주석 해제
# ──────────────────────────────── 환경 변수 / 상수 ───────────────────────────
ANTHROPIC_KEY = os.getenv("API_KEY", "")
BRAVE_KEY = os.getenv("SERPHOUSE_API_KEY", "") # 이름 유지
BRAVE_ENDPOINT = "https://api.search.brave.com/res/v1/web/search"
IMAGE_API_URL = "http://211.233.58.201:7896"
MAX_TOKENS = 7_999
# 블로그 템플릿 및 스타일 정의
BLOG_TEMPLATES = {
"standard": "전문 블로그 작성 전문가로서 8단계 프레임워크를 따라 자연스럽고 매력적인 글 작성",
"tutorial": "단계별 튜토리얼 형식으로, 명확한 과정과 결과를 보여주는 가이드 작성",
"review": "제품/서비스 분석 중심의 리뷰 형식, 장단점 분석과 추천 포함",
"storytelling": "개인 경험이나 사례를 중심으로 한 스토리텔링 형식의 블로그 작성",
"seo_optimized": "검색엔진 최적화(SEO)를 고려한 키워드 중심 블로그 작성"
}
BLOG_TONES = {
"professional": "전문적이고 공식적인 어조로 작성",
"casual": "친근하고 대화체 중심의 편안한 톤으로 작성",
"humorous": "유머와 재치를 가미한 가벼운 어조로 작성",
"storytelling": "이야기를 들려주듯 감성적이고 몰입감 있는 톤으로 작성"
}
# ──────────────────────────────── 로깅 ──────────────────────────────────────
logging.basicConfig(level=logging.INFO,
format="%(asctime)s - %(levelname)s - %(message)s")
# ──────────────────────────────── Anthropic Client ─────────────────────────
client = anthropic.Anthropic(api_key=ANTHROPIC_KEY)
# ──────────────────────────────── 블로그 작성 시스템 프롬프트 ────────────────
def get_system_prompt(template="standard", tone="professional", word_count=1750) -> str:
base_prompt = """
당신은 전문 블로그 작성 전문가입니다. 모든 블로그 글 작성 요청에 대해 다음의 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. 소통 채널 안내
"""
# 템플릿별 추가 지침
template_guides = {
"tutorial": """
이 블로그는 튜토리얼 형식으로 작성해 주세요:
- 명확한 목표와 최종 결과물 먼저 제시
- 단계별로 명확하게 구분된 과정 설명
- 각 단계마다 이미지를 삽입할 위치 표시
- 예상 소요 시간과 난이도 명시
- 필요한 도구나 사전 지식 안내
- 문제해결 팁과 자주 발생하는 실수 포함
- 완료 후 다음 단계나 응용법 제안
""",
"review": """
이 블로그는 리뷰 형식으로 작성해 주세요:
- 객관적 사실과 주관적 평가 구분
- 명확한 평가 기준 제시
- 장점과 단점 균형있게 서술
- 유사 제품/서비스와 비교
- 누구에게 적합한지 타겟 설명
- 구체적인 사용 경험과 결과 포함
- 최종 추천 여부와 대안 제시
""",
"storytelling": """
이 블로그는 스토리텔링 형식으로 작성해 주세요:
- 실제 인물이나 사례로 시작
- 문제 상황과 감정적 연결 강화
- 갈등과 해결과정 중심의 내러티브
- 교훈과 배움을 자연스럽게 포함
- 독자가 공감할 수 있는 감정선 유지
- 이야기와 유용한 정보의 균형 유지
- 독자에게 자신의 이야기를 생각해보게 유도
""",
"seo_optimized": """
이 블로그는 SEO 최적화 형식으로 작성해 주세요:
- 핵심 키워드를 제목, 소제목, 첫 단락에 배치
- 관련 키워드를 자연스럽게 본문에 분산
- 300-500자 분량의 명확한 단락 구성
- 질문 형식의 소제목 활용
- 목록, 표, 강조 텍스트 등 다양한 서식 활용
- 내부 링크 삽입 위치 표시
- 2000-3000자 이상의 충분한 콘텐츠 제공
"""
}
# 톤별 추가 지침
tone_guides = {
"professional": "전문적이고 권위있는 어조로 작성하되, 전문 용어는 적절히 설명해 주세요. 데이터와 연구 결과를 중심으로 논리적 흐름을 유지하세요.",
"casual": "친근하고 대화하듯 편안한 어조로 작성해 주세요. '~네요', '~해요' 같은 대화체를 사용하고, 개인적 경험과 비유를 통해 내용을 전달하세요.",
"humorous": "유머와 재치있는 표현을 적절히 활용해 주세요. 재미있는 비유나 예시, 가벼운 농담을 포함하되, 정보의 정확성과 유용성은 유지하세요.",
"storytelling": "이야기를 들려주듯 감성적이고 몰입감 있는 톤으로 작성해 주세요. 인물, 배경, 갈등, 해결과정이 담긴 내러티브 구조를 활용하세요."
}
# 최종 프롬프트 조합
final_prompt = base_prompt
# 선택된 템플릿 지침 추가
if template in template_guides:
final_prompt += "\n" + template_guides[template]
# 선택된 톤 지침 추가
if tone in tone_guides:
final_prompt += f"\n\n톤앤매너: {tone_guides[tone]}"
# 글자 수 지침 추가
final_prompt += f"\n\n작성 시 준수사항\n9.1. 글자 수: {word_count-250}-{word_count+250}자 내외\n9.2. 문단 길이: 3-4문장 이내\n9.3. 시각적 구분: 소제목, 구분선, 번호 목록 활용\n9.4. 데이터: 모든 정보의 출처 명시\n9.5. 가독성: 명확한 단락 구분과 강조점 사용"
return final_prompt
# ──────────────────────────────── Brave Search API ─────────────────────────
def brave_search(query: str, count: int = 5):
"""
Brave Web Search API 호출 → list[dict]
반환 필드: index, title, link, snippet, displayed_link
"""
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)}
for attempt in range(3): # 최대 3번 재시도
try:
r = requests.get(BRAVE_ENDPOINT, headers=headers, params=params, timeout=15)
r.raise_for_status()
data = r.json()
# 결과 형식 확인 및 로깅
logging.info(f"Brave 검색 결과 데이터 구조: {list(data.keys())}")
raw = data.get("web", {}).get("results") or data.get("results", [])
if not raw:
logging.warning(f"Brave 검색 결과 없음. 응답: {data}")
raise ValueError("검색 결과가 없습니다")
arts = []
for i, res in enumerate(raw[:count], 1):
url = res.get("url", res.get("link", ""))
host = re.sub(r"https?://(www\.)?", "", url).split("/")[0]
arts.append({
"index": i,
"title": res.get("title", "제목 없음"),
"link": url,
"snippet": res.get("description", res.get("text", "내용 없음")),
"displayed_link": host
})
logging.info(f"Brave 검색 성공: {len(arts)}개 결과")
return arts
except Exception as e:
logging.error(f"Brave 검색 실패 (시도 {attempt+1}/3): {e}")
if attempt < 2: # 마지막 시도가 아니면 대기 후 재시도
time.sleep(2)
return [] # 모든 시도 실패 시 빈 목록 반환
def mock_results(query: str) -> str:
"""검색 API 실패 시 가상 검색 결과 제공"""
ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
return (f"# 검색 결과 대체 내용 (생성: {ts})\n\n"
f"검색 API 호출이 실패했습니다. 주제 '{query}'에 대해 기존 지식을 활용해 답변해 주세요.\n\n"
f"다음 내용이 도움이 될 수 있습니다:\n\n"
f"- {query}의 기본 개념과 중요성\n"
f"- 일반적으로 알려진 관련 통계와 트렌드\n"
f"- 해당 주제에 대한 전문가들의 일반적인 견해\n"
f"- 독자들이 실제로 궁금해할 만한 질문들\n\n"
f"참고: 이 내용은 실시간 검색 결과가 아닌 대체 안내입니다.\n\n")
def do_web_search(query: str) -> str:
"""웹 검색 수행 및 결과 포맷팅"""
try:
arts = brave_search(query, 5)
if not arts:
logging.warning("검색 결과 없음, 대체 콘텐츠 사용")
return mock_results(query)
hdr = "# 웹 검색 결과\n아래 정보를 참고해서 답변하세요.\n\n"
body = "\n".join(
f"### Result {a['index']}: {a['title']}\n\n{a['snippet']}\n\n"
f"**출처**: [{a['displayed_link']}]({a['link']})\n\n---\n"
for a in arts
)
return hdr + body
except Exception as e:
logging.error(f"웹 검색 전체 프로세스 실패: {str(e)}")
return mock_results(query)
# ──────────────────────────────── 이미지 · 변환 유틸 ────────────────────────
def generate_image(prompt, w=768, h=768, g=3.5, steps=30, seed=3):
if not prompt: return None, "프롬프트 부족"
try:
res = Client(IMAGE_API_URL).predict(
prompt=prompt, width=w, height=h, guidance=g,
inference_steps=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: str, topic: str):
sys = f"다음 글로부터 영어 1줄 이미지 프롬프트 생성:\n{topic}"
try:
res = client.messages.create(
model="claude-3-7-sonnet-20250219",
max_tokens=80, system=sys,
messages=[{"role": "user", "content": blog}]
)
return res.content[0].text.strip()
except Exception:
return f"A professional photo related to {topic}, high quality"
def md_to_html(md: str, title="Ginigen Blog"):
return f"{title}{markdown.markdown(md)}"
def keywords(text: str, top=5):
return " ".join(re.sub(r"[^가-힣a-zA-Z0-9\\s]", "", text).split()[:top])
# ──────────────────────────────── Streamlit UI ────────────────────────────
def ginigen_app():
st.title("Ginigen Blog")
# 세션 기본값
defaults = dict(
ai_model="claude-3-7-sonnet-20250219",
messages=[],
auto_save=True,
generate_image=False,
use_web_search=False,
blog_template="standard",
blog_tone="professional",
word_count=1750
)
for k, v in defaults.items():
st.session_state.setdefault(k, v)
# ── 사이드바 컨트롤
sb = st.sidebar
sb.title("블로그 설정")
# 블로그 템플릿 및 스타일 선택
sb.subheader("블로그 스타일 설정")
sb.selectbox("블로그 템플릿", options=list(BLOG_TEMPLATES.keys()),
format_func=lambda x: x.replace("_", " ").title(),
key="blog_template")
sb.selectbox("블로그 톤", options=list(BLOG_TONES.keys()),
format_func=lambda x: x.replace("_", " ").title(),
key="blog_tone")
sb.slider("블로그 길이 (단어 수)", 800, 3000, 1750, key="word_count")
sb.subheader("기타 설정")
sb.toggle("자동 저장", key="auto_save")
sb.toggle("이미지 자동 생성", key="generate_image")
# 웹 검색 토글 (모니터링을 위해 유지하되 기본값은 False)
search_enabled = sb.toggle("웹 검색 사용", value=False, key="use_web_search")
if search_enabled:
st.warning("⚠️ 웹 검색 기능은 현재 불안정할 수 있습니다. 검색 결과가 없으면 기본 지식으로 대체됩니다.")
# ── 최근 블로그 다운로드 (마크다운 / HTML)
latest_blog = next(
(m["content"] for m in reversed(st.session_state.messages)
if m["role"] == "assistant" and m["content"].strip()), None)
if latest_blog:
title = re.search(r"# (.*?)(\n|$)", latest_blog)
title = title.group(1).strip() if title else "blog"
sb.subheader("최근 블로그 다운로드")
c1, c2 = sb.columns(2)
c1.download_button("Markdown", latest_blog,
file_name=f"{title}.md", mime="text/markdown")
c2.download_button("HTML", md_to_html(latest_blog, title),
file_name=f"{title}.html", mime="text/html")
# ── JSON 대화 기록 업로드
up = sb.file_uploader("대화 기록 불러오기 (.json)", type=["json"])
if up:
try:
st.session_state.messages = json.load(up)
sb.success("대화 기록 불러오기 완료")
except Exception as e:
sb.error(f"불러오기 실패: {e}")
# ── JSON 대화 기록 다운로드
if sb.button("대화 기록 JSON 다운로드"):
sb.download_button("저장", json.dumps(st.session_state.messages,
ensure_ascii=False, indent=2),
file_name="chat_history.json",
mime="application/json")
# ── 기존 메시지 렌더링
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(); answer = ""
# 선택된 템플릿, 톤, 단어 수로 시스템 프롬프트 생성
sys_prompt = get_system_prompt(
template=st.session_state.blog_template,
tone=st.session_state.blog_tone,
word_count=st.session_state.word_count
)
if st.session_state.use_web_search:
with st.spinner("웹 검색 중…"):
search_md = do_web_search(keywords(prompt))
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:
answer += t or ""
placeholder.markdown(answer + "▌")
placeholder.markdown(answer)
# 이미지 옵션
if st.session_state.generate_image:
with st.spinner("이미지 생성 중…"):
ip = extract_image_prompt(answer, prompt)
img, cap = generate_image(ip)
if img:
st.image(img, caption=cap)
st.session_state.messages.append(
{"role": "assistant", "content": answer,
"image": img, "image_caption": cap})
answer_entry_saved = True
if not st.session_state.generate_image:
st.session_state.messages.append(
{"role": "assistant", "content": answer})
# 본문 다운로드 버튼 (MD / HTML)
st.subheader("이 블로그 다운로드")
b1, b2 = st.columns(2)
b1.download_button("Markdown", answer,
file_name=f"{prompt[:30]}.md", mime="text/markdown")
b2.download_button("HTML", md_to_html(answer, prompt[:30]),
file_name=f"{prompt[:30]}.html", mime="text/html")
# ── 자동 백업 저장
if st.session_state.auto_save and st.session_state.messages:
try:
fn = f"chat_history_auto_{datetime.now():%Y%m%d_%H%M%S}.json"
with open(fn, "w", encoding="utf-8") as fp:
json.dump(st.session_state.messages, fp,
ensure_ascii=False, indent=2)
except Exception as e:
logging.error(f"자동 저장 실패: {e}")
# ──────────────────────────────── main / requirements ──────────────────────
def main(): ginigen_app()
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()