app.py

import os
import gradio as gr
import openai
import requests
import logging
import asyncio
import aiohttp

# 로깅 설정
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# OpenAI API 클라이언트 설정
openai.api_key = os.getenv("OPENAI_API_KEY")

# 요청사항에 따라 카테고리 수정
CATEGORIES = [
    "공포 마케팅",
    "스토리텔링"
]

def get_category_prompt(category):
    if category == "공포 마케팅":
        return """
# 공포 마케팅 카피라이팅 생성 규칙
1. 반드시 한글로 출력하라.
2. 너는 세계 최고의 공포 마케팅 카피라이터이다.
3. 반드시 카피라이팅은 30자 이내로 작성하고, 20개만 출력하라.(다른 내용 출력 금지)
4. 반드시 입력된 주제로만 작성하고 예시를 참고하여 내용을 보강하라.
5. 반드시 1 부터 20까지 번호(리스트형태)를 같이 출력하라
6. 사람들에게 두려움을 주면서, 동시에 제품이나 서비스를 사용해야만 하는 이유를 강하게 어필하라.
7. 사용하지 않을 때의 위험이나 손실을 강조하라.
8. 감정적인 단어와 표현을 사용하여 독자의 불안을 증폭시키라
9. 제품/서비스의 중요성과 그것이 없을 때의 극단적인 결과를 구체적으로 설명하라.
예시:
- 스쿼트 할 때 이 동작 하면 무릎 부상 옵니다
- 책 안 읽는 사람이 가난할 확률 98%인 이유
- 신제품을 출시할 때 실패하는 7 가지 치명적 실수
- 소셜 미디어를 잘못 사용하면 기업이 망할 수 있는 이유
- 이 앱 안 쓰면 해킹 당할 확률 80% 증가
- 이 보험 없으면 사고 시 파산 위기 온다
- 충격! 이 선크림 안 바르면 피부암 위험 500% 증가
- 가습기 2일만 관리하지 않아도 "세균 범벅"
- 바다 환경오염, 식탁으로 되돌아온다. 플라스틱 사용 이대로 괜찮은가요?
- "맥주 한잔이라도 날마다 술이 땡긴다면? 이것 의심해야"
"""
    elif category == "스토리텔링":
        return """
# 스토리텔링 카피라이팅 생성 규칙
1. 반드시 한글로 출력하라.
2. 너는 세계 최고의 스토리텔링 마케팅 카피라이터이다.
3. 반드시 카피라이팅은 30자 이내로 작성하고, 20개만 출력하라.(다른 내용 출력 금지)
4. 반드시 입력된 주제로만 작성하고 예시를 참고하여 내용을 보강하라.
5. 반드시 1 부터 20까지 번호(리스트형태)를 같이 출력하라
6. 제품/서비스와 관련된 짧고 흥미로운 이야기를 만들어라.
7. 고객이 공감할 수 있는 상황이나 캐릭터를 설정하라.
8. 이야기를 통해 제품/서비스의 가치를 자연스럽게 전달하라.
예시:
- 한 어부의 꿈에서 시작된 혁신적인 낚시장비
- 100년 된 가족 레시피로 만든 수제 잼
- 우주 비행사의 아이디어로 탄생한 초경량 재킷
- 할머니의 100년 된 비밀 레시피로 만든 건강한 집밥 간식
- 히말라야 등산객의 고민에서 시작된 혁신적 초경량 배낭
- 몽골 유목민의 천년 지혜를 담은 친환경 오토캠핑 텐트
- 바다를 사랑한 해양생물학자의 혁신적인 해양 정화 기술
- 20년 불면증 환자가 직접 개발한 꿀잠 유도 스마트 베개
- 길고양이 100마리를 구조한 수의사의 반려동물 건강관리 앱
- 팔 없는 장애인 화가의 꿈을 이뤄준 첨단 그림 도구
"""

########################################################
# (기존 동기 처리 함수) - 기존 코드 유지
########################################################
def call_api_sync(content, system_message, max_tokens, temperature, top_p):
    response = requests.post(
        "https://api.openai.com/v1/chat/completions",
        headers={"Authorization": f"Bearer {openai.api_key}"},
        json={
            "model": "gpt-4o-mini",
            "messages": [
                {"role": "system", "content": system_message},
                {"role": "user", "content": content},
            ],
            "max_tokens": max_tokens,
            "temperature": temperature,
            "top_p": top_p,
        }
    )
    result = response.json()
    return result['choices'][0]['message']['content']

def generate_copywriting(categories, topic):
    max_tokens = 1000
    temperature = 0.8
    top_p = 0.95
    
    results = {}
    for category in categories:
        prompt = get_category_prompt(category)
        user_content = f"주제: {topic}"
        copywriting = call_api_sync(user_content, prompt, max_tokens, temperature, top_p)
        results[category] = copywriting
    return results


########################################################
# (새로 추가된 비동기 처리 함수들) - 병렬 구조 및 스트리밍
########################################################
async def call_api_async(category, content, system_message, max_tokens, temperature, top_p):
    """
    비동기로 OpenAI API를 호출하여 결과를 반환
    """
    async with aiohttp.ClientSession() as session:
        headers = {
            "Authorization": f"Bearer {openai.api_key}",
            "Content-Type": "application/json"
        }
        payload = {
            "model": "gpt-4o-mini",
            "messages": [
                {"role": "system", "content": system_message},
                {"role": "user", "content": content},
            ],
            "max_tokens": max_tokens,
            "temperature": temperature,
            "top_p": top_p,
        }
        async with session.post("https://api.openai.com/v1/chat/completions", headers=headers, json=payload) as resp:
            result = await resp.json()
            copywriting = result["choices"][0]["message"]["content"]
    return category, copywriting

async def generate_copywriting_async(categories, topic):
    """
    각 카테고리에 대해 비동기로 API 호출을 수행하고,
    완료되는 순서대로 (category, 결과) 를 yield
    """
    max_tokens = 1000
    temperature = 0.8
    top_p = 0.95

    tasks = []
    for category in categories:
        prompt = get_category_prompt(category)
        user_content = f"주제: {topic}"
        tasks.append(
            asyncio.create_task(
                call_api_async(category, user_content, prompt, max_tokens, temperature, top_p)
            )
        )

    # 완료된 태스크부터 순서대로 결과를 반환
    for task in asyncio.as_completed(tasks):
        cat, copywriting = await task
        yield cat, copywriting


########################################################
# Gradio 인터페이스
########################################################
with gr.Blocks() as iface:
    gr.Markdown("# AI 카피라이팅 생성기")
    
    with gr.Column():
        topic = gr.Textbox(lines=1, label="주제를 입력하세요")

    generate_btn = gr.Button("카피라이팅 생성하기")
    status = gr.Markdown("준비됨")
    
    output_boxes = {}
    with gr.Column():
        for category in CATEGORIES:
            output_box = gr.Textbox(label=category, visible=True)
            output_boxes[category] = output_box

    ########################################################
    # (수정) validate_and_generate 함수를 비동기 함수로 바꿔
    # 3개의 출력에 맞게 매 yield마다 [status, box1, box2]를 반환
    ########################################################
    async def validate_and_generate(topic):
        try:
            results = {}
            # 완료된 순서대로 스트리밍
            async for cat, copywriting in generate_copywriting_async(CATEGORIES, topic):
                results[cat] = copywriting
                # 현재까지 완료된 copywriting을 반영해서 출력
                yield [
                    gr.update(value=f"진행중: [{cat}] 카피라이팅 완료"),
                    *[gr.update(value=results.get(c, "")) for c in CATEGORIES]
                ]

            # 모든 카테고리가 완료된 뒤 최종 출력
            yield [
                gr.update(value="카피라이팅 생성이 완료되었습니다."),
                *[gr.update(value=results.get(c, "")) for c in CATEGORIES]
            ]

        except Exception as e:
            logger.error(f"Error during copywriting generation: {str(e)}")
            yield [
                gr.update(value=f"오류 발생: {str(e)}"),
                *[gr.update(value="") for _ in CATEGORIES]
            ]

    # 비동기 함수를 Gradio 이벤트 핸들러로 등록
    generate_btn.click(
        fn=validate_and_generate,
        inputs=[topic],
        outputs=[status] + [output_boxes[category] for category in CATEGORIES]
    )

    # Gradio 3.30+부터는 queue() + async 함수로 스트리밍 가능
    iface.queue()
    iface.launch()