# ──────────────────────────────── Imports ────────────────────────────────
import os, json, re, logging, requests, markdown, time, io
from datetime import datetime
import streamlit as st
from openai import OpenAI # OpenAI 라이브러리
from gradio_client import Client
import pandas as pd
import PyPDF2 # For handling PDF files
# ──────────────────────────────── Environment Variables / Constants ─────────────────────────
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY", "")
BRAVE_KEY = os.getenv("SERPHOUSE_API_KEY", "") # Keep this name
BRAVE_ENDPOINT = "https://api.search.brave.com/res/v1/web/search"
IMAGE_API_URL = "http://211.233.58.201:7896"
MAX_TOKENS = 7999
# Blog template and style definitions (in English)
BLOG_TEMPLATES = {
"ginigen": "Recommended style by Ginigen", # ← 복구
"standard": "Standard 8-step framework blog",
"tutorial": "Step-by-step tutorial format",
"review": "Product/service review format",
"storytelling": "Storytelling format",
"seo_optimized": "SEO-optimized blog",
# New specialized templates
"insta": "Instagram Reels script",
"thread": "SNS Thread post",
"shortform": "60-sec Short-form video",
"youtube": "YouTube script",
}
# ───────── Blog tone definitions ─────────
BLOG_TONES = {
"professional": "Professional and formal tone",
"casual": "Friendly and conversational tone",
"humorous": "Humorous approach",
"storytelling": "Story-driven approach",
}
# Example blog topics
EXAMPLE_TOPICS = {
"example1": "Changes to the real estate tax system in 2025: Impact on average households and tax-saving strategies",
"example2": "Summer festivals in 2025: A comprehensive guide to major regional events and hidden attractions",
"example3": "Emerging industries to watch in 2025: An investment guide focused on AI opportunities"
}
# ──────────────────────────────── Logging ────────────────────────────────
logging.basicConfig(level=logging.INFO,
format="%(asctime)s - %(levelname)s - %(message)s")
# ──────────────────────────────── OpenAI Client ──────────────────────────
# OpenAI 클라이언트에 타임아웃과 재시도 로직 추가
@st.cache_resource
def get_openai_client():
"""Create an OpenAI client with timeout and retry settings."""
if not OPENAI_API_KEY:
raise RuntimeError("⚠️ OPENAI_API_KEY 환경 변수가 설정되지 않았습니다.")
return OpenAI(
api_key=OPENAI_API_KEY,
timeout=60.0, # 타임아웃 60초로 설정
max_retries=3 # 재시도 횟수 3회로 설정
)
# ──────────────────────────────── Blog Creation System Prompt ─────────────
def get_system_prompt(template="ginigen", tone="professional", word_count=1750, include_search_results=False, include_uploaded_files=False) -> str:
"""
Generate a system prompt that includes:
- The 8-step blog writing framework
- The selected template and tone
- Guidelines for using web search results and uploaded files
"""
# Ginigen recommended style prompt (English version)
ginigen_prompt = """
당신은 뛰어난 한국어 SEO 카피라이터입니다.
◆ 목적
'Blog Template'의 선택에 따라 블로그 또는 릴스, 쓰레드, 유튜브 관련 전문 글을 작성하여야 한다.
항상 **[핵심부터 제시 → 간결‧명료하게 → 독자 혜택 강조 → 행동 유도]**의 4원칙을 따르세요.
◆ 완성 형식 (Markdown 사용, 불필요한 설명 금지)
제목
이모지 + 궁금증 질문/감탄사 + 핵심 키워드 (70자 이내)
예시: # 🧬 염증만 줄여도 살이 빠진다?! 퀘르세틴 5가지 놀라운 효능
Hook (2~3줄)
문제 제시 → 해결 키워드 언급 → 이 글을 읽어야 하는 이유 요약
--- 구분선
섹션 1: 핵심 개념 소개
## 🍏 [키워드]란 무엇인가?
1~2문단 정의 + 📌 한줄 요약
---
섹션 2: 5가지 이점/이유
## 💪 [키워드]가 유익한 5가지 이유
각 소제목 형식:
1. [키워드 중심 소제목]
1~2문단 설명
✔ 핵심 포인트 한줄 강조
총 5개 항목
섹션 3: 섭취/활용 방법
## 🥗 [키워드] 제대로 활용하는 법!
이모지 불릿 5개 정도 + 추가 팁
---
마무리 행동 유도
## 📌 결론 – 지금 바로 [키워드] 시작하세요!
2~3문장으로 혜택/변화를 요약 → 행동 촉구 (구매, 구독, 공유 등)
---
핵심 요약 표
항목 효과
[키워드] [효과 요약]
주요 음식/제품 [목록]
---
퀴즈 & CTA
간단한 Q&A 퀴즈 (1문항) → 정답 공개
“도움이 되셨다면 공유/댓글 부탁드립니다” 문구
다음 글 예고
◆ 추가 지침
전체 분량 1,200~1,800단어.
쉬운 어휘·짧은 문장 사용, 이모지·굵은 글씨·인용으로 가독성 강화.
구체적 수치, 연구 결과, 비유로 신뢰도 ↑.
“프롬프트”, “지시사항” 등 메타 언급 금지.
대화체이면서도 전문성을 유지.
외부 출처가 없다면 “연구에 따르면” 같은 표현 최소화.
◆ 출력
위 형식을 따른 완성 블로그 글만 반환하세요. 추가 설명은 포함하지 않습니다.
"""
# Standard 8-step framework (English version)
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 = {
"insta": """
너는 인스타그램 릴스 스크립트(대본) 생성 전문가 역할이다 :
블로그 스타일로 생성하지 말고, 너는 다음 지침만을 따라 글을 작성하여야 한다.
당신은 **〈Universal Reels Strategist GPT〉**다.
목표: 사용자가 제시한 주제·제품·서비스를 바탕으로 저장‧공유‧행동을 유도하는 60초 이하 숏폼 영상을 **한 번에 완성**해 주는 것.
────────────── 기본 원칙 ──────────────
1. **본능 4대 욕구 연결**
① 돈·시간 절약(생존)
② 건강·아름다움(생존+미적 만족)
③ 인간관계·사랑·사회적 인정
④ 문제 해결·성장(능력·지식 향상)
→ 최소 1개 이상과 사용자 주제를 연결해라.
2. **표본 이론(대중화 확장)**
• 주제가 좁으면 ‘누구에게나 적용 가능한 실익’으로 넓혀라.
예) 지방 소형 헬스장 홍보 → “하루 5분 뱃살 태우는 홈트”.
3. **6단계 제작 프로세스**
① 레퍼런스·경쟁 사례 분석
② 주제·포지셔닝 확정(표본 확장 포함)
③ 후킹+시퀀스 스크립트 작성 (아래 아웃풋 형식 사용)
④ 촬영·편집 가이드(필요 장비·구도·BGM 등)
⑤ 카피 보완(제목·본문·캡션)
⑥ 행동 유도 문구(CTA) 삽입
4. **후킹 3초 규칙**
• 시작 3초 안에 **논란, 호기심, 수치화된 이득** 중 하나를 폭발적으로 제시.
• 숫자·구체 단어·강한 동사 사용. (예: “7일 만에 매출 두 배?”)
5. **CTA 필수**
• 저장, 공유, 댓글, 구매, 신청, 예약 등 최소 1개를 명시적 문장으로 요구.
6. **톤‧스타일**
• 친구처럼 직설·간결.
• 불필요한 이모지·특수문자 금지(‘!’ ‘?’ 만 허용).
• 한국어가 기본이지만, 사용자가 영어로 요청하면 동일 규칙을 영어로 제공.
7. **정보 수집**
• 업종·타깃·전환 목표·예산·촬영 가능 장비가 불명확하면 **한 번에 묶어** 물어본다.
8. **출력 형식** (모든 항목은 1~2줄 내외, 번호 그대로 유지)
1) 제목(20자 이하)
2) 후킹 대사(첫 3초)
3) 시퀀스 스크립트(장면별 핵심 대사·자막)
4) 핵심 메시지 요약
5) CTA 문구
6) 캡션 예시(이득→공감→행동, 3문장)
7) 해시태그(쉼표로 구분, 특수문자 제외)
8) 촬영·편집 팁(필요시)
9. **검증 체크리스트**
• 본능 자극 포인트 존재?
• 후킹 3초 규칙 충족?
• CTA 포함? → 하나라도 ‘아니오’면 스스로 수정 후 출력.
───────── 예시 입력 & 요약 출력 ─────────
사용자: “주제: 1인 세무사 사무실 신규 고객 확보”
GPT 출력(요약):
1) 제목: 세무비용 30% 줄이는 법
2) 후킹: “10분 전화로 세금 300만 원 아꼈어요?”
3) 시퀀스: 장면1 세금고지서 쇼크 → “① 불필요 공제 찾기” …
… 이하 형식 동일
────────────────────────────
**모든 답변은 위 규칙을 어기면 자동으로 재검토하고 수정하라.**
데이터가 없는것은 웹검색으로 정보를 서치해서 찾아내야 한다.
""",
"thread": """
너는 쓰레드 포스트 생성 전문가 역할이다 :
블로그 스타일로 생성하지 말고, 너는 다음 지침만을 따라 글을 작성하여야 한다.
You are a Korean tech‑savvy copywriter who writes short, hype‑driven SNS thread posts.
When given a {product_name} and its {key_highlights}, output a thread in the following style:
[1] 시작
– 한 줄 훅: 🔥 같은 이모지 + 타깃 독자 소환 + 짧은 감탄
– 두 번째 줄: “{product_name}가/이 진짜 일 냈다” 또는 동등한 임팩트 문장
[2] 정의 & 맥락
– “{unique_point}? 그게 뭐야?” 식 질문
– 1~2문장으로 개념 설명, 세계적 사례·레퍼런스 한 줄
[3] numbered 핵심 포인트
– 각 포인트는 “{번호}/ {소제목}” 형식
– 이후 1~3줄로 {소제목}를 상세 설명
– 설명은 구어체, 문장 짧게, ‘!’ 활용
– 구체 예시·비교·데이터를 포함하되 한 문단 ≤3줄
– 최소 3개, 최대 6개 포인트
[4] 결론
– “{마지막번호+1}/ 결론 : …” 형식
– 문제 해결·가치 요약
– ‘‘이제 {call_to_action}’’ 식 직접 행동 유도
스타일 규칙:
- 한국어 위주, 필요 시 영어 기술용어 그대로 삽입
- 문장마다 엔터, 블록 단락 구분
- 특수문자는 ‘!’ ‘?’ 외 최소화
- 전체 길이 250~400자
- 이모지는 제목·중요 포인트에만 1~3개 사용
- 존댓말 대신 친근한 반말
""",
"shortform": """
너는 숏폼 스크립트(대본) 생성 전문가 역할이다 :
블로그 스타일로 생성하지 말고, 너는 다음 지침만을 따라 글을 작성하여야 한다.
### 🎛️ GPTS 시스템 프롬프트 — 1분 숏폼 영상 대본 작성기
너는 **“1 분 숏폼 영상 대본 자동화 AI”**다.
사용자가 주제·제품·서비스·타깃 시청자·톤(선택)을 입력하면, 아래 포맷을 **한국어**로 완성된 대본으로 출력한다.
- 총 길이는 **60 초 이내**.
- 각 구간은 **타임코드(초)**와 **구간명**을 대괄호로 표기.
- 문장은 짧고 임팩트 있게, 1 문장 ≈ 1.5 초 기준.
- 이모지 사용은 자유지만 과도하지 않게(0–2개).
- 특수문자는 ‘!’와 ‘?’만 허용.
🟡 **출력 포맷**
[0-3초 | Hook]
{시청자 스크롤을 멈출 한마디}
[4-15초 | Problem]
{시청자 공감 포인트를 정확히 짚기}
[16-30초 | Solution]
{제품/서비스/아이디어 소개 + 핵심 기능}
[31-45초 | Proof]
{효과 증명·데이터·후기 + 경쟁 제품과 차별점}
[46-55초 | Callback/Emotion]
{Hook를 자연스럽게 회수하거나 감정 자극}
[56-60초 | CTA]
{구매·클릭·팔로우 등 명확한 행동 유도}
🟡 **작성 규칙**
1. **Hook** – 놀라움·궁금증·공감 중 하나로 강렬한 한 문장.
2. **Problem** – 대상 시청자의 불편·고민을 구체적으로 언급.
3. **Solution** – 제품·서비스로 문제 해결, 핵심 기능을 쉬운 표현으로.
4. **Proof** – 수치·후기·전문가 언급 등 신뢰 요소 1-2개 + 차별점.
5. **Callback/Emotion** – 훅을 변주하거나 희망·긴급 감정 자극.
6. **CTA** – 구체적 행동 + 한정성·긴급성 언어.
🟡 **프롬프트 입력 예시**
주제: 스마트 무선 청소기
톤: 친근하고 유머러스
-사용자의 영상 목적(예:제품 홍보, 사용법 안내, 유용성 설명 등)과 타깃 시청자 그리고 시청자에게 전달하고 싶은 주요 메시지에 대한 정보를 받을수 있어야해 답변 예시도 함께 보여주고
-최대 4개의 이모지를 사용해줘
-시나리오는 영상과 대본을 구분할수 있게 출력해줘
-결과물 출력시 하단에 따로 이미지도 함께 생성해줘 관련 배경이미지로 생성하되 제품은 생성하지말것. 그리고 캡션/카피라이팅등 텍스트를 이미지 안에 절대 생성 하지마
""",
"youtube": """
너는 유튜브 스크립트(대본) 생성 전문가 역할이다 :
블로그 스타일로 생성하지 말고, 너는 다음 지침만을 따라 글을 작성하여야 한다.
"""
}
# 어조(톤)별 추가 지침
tone_guides = {
"professional": "전문적이고 권위 있는 문체를 사용합니다. 기술 용어를 명확히 설명하고, 데이터나 연구 결과를 제시하여 논리적 흐름을 유지하세요.",
"casual": "편안하고 대화체에 가까운 스타일을 사용합니다. 개인 경험·공감 가는 예시를 들고, 친근한 어조(예: '정말 유용해요!')를 활용하세요.",
"humorous": "유머와 재치 있는 표현을 사용합니다. 재미있는 비유나 농담을 추가하되, 정확성과 유용성을 유지하세요.",
"storytelling": "이야기를 들려주듯 서술합니다. 감정 깊이와 서사적 흐름을 유지하고, 인물·배경·갈등·해결을 녹여내세요."
}
# 웹 검색 결과 사용 지침
search_guide = """
[웹 검색 결과 활용 가이드]
- 검색 결과의 핵심 정보를 블로그에 정확히 통합하세요.
- 최신 데이터, 통계, 사례를 포함하세요.
- 인용 시 본문에서 출처를 명확히 표기하세요 (예: "XYZ 웹사이트에 따르면 …").
- 글 마지막에 '참고 자료' 섹션을 두고 주요 출처와 링크를 나열하세요.
- 상반되는 정보가 있다면 다양한 관점을 함께 제시하세요.
- 최신 트렌드와 데이터를 반드시 반영하세요.
"""
# 업로드 파일 사용 지침
upload_guide = """
[업로드된 파일 활용 지침 (최우선)]
- 업로드된 파일은 블로그의 핵심 정보원이어야 합니다.
- 파일 속 데이터·통계·예시를 면밀히 검토해 통합하세요.
- 주요 수치·주장은 직접 인용하고 충분히 설명하세요.
- 파일 내용을 블로그의 핵심 요소로 강조하세요.
- 출처를 명확히 표기하세요 (예: "업로드된 데이터에 따르면 …").
- CSV 파일은 중요한 수치나 통계를 상세히 다루세요.
- PDF 파일은 핵심 문장이나 진술을 인용하세요.
- 텍스트 파일의 관련 내용을 효과적으로 통합하세요.
- 파일 내용이 다소 벗어나 보여도 주제와 연결고리를 찾아 서술하세요.
- 글 전반에 걸쳐 일관되게 파일 데이터를 반영하세요.
"""
# Choose base prompt
if template == "ginigen":
final_prompt = ginigen_prompt
else:
final_prompt = base_prompt
# If the user chose a specific template (and not ginigen), append the relevant guidelines
if template != "ginigen" and template in template_guides:
final_prompt += "\n" + template_guides[template]
# If a specific tone is selected, append that guideline
if tone in tone_guides:
final_prompt += f"\n\nTone and Manner: {tone_guides[tone]}"
# If web search results should be included
if include_search_results:
final_prompt += f"\n\n{search_guide}"
# If uploaded files should be included
if include_uploaded_files:
final_prompt += f"\n\n{upload_guide}"
# Word count guidelines
final_prompt += (
f"\n\nWriting Requirements:\n"
f"9.1. Word Count: around {word_count-250}-{word_count+250} characters\n"
f"9.2. Paragraph Length: 3-4 sentences each\n"
f"9.3. Visual Cues: Use subheadings, separators, and bullet/numbered lists\n"
f"9.4. Data: Cite all sources\n"
f"9.5. Readability: Use clear paragraph breaks and highlights where necessary"
)
return final_prompt
# ──────────────────────────────── Brave Search API ────────────────────────
@st.cache_data(ttl=3600)
def brave_search(query: str, count: int = 20):
"""
Call the Brave Web Search API → list[dict]
Returns fields: index, title, link, snippet, displayed_link
"""
if not BRAVE_KEY:
raise RuntimeError("⚠️ SERPHOUSE_API_KEY (Brave API Key) environment variable is empty.")
headers = {
"Accept": "application/json",
"Accept-Encoding": "gzip",
"X-Subscription-Token": BRAVE_KEY
}
params = {"q": query, "count": str(count)}
for attempt in range(3):
try:
r = requests.get(BRAVE_ENDPOINT, headers=headers, params=params, timeout=15)
r.raise_for_status()
data = r.json()
logging.info(f"Brave search result data structure: {list(data.keys())}")
raw = data.get("web", {}).get("results") or data.get("results", [])
if not raw:
logging.warning(f"No Brave search results found. Response: {data}")
raise ValueError("No search results found.")
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", "No title"),
"link": url,
"snippet": res.get("description", res.get("text", "No snippet")),
"displayed_link": host
})
logging.info(f"Brave search success: {len(arts)} results")
return arts
except Exception as e:
logging.error(f"Brave search failure (attempt {attempt+1}/3): {e}")
if attempt < 2:
time.sleep(2)
return []
def mock_results(query: str) -> str:
"""Fallback search results if API fails"""
ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
return (f"# Fallback Search Content (Generated: {ts})\n\n"
f"The search API request failed. Please generate the blog based on any pre-existing knowledge about '{query}'.\n\n"
f"You may consider the following points:\n\n"
f"- Basic concepts and importance of {query}\n"
f"- Commonly known related statistics or trends\n"
f"- Typical expert opinions on this subject\n"
f"- Questions that readers might have\n\n"
f"Note: This is fallback guidance, not real-time data.\n\n")
def do_web_search(query: str) -> str:
"""Perform web search and format the results."""
try:
arts = brave_search(query, 20)
if not arts:
logging.warning("No search results, using fallback content")
return mock_results(query)
hdr = "# Web Search Results\nUse the information below to enhance the reliability of your blog. When you quote, please cite the source, and add a References section at the end of the blog.\n\n"
body = "\n".join(
f"### Result {a['index']}: {a['title']}\n\n{a['snippet']}\n\n"
f"**Source**: [{a['displayed_link']}]({a['link']})\n\n---\n"
for a in arts
)
return hdr + body
except Exception as e:
logging.error(f"Web search process failed: {str(e)}")
return mock_results(query)
# ──────────────────────────────── File Upload Handling ─────────────────────
def process_text_file(file):
"""Handle text file"""
try:
content = file.read()
file.seek(0)
text = content.decode('utf-8', errors='ignore')
if len(text) > 10000:
text = text[:9700] + "...(truncated)..."
result = f"## Text File: {file.name}\n\n"
result += text
return result
except Exception as e:
logging.error(f"Error processing text file: {str(e)}")
return f"Error processing text file: {str(e)}"
def process_csv_file(file):
"""Handle CSV file"""
try:
content = file.read()
file.seek(0)
df = pd.read_csv(io.BytesIO(content))
result = f"## CSV File: {file.name}\n\n"
result += f"- Rows: {len(df)}\n"
result += f"- Columns: {len(df.columns)}\n"
result += f"- Column Names: {', '.join(df.columns.tolist())}\n\n"
result += "### Data Preview\n\n"
preview_df = df.head(10)
try:
markdown_table = preview_df.to_markdown(index=False)
if markdown_table:
result += markdown_table + "\n\n"
else:
result += "Unable to display CSV data.\n\n"
except Exception as e:
logging.error(f"Markdown table conversion error: {e}")
result += "Displaying data as text:\n\n"
result += str(preview_df) + "\n\n"
num_cols = df.select_dtypes(include=['number']).columns
if len(num_cols) > 0:
result += "### Basic Statistical Information\n\n"
try:
stats_df = df[num_cols].describe().round(2)
stats_markdown = stats_df.to_markdown()
if stats_markdown:
result += stats_markdown + "\n\n"
else:
result += "Unable to display statistical information.\n\n"
except Exception as e:
logging.error(f"Statistical info conversion error: {e}")
result += "Unable to generate statistical information.\n\n"
return result
except Exception as e:
logging.error(f"CSV file processing error: {str(e)}")
return f"Error processing CSV file: {str(e)}"
def process_pdf_file(file):
"""Handle PDF file"""
try:
# Read file in bytes
file_bytes = file.read()
file.seek(0)
# Use PyPDF2
pdf_file = io.BytesIO(file_bytes)
reader = PyPDF2.PdfReader(pdf_file, strict=False)
# Basic info
result = f"## PDF File: {file.name}\n\n"
result += f"- Total pages: {len(reader.pages)}\n\n"
# Extract text by page (limit to first 5 pages)
max_pages = min(5, len(reader.pages))
all_text = ""
for i in range(max_pages):
try:
page = reader.pages[i]
page_text = page.extract_text()
current_page_text = f"### Page {i+1}\n\n"
if page_text and len(page_text.strip()) > 0:
# Limit to 1500 characters per page
if len(page_text) > 1500:
current_page_text += page_text[:1500] + "...(truncated)...\n\n"
else:
current_page_text += page_text + "\n\n"
else:
current_page_text += "(No text could be extracted from this page)\n\n"
all_text += current_page_text
# If total text is too long, break
if len(all_text) > 8000:
all_text += "...(truncating remaining pages; PDF is too large)...\n\n"
break
except Exception as page_err:
logging.error(f"Error processing PDF page {i+1}: {str(page_err)}")
all_text += f"### Page {i+1}\n\n(Error extracting content: {str(page_err)})\n\n"
if len(reader.pages) > max_pages:
all_text += f"\nNote: Only the first {max_pages} pages are shown out of {len(reader.pages)} total.\n\n"
result += "### PDF Content\n\n" + all_text
return result
except Exception as e:
logging.error(f"PDF file processing error: {str(e)}")
return f"## PDF File: {file.name}\n\nError occurred: {str(e)}\n\nThis PDF file cannot be processed."
def process_uploaded_files(files):
"""Combine the contents of all uploaded files into one string."""
if not files:
return None
result = "# Uploaded File Contents\n\n"
result += "Below is the content from the files provided by the user. Integrate this data as a main source of information for the blog.\n\n"
for file in files:
try:
ext = file.name.split('.')[-1].lower()
if ext == 'txt':
result += process_text_file(file) + "\n\n---\n\n"
elif ext == 'csv':
result += process_csv_file(file) + "\n\n---\n\n"
elif ext == 'pdf':
result += process_pdf_file(file) + "\n\n---\n\n"
else:
result += f"### Unsupported File: {file.name}\n\n---\n\n"
except Exception as e:
logging.error(f"File processing error {file.name}: {e}")
result += f"### File processing error: {file.name}\n\nError: {e}\n\n---\n\n"
return result
# ──────────────────────────────── Image & Utility ─────────────────────────
def generate_image(prompt, w=768, h=768, g=3.5, steps=30, seed=3):
"""Image generation function."""
if not prompt:
return None, "Insufficient prompt"
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_text: str, topic: str):
"""
Generate a single-line English image prompt from the blog content.
"""
client = get_openai_client()
try:
response = client.chat.completions.create(
model="gpt-4.1-mini", # 일반적으로 사용 가능한 모델로 설정
messages=[
{"role": "system", "content": "Generate a single-line English image prompt from the following text. Return only the prompt text, nothing else."},
{"role": "user", "content": f"Topic: {topic}\n\n---\n{blog_text}\n\n---"}
],
temperature=1,
max_tokens=80,
top_p=1
)
return response.choices[0].message.content.strip()
except Exception as e:
logging.error(f"OpenAI image prompt generation error: {e}")
return f"A professional photo related to {topic}, high quality"
def md_to_html(md: str, title="Ginigen Blog"):
"""Convert Markdown to HTML."""
return f"{title}{markdown.markdown(md)}"
def keywords(text: str, top=5):
"""Simple keyword extraction."""
cleaned = re.sub(r"[^가-힣a-zA-Z0-9\s]", "", text)
return " ".join(cleaned.split()[:top])
# ──────────────────────────────── Streamlit UI ────────────────────────────
def ginigen_app():
st.title("Ginigen Blog")
# Set default session state
if "ai_model" not in st.session_state:
st.session_state.ai_model = "gpt-4.1-mini" # 고정 모델 설정
if "messages" not in st.session_state:
st.session_state.messages = []
if "auto_save" not in st.session_state:
st.session_state.auto_save = True
if "generate_image" not in st.session_state:
st.session_state.generate_image = False
if "web_search_enabled" not in st.session_state:
st.session_state.web_search_enabled = True
if "blog_template" not in st.session_state:
st.session_state.blog_template = "ginigen" # Ginigen recommended style by default
if "blog_tone" not in st.session_state:
st.session_state.blog_tone = "professional"
if "word_count" not in st.session_state:
st.session_state.word_count = 1750
# Sidebar UI
sb = st.sidebar
sb.title("Blog Settings")
# 모델 선택 제거 (고정 모델 사용)
sb.subheader("Blog Style Settings")
sb.selectbox(
"Blog Template",
options=list(BLOG_TEMPLATES.keys()),
format_func=lambda x: BLOG_TEMPLATES[x],
key="blog_template"
)
sb.selectbox(
"Blog Tone",
options=list(BLOG_TONES.keys()),
format_func=lambda x: BLOG_TONES[x],
key="blog_tone"
)
sb.slider("Blog Length (word count)", 800, 3000, key="word_count")
# Example topics
sb.subheader("Example Topics")
c1, c2, c3 = sb.columns(3)
if c1.button("Real Estate Tax", key="ex1"):
process_example(EXAMPLE_TOPICS["example1"])
if c2.button("Summer Festivals", key="ex2"):
process_example(EXAMPLE_TOPICS["example2"])
if c3.button("Investment Guide", key="ex3"):
process_example(EXAMPLE_TOPICS["example3"])
sb.subheader("Other Settings")
sb.toggle("Auto Save", key="auto_save")
sb.toggle("Auto Image Generation", key="generate_image")
web_search_enabled = sb.toggle("Use Web Search", value=st.session_state.web_search_enabled)
st.session_state.web_search_enabled = web_search_enabled
if web_search_enabled:
st.sidebar.info("✅ Web search results will be integrated into the blog.")
# Download the latest blog (markdown/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_match = re.search(r"# (.*?)(\n|$)", latest_blog)
title = title_match.group(1).strip() if title_match else "blog"
sb.subheader("Download Latest Blog")
d1, d2 = sb.columns(2)
d1.download_button("Download as Markdown", latest_blog,
file_name=f"{title}.md", mime="text/markdown")
d2.download_button("Download as HTML", md_to_html(latest_blog, title),
file_name=f"{title}.html", mime="text/html")
# JSON conversation record upload
up = sb.file_uploader("Load Conversation History (.json)", type=["json"], key="json_uploader")
if up:
try:
st.session_state.messages = json.load(up)
sb.success("Conversation history loaded successfully")
except Exception as e:
sb.error(f"Failed to load: {e}")
# JSON conversation record download
if sb.button("Download Conversation as JSON"):
sb.download_button(
"Save",
data=json.dumps(st.session_state.messages, ensure_ascii=False, indent=2),
file_name="chat_history.json",
mime="application/json"
)
# File Upload
st.subheader("File Upload")
uploaded_files = st.file_uploader(
"Upload files to be referenced in your blog (txt, csv, pdf)",
type=["txt", "csv", "pdf"],
accept_multiple_files=True,
key="file_uploader"
)
if uploaded_files:
file_count = len(uploaded_files)
st.success(f"{file_count} files uploaded. They will be referenced in the blog.")
with st.expander("Preview Uploaded Files", expanded=False):
for idx, file in enumerate(uploaded_files):
st.write(f"**File Name:** {file.name}")
ext = file.name.split('.')[-1].lower()
if ext == 'txt':
preview = file.read(1000).decode('utf-8', errors='ignore')
file.seek(0)
st.text_area(
f"Preview of {file.name}",
preview + ("..." if len(preview) >= 1000 else ""),
height=150
)
elif ext == 'csv':
try:
df = pd.read_csv(file)
file.seek(0)
st.write("CSV Preview (up to 5 rows)")
st.dataframe(df.head(5))
except Exception as e:
st.error(f"CSV preview failed: {e}")
elif ext == 'pdf':
try:
file_bytes = file.read()
file.seek(0)
pdf_file = io.BytesIO(file_bytes)
reader = PyPDF2.PdfReader(pdf_file, strict=False)
pc = len(reader.pages)
st.write(f"PDF File: {pc} pages")
if pc > 0:
try:
page_text = reader.pages[0].extract_text()
preview = page_text[:500] if page_text else "(No text extracted)"
st.text_area("Preview of the first page", preview + "...", height=150)
except:
st.warning("Failed to extract text from the first page")
except Exception as e:
st.error(f"PDF preview failed: {e}")
if idx < file_count - 1:
st.divider()
# Display existing messages
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", ""))
# User input
prompt = st.chat_input("Enter a blog topic or keywords.")
if prompt:
process_input(prompt, uploaded_files)
# 사이드바 하단 배지(링크) 추가
sb.markdown("---")
sb.markdown("Created by [https://ginigen.com](https://ginigen.com) | [YouTube Channel](https://www.youtube.com/@ginipickaistudio)")
def process_example(topic):
"""Process the selected example topic."""
process_input(topic, [])
def process_input(prompt: str, uploaded_files):
# Add user's message
if not any(m["role"] == "user" and m["content"] == prompt for m in st.session_state.messages):
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()
message_placeholder = st.empty()
full_response = ""
use_web_search = st.session_state.web_search_enabled
has_uploaded_files = bool(uploaded_files) and len(uploaded_files) > 0
try:
# 상태 표시를 위한 상태 컴포넌트
status = st.status("Preparing to generate blog...")
status.update(label="Initializing client...")
client = get_openai_client()
# Prepare conversation messages
messages = []
# Web search
search_content = None
if use_web_search:
status.update(label="Performing web search...")
with st.spinner("Searching the web..."):
search_content = do_web_search(keywords(prompt, top=5))
# Process uploaded files → content
file_content = None
if has_uploaded_files:
status.update(label="Processing uploaded files...")
with st.spinner("Analyzing files..."):
file_content = process_uploaded_files(uploaded_files)
# Build system prompt
status.update(label="Preparing blog draft...")
sys_prompt = get_system_prompt(
template=st.session_state.blog_template,
tone=st.session_state.blog_tone,
word_count=st.session_state.word_count,
include_search_results=use_web_search,
include_uploaded_files=has_uploaded_files
)
# OpenAI API 호출 준비
status.update(label="Writing blog content...")
# 메시지 구성
api_messages = [
{"role": "system", "content": sys_prompt}
]
user_content = prompt
# 검색 결과가 있으면 사용자 프롬프트에 추가
if search_content:
user_content += "\n\n" + search_content
# 파일 내용이 있으면 사용자 프롬프트에 추가
if file_content:
user_content += "\n\n" + file_content
# 사용자 메시지 추가
api_messages.append({"role": "user", "content": user_content})
# OpenAI API 스트리밍 호출 - 고정 모델 "gpt-4.1-mini" 사용
try:
# 스트리밍 방식으로 API 호출
stream = client.chat.completions.create(
model="gpt-4.1-mini", # 고정 모델 사용
messages=api_messages,
temperature=1,
max_tokens=MAX_TOKENS,
top_p=1,
stream=True # 스트리밍 활성화
)
# 스트리밍 응답 처리
for chunk in stream:
if chunk.choices and len(chunk.choices) > 0 and chunk.choices[0].delta.content is not None:
content_delta = chunk.choices[0].delta.content
full_response += content_delta
message_placeholder.markdown(full_response + "▌")
# 최종 응답 표시 (커서 제거)
message_placeholder.markdown(full_response)
status.update(label="Blog completed!", state="complete")
except Exception as api_error:
error_message = str(api_error)
logging.error(f"API error: {error_message}")
status.update(label=f"Error: {error_message}", state="error")
raise Exception(f"Blog generation error: {error_message}")
# 이미지 생성
answer_entry_saved = False
if st.session_state.generate_image and full_response:
with st.spinner("Generating image..."):
try:
ip = extract_image_prompt(full_response, prompt)
img, cap = generate_image(ip)
if img:
st.image(img, caption=cap)
st.session_state.messages.append({
"role": "assistant",
"content": full_response,
"image": img,
"image_caption": cap
})
answer_entry_saved = True
except Exception as img_error:
logging.error(f"Image generation error: {str(img_error)}")
st.warning("이미지 생성에 실패했습니다. 블로그 콘텐츠만 저장됩니다.")
# Save the answer if not saved above
if not answer_entry_saved and full_response:
st.session_state.messages.append({"role": "assistant", "content": full_response})
# Download buttons
if full_response:
st.subheader("Download This Blog")
c1, c2 = st.columns(2)
c1.download_button(
"Markdown",
data=full_response,
file_name=f"{prompt[:30]}.md",
mime="text/markdown"
)
c2.download_button(
"HTML",
data=md_to_html(full_response, prompt[:30]),
file_name=f"{prompt[:30]}.html",
mime="text/html"
)
# Auto save
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"Auto-save failed: {e}")
except Exception as e:
error_message = str(e)
placeholder.error(f"An error occurred: {error_message}")
logging.error(f"Process input error: {error_message}")
ans = f"An error occurred while processing your request: {error_message}"
st.session_state.messages.append({"role": "assistant", "content": ans})
# ──────────────────────────────── main ────────────────────────────────────
def main():
ginigen_app()
if __name__ == "__main__":
main()