app.py

import gradio as gr
import requests
import time

API_HOME = 'https://api.fliki.ai/v1'
END_LANGUAGES = '/languages'
END_DIALECTS = '/dialects'
END_VOICES = '/voices'
END_GENERATE = '/generate'
END_STATUS = '/generate/status'
API_KEY = 'ZZUIQ4OZASNRQ8B8WYHNW'
brand = '파인애플 노트북'

def set_header():
    header = {
        "Content-Type": "application/json",
        "Authorization": "Bearer " + API_KEY,
    }
    return header

# 기존의 함수들...


def set_header():
    header = {
        "Content-Type": "application/json",
        "Authorization": "Bearer " + API_KEY,
    }
    return header

# 언어 리스트를 가져오는 함수.
def get_language_id():
    url = API_HOME + END_LANGUAGES
    headers = set_header()
    response = requests.request("GET", url, headers=headers).json()
    print(f"언어 리스트: {response}")
    # Korean: 61b8b2fa4268666c126babf1
    return response


# 언어지역 리스트를 가져오는 함수.
def get_dialects():
    url = API_HOME + END_DIALECTS
    headers = set_header()
    response = requests.request("GET", url, headers=headers).json()
    print(f"언어지역 리스트: {response}")
    # Korea: 61b8b30f4268666c126bac8d
    return response


# 지역언어 리스트를 가져오는 함수.
def get_voices(language_id, dialect_id):
    url = API_HOME + END_VOICES
    headers = set_header()
    payload = {
        'languageId': language_id,
        'dialectId': dialect_id
    }
    response = requests.request("POST", url, json=payload, headers=headers).json()
    print(f"voice: {response}")
    return response


# 비디오 또는 오디오 컨텐츠 생성하는 함수.
def generate_contents(user_content, brand_name):
    global brand
    brand = brand_name  # 브랜드명 설정
    url = API_HOME + END_GENERATE
    VOICE_ID = '65934ac2bc02ab1c006755fa'  # Korean, Korea, Wolf Sea

    payload = {
        "format": "video",  # video | audio
        "scenes": [
            {
                "content": user_content,  # 사용자가 입력한 텍스트
                "voiceId": VOICE_ID
            },
            # ... 다른 씬들이 필요하다면 여기에 추가 ...
        ],
        "settings": {
            'aspectRatio': 'portrait',  # 'portrait', 'square', 'horizontal'
            'subtitle': {
                'fontColor': 'yellow',
                'backgroundColor': 'black',
                'placement': 'bottom',
                'display': 'phrase',
            },
        },
        "backgroundMusicKeywords": "happy, lofi, beats"
    }
    
    headers = set_header()
    response = requests.request("POST", url, json=payload, headers=headers).json()

    # HTTP 상태코드 확인
    if response['success'] == True:
        # 응답값 예시: {'success': True, 'data': {'id': '6593aa11d153762d8aa022e9'}}
        print(f"정상: {response['data']['id']}")
        return response['data']['id']
    else:
        print(f"request 실패: {response}")
        return 'FAILED'


# /generate API 호출 성공 후 받은 id로 컨텐츠 생성 상태를 확인하는 함수.
def generate_status(content_id):
    url = API_HOME + END_STATUS
    payload = {"id": content_id}
    headers = set_header()
    response = requests.request("POST", url, json=payload, headers=headers).json()

    # HTTP 상태코드 확인
    if response['success'] == True:
        print(f"정상: {response}")
        if response['data']['status'] == 'queued':
            print('아직 컨텐츠 URL이 생성되지 않았습니다.')
            return 'PENDING'
        if response['data']['status'] == 'processing':
            print('컨텐츠 URL이 아직 생성 중 입니다.')
            return 'PENDING' 
        elif response['data']['status'] == 'success':
            return response['data']['file']

    else:
        print(f"request 실패")


def download_content(url):
    data = requests.get(url, timeout=30)
    save_path = "downloaded_example.mp4"
    with open(save_path, 'wb') as file:
        for chunk in data.iter_content(chunk_size=128):
            file.write(chunk)
        print(f"MP4 파일이 성공적으로 저장되었습니다. 경로: {save_path}")
    return 'success'


# # 실행.

# get_language_id() # Korean: 61b8b2fa4268666c126babf1
# get_dialects() # Korea: 61b8b30f4268666c126bac8d
# get_voices('61b8b2fa4268666c126babf1', '61b8b30f4268666c126bac8d')

try:
    id = generate_contents()
    if id != 'FAILED':
        status = 'PENDING'
        while status == 'PENDING':
            time.sleep(9)  # 9초 기다림
            print('9초 대기 후 상태확인 함수를 호출합니다.')
            status = generate_status(id)
            if status != 'PENDING': break

        print(f"컨텐츠 생성 성공: {status}")
    else:
        print('생성 실패..')
except Exception as e:
    print(f"오류 발생: {e}")

    
# Gradio 인터페이스를 위한 함수
def gradio_generate_contents(user_content, brand_name):
    return generate_contents(user_content, brand_name)  # 두 입력 모두 전달

# Gradio 인터페이스 수정
iface = gr.Interface(
    fn=gradio_generate_contents,  # 컨텐츠 생성 함수
    inputs=[gr.inputs.Textbox(label="컨텐츠 텍스트"), gr.inputs.Textbox(label="브랜드명")],    
    outputs=gr.outputs.Textbox(label="컨텐츠 ID")
)


status_iface = gr.Interface(
    fn=gradio_generate_status,  # 컨텐츠 상태 확인 함수
    inputs=gr.inputs.Textbox(label="컨텐츠 ID"),
    outputs=gr.outputs.Textbox(label="컨텐츠 상태 또는 URL")
)

iface.launch()
status_iface.launch()