Update app.py

app.py

@@ -1,232 +1,43 @@
 import gradio as gr
+from PIL import Image
+import numpy as np
+import cv2
 import os
-from huggingface_hub import InferenceClient
 
-import pandas as pd
-import openpyxl
-from openpyxl.utils.dataframe import dataframe_to_rows
-from datetime import datetime
-from io import BytesIO
-
-from langchain.text_splitter import RecursiveCharacterTextSplitter
-from langchain_community.vectorstores import Chroma
-from langchain_community.embeddings import HuggingFaceEmbeddings
-from langchain.memory import ConversationBufferMemory
-from langchain.chains import create_retrieval_chain, create_history_aware_retriever
-
-from pathlib import Path
-import chromadb
-from unidecode import unidecode
-
-import re
-from langchain.schema import Document
-from langchain.prompts import PromptTemplate
-
-# API 클라이언트 초기화
-llm_client = InferenceClient("CohereForAI/c4ai-command-r-plus", token=os.getenv("HF_TOKEN"))
-
-# 긴 글 텍스트 결과를 위한 함수 정의
-def process_file(file):
-    df = read_excel_data(file)
-    if df is None:
-        raise ValueError("Excel 데이터를 읽는 중 오류가 발생했습니다.")
-    longest_reviews = extract_longest_reviews(df)
-    if longest_reviews is None:
-        raise ValueError("가장 긴 리뷰를 추출하는 중 오류가 발생했습니다.")
-    result_file = save_to_excel(longest_reviews)
-    if result_file is None:
-        raise ValueError("엑셀 파일로 저장하는 중 오류가 발생했습니다.")
-    return result_file
-
-def analyze_and_initialize_db(file, chunk_size, chunk_overlap, progress=gr.Progress()):
-    try:
-        print("파일 처리 시작")
-        result_file = process_file(file)
-        list_file_path = [result_file]
-        print("데이터베이스 초기화 시작")
-        vector_db, collection_name, db_status = initialize_database(list_file_path, chunk_size, chunk_overlap, progress)
-        print("데이터베이스 초기화 완료")
-        return vector_db, collection_name, db_status, list_file_path, result_file
-    except Exception as e:
-        print(f"Database initialization error: {e}")
-        return None, None, "Failed", None, None
-
-def long_text_result(file):
-    try:
-        print("긴 텍스트 결과 분석 시작")
-        progress = gr.Progress()
-        vector_db, collection_name, db_status, list_file_path, result_file = analyze_and_initialize_db(file, 600, 40, progress)
-        print(f"DB 상태: {db_status}")
-        if db_status == "Complete!":
-            analysis = "분석완료"
-            # Vector DB와 관련 chain 설정
-            global chat_chain
-            memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True)
-            prompt_template = PromptTemplate(template="Based on the following document, please provide a relevant response:", input_variables=["context"])
-            retriever = create_history_aware_retriever(
-                retriever=vector_db.as_retriever(),
-                llm=llm_client,
-                prompt=prompt_template
-            )
-            chat_chain = create_retrieval_chain(
-                retriever=retriever,
-                llm=llm_client,
-                memory=memory
-            )
-        else:
-            analysis = "분석 실패"
-        return analysis
-    except Exception as e:
-        print(f"Long text result error: {e}")
-        return "분석 실패"
-
-def chatbot_response(input_text):
-    try:
-        response = chat_chain.run(input_text)
-        return response
-    except Exception as e:
-        print(f"Chatbot response error: {e}")
-        return "챗봇 응답 생성 실패"
-
-# 엑셀 데이터 읽기
-def read_excel_data(file):
-    try:
-        print("엑셀 데이터 읽기 시작")
-        df = pd.read_excel(BytesIO(file), usecols="B, C, D, E", skiprows=1, names=["Review Date", "Option", "Review", "ReviewScore"])
-        df['Review Date'] = pd.to_datetime(df['Review Date']).dt.tz_localize(None).dt.date
-        df['Year-Month'] = df['Review Date'].astype(str).str.slice(0, 7)
-        df['Year'] = df['Review Date'].astype(str).str.slice(0, 4)
-        df['Month'] = df['Review Date'].astype(str).str.slice(5, 7)
-        df['Day'] = df['Review Date'].astype(str).str.slice(8, 10)
-        df['Option1'] = df['Option'].str.split(" / ").str[0]  # Extract primary option
-        df['Review Length'] = df['Review'].str.len()  # Calculate review length
-        return df
-    except Exception as e:
-        print(f"Error reading Excel data: {e}")
-        return None
-
-def extract_longest_reviews(df):
-    try:
-        print("긴 리뷰 추출 시작")
-        longest_reviews = df.groupby('ReviewScore', group_keys=False).apply(lambda x: x.nlargest(100, 'Review Length', keep='all')).reset_index(drop=True)
-        return longest_reviews.drop(columns=['Review Length', 'Year-Month', 'Year', 'Month', 'Day', 'Option1'])  # Drop unnecessary columns
-    except Exception as e:
-        print(f"Error extracting longest reviews: {e}")
-        return None
-
-def save_to_excel(longest_reviews):
-    try:
-        print("엑셀로 저장 시작")
-        wb = openpyxl.Workbook()
-        ws = wb.active
-        ws.title = "긴 리뷰 내용"
-
-        for r in dataframe_to_rows(longest_reviews, index=False, header=True):
-            ws.append(r)
-        ws.sheet_properties.tabColor = "00FF00"  # Green color
-
-        file_path = "리뷰분석_긴리뷰_다운로드.xlsx"
-        wb.save(file_path)
-        return file_path
-    except Exception as e:
-        print(f"Error saving to Excel: {e}")
-        return None
-
-def create_collection_name(filepath):
-    try:
-        collection_name = Path(filepath).stem
-        collection_name = collection_name.replace(" ","-") 
-        collection_name = unidecode(collection_name)
-        collection_name = re.sub('[^A-Za-z0-9]+', '-', collection_name)
-        collection_name = collection_name[:50]
-        if len(collection_name) < 3:
-            collection_name = collection_name + 'xyz'
-        if not collection_name[0].isalnum():
-            collection_name = 'A' + collection_name[1:]
-        if not collection_name[-1].isalnum():
-            collection_name = collection_name[:-1] + 'Z'
-        return collection_name
-    except Exception as e:
-        print(f"Error creating collection name: {e}")
-        return "default-collection"
-
-def load_doc(list_file_path, chunk_size, chunk_overlap):
-    try:
-        print("문서 로드 시작")
-        pages = []
-        for file_path in list_file_path:
-            if file_path.endswith('.xlsx'):
-                df = pd.read_excel(file_path)
-                for _, row in df.iterrows():
-                    pages.append(Document(page_content=row.to_string()))
+def image_to_video(image):
+    # 이미지를 NumPy 배열로 변환
+    image_array = np.array(image)
+    
+    # 비디오 저장 경로와 이름 설정
+    output_path = '/mnt/data/output_video.mp4'
+    
+    # 비디오 라이터 설정
+    height, width, layers = image_array.shape
+    size = (width, height)
+    video = cv2.VideoWriter(output_path, cv2.VideoWriter_fourcc(*'mp4v'), 1, size)
+    
+    # 이미지를 사용하여 5초 동안의 비디오 생성 (5 프레임, 각각 1초 지속)
+    for _ in range(5):  # 5 프레임 생성
+        video.write(image_array)
+    
+    # 비디오 작성 종료
+    video.release()
+    
+    return output_path
+
+def setup_interface():
+    # Gradio 인터페이스 구성
+    with gr.Blocks() as demo:
+        gr.Markdown("### 이미지를 업로드하면 5초짜리 비디오를 생성합니다.")
         
-        text_splitter = RecursiveCharacterTextSplitter(chunk_size=chunk_size, chunk_overlap=chunk_overlap)
-        doc_splits = text_splitter.split_documents(pages)
-        return doc_splits
-    except Exception as e:
-        print(f"Error loading documents: {e}")
-        return []
-
-def create_db(splits, collection_name):
-    try:
-        print("DB 생성 시작")
-        embedding = HuggingFaceEmbeddings()
-        new_client = chromadb.EphemeralClient()
-        vectordb = Chroma.from_documents(
-            documents=splits,
-            embedding=embedding,
-            client=new_client,
-            collection_name=collection_name,
-        )
-        return vectordb
-    except Exception as e:
-        print(f"Error creating database: {e}")
-        return None
-
-def initialize_database(list_file_path, chunk_size, chunk_overlap, progress=gr.Progress()):
-    try:
-        print("데이터베이스 초기화 시작")
-        collection_name = create_collection_name(list_file_path[0])
-        doc_splits = load_doc(list_file_path, chunk_size, chunk_overlap)
-        vector_db = create_db(doc_splits, collection_name)
-        if vector_db:
-            print("데이터베이스 초기화 성공")
-            return vector_db, collection_name, "Complete!"
-        else:
-            print("데이터베이스 생성 실패")
-            return None, collection_name, "Failed"
-    except Exception as e:
-        print(f"Error initializing database: {e}")
-        return None, None, "Failed"
-
-# 새로운 탭에 왼쪽 긴 텍스트 결과와 오른쪽 챗봇 인터페이스 구성
-iface = gr.Blocks()
-
-with iface:
-    with gr.Row():
-        with gr.Column():
-            uploaded_file = gr.File(file_count="single", type="binary", label="엑셀 파일 업로드")
-            analysis_status = gr.Textbox(label="분석 상태[데이터에 따라 최대 3분이상 시간이 걸릴수 있습니다.]", value="", lines=1, interactive=False)
-            long_text_output = gr.Textbox(label="상품의 장단점 10가지를 분석해드립니다.", lines=27, interactive=False)
-            uploaded_file.upload(long_text_result, inputs=uploaded_file, outputs=analysis_status)
-        with gr.Column():
-            chatbot_input = gr.Textbox(label="챗봇 입력", placeholder="이 상품에 대한 추가적인 자세한 분석내용을 챗봇에게 질문하세요.")
-            chatbot_examples = gr.Dropdown(
-                ["기능적인 내용 중 만족/불만족 항목을 20개씩 분석해주세요",
-                 "디자인적인 내용 중 만족/불만족 항목을 20개씩 분석해주세요.",
-                 "감성적인 내용 중 만족/불만족 항목을 20개씩 분석해주세요.",
-                 "추가로 20개 더 해주세요."],
-                label="챗봇 예시항목 선택"
-            )
-            chatbot_output = gr.Textbox(label="챗봇 응답", lines=20)  # 응답 칸을 길게 설정
-            
-            with gr.Row():
-                chatbot_button = gr.Button("챗봇에게 질문하기")                
-                clear_button = gr.Button("모두 지우기")
-                
-            chatbot_button.click(chatbot_response, inputs=chatbot_input, outputs=chatbot_output)
-            clear_button.click(fn=lambda: "", inputs=None, outputs=chatbot_output)  # 모두 지우기 버튼 클릭 시 응답 내용 초기화
-            chatbot_examples.change(fn=lambda x: x, inputs=chatbot_examples, outputs=chatbot_input)
+        with gr.Row():
+            image_input = gr.Image(type="pil")
+            video_output = gr.Video(label="생성된 비디오")
+        
+        image_input.change(image_to_video, inputs=image_input, outputs=video_output)
+    
+    return demo
 
-if __name__ == "__main__":
-    iface.launch()
+# 인터페이스 실행
+demo = setup_interface()
+demo.launch()