Update app.py

app.py

@@ -1,723 +1,3 @@
-import gradio as gr
-import hashlib
-import re
-from datetime import datetime
 import os
-import json
-import schedule
-import threading
-import time
-import dns.resolver
-from huggingface_hub import Repository
-import spaces
-import argparse
-from os import path
-import shutil
-from safetensors.torch import load_file
-from huggingface_hub import hf_hub_download
-import torch
-from diffusers import FluxPipeline
-from diffusers.pipelines.stable_diffusion import safety_checker
-from PIL import Image
-from transformers import pipeline
-import replicate
-import logging
-import requests
-from pathlib import Path
-import cv2
-import numpy as np
-import sys
-import io
-
-# 로깅 설정
-logging.basicConfig(level=logging.INFO)
-logger = logging.getLogger(__name__)
-
-# Hugging Face Dataset Repo 설정
-HF_TOKEN = os.getenv("HF_TOKEN")
-REPO_ID = "ginigen/MEMBERSHIP"
-LOCAL_DIR = "./my_dataset"
-
-repo = Repository(
-    local_dir=LOCAL_DIR,
-    clone_from=REPO_ID,
-    use_auth_token=HF_TOKEN,
-    repo_type="dataset"
-)
-
-DATA_FILE = os.path.join(LOCAL_DIR, "data.json")
-
-current_user = {"email": None, "points": 0}
-ADMIN_EMAIL = "[email protected]"
-ADMIN_PASS = "Arxiv4837!@"
-
-# Setup and initialization code
-cache_path = path.join(path.dirname(path.abspath(__file__)), "models")
-PERSISTENT_DIR = os.environ.get("PERSISTENT_DIR", ".")
-
-# API 설정
-CATBOX_USER_HASH = "e7a96fc68dd4c7d2954040cd5"
-REPLICATE_API_TOKEN = os.getenv("API_KEY")
-
-# 환경 변수 설정
-os.environ["TRANSFORMERS_CACHE"] = cache_path
-os.environ["HF_HUB_CACHE"] = cache_path
-os.environ["HF_HOME"] = cache_path
-
-# 번역기 초기화
-translator = pipeline("translation", model="Helsinki-NLP/opus-mt-ko-en", device="cpu")
-
-if not path.exists(cache_path):
-    os.makedirs(cache_path, exist_ok=True)
-
-
-
-def load_data():
-    try:
-        if os.path.exists(DATA_FILE):
-            with open(DATA_FILE, 'r', encoding='utf-8') as f:
-                return json.load(f)
-    except Exception as e:
-        print(f"Load error: {e}")
-    return {"users": {}}
-
-def save_data(data):
-    try:
-        with open(DATA_FILE, 'w', encoding='utf-8') as f:
-            json.dump(data, f, ensure_ascii=False, indent=4)
-        repo.git_add()
-        repo.git_commit("Update data.json")
-        repo.git_push()
-        return True
-    except Exception as e:
-        print(f"Save error: {e}")
-        return False
-
-def init_db():
-    print("Initializing database...")
-    try:
-        data = load_data()
-        if ADMIN_EMAIL not in data["users"]:
-            data["users"][ADMIN_EMAIL] = {
-                "password": hash_password(ADMIN_PASS),
-                "registration_date": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
-                "points": 999,
-                "is_admin": 1
-            }
-            save_data(data)
-            print("Admin account created")
-        print("Database initialized successfully")
-        return True
-    except Exception as e:
-        print(f"Init error: {e}")
-        return False
-
-def hash_password(password: str) -> str:
-    if not password:
-        return ""
-    return hashlib.sha256(password.encode("utf-8")).hexdigest()
-
-def is_valid_email(email: str) -> bool:
-    pattern = r'^[\w\.-]+@[\w\.-]+\.\w+$'
-    return re.match(pattern, email) is not None
-
-def is_email_domain_valid(email: str) -> bool:
-    try:
-        domain = email.split("@")[1]
-    except IndexError:
-        return False
-    try:
-        records = dns.resolver.resolve(domain, 'MX')
-        if len(records) > 0:
-            return True
-    except (dns.resolver.NXDOMAIN, dns.resolver.NoAnswer, dns.exception.Timeout):
-        pass
-    return False
-
-def register(email, password):
-    if not email or not password:
-        return "Please fill all fields."
-    if not is_valid_email(email):
-        return "Invalid email format. Please try again."
-    if not is_email_domain_valid(email):
-        return "This email domain seems invalid. Please use a different address."
-    if len(password) < 6:
-        return "Password must be at least 6 characters long."
-
-    data = load_data()
-    if email.strip() in data["users"]:
-        return "This email is already registered."
-
-    data["users"][email.strip()] = {
-        "password": hash_password(password.strip()),
-        "registration_date": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
-        "points": 15,  # 초기 포인트 15로 설정
-        "is_admin": 0
-    }
-    if save_data(data):
-        return "Registration successful!"
-    return "Error occurred during registration."
-
-def login(email, password):
-    global current_user
-    if not email or not password:
-        return {"value": "Please enter both email and password."}
-
-    data = load_data()
-    user_data = data["users"].get(email.strip())
-    if user_data and user_data["password"] == hash_password(password.strip()):
-        current_user = {
-            "email": email.strip(),
-            "points": user_data["points"],
-            "is_admin": user_data.get("is_admin", 0)
-        }
-        if current_user["is_admin"] == 1:
-            return {"value": "Logged in as ADMIN."}
-        else:
-            return {"value": f"Welcome! You have {user_data['points']} points."}
-    current_user = {"email": None, "points": 0, "is_admin": 0}
-    return {"value": "Wrong email or password."}
-
-def use_point(points=5):
-    if not current_user["email"]:
-        return "You need to log in first."
-    data = load_data()
-    user_data = data["users"][current_user["email"]]
-    if user_data["points"] < points:
-        return f"Not enough points. Required: {points} points"
-    user_data["points"] -= points
-    current_user["points"] = user_data["points"]
-    if save_data(data):
-        return f"Points used! Remaining points: {user_data['points']}"
-    return "Error occurred while using points."
-
-def get_profile():
-    if not current_user["email"]:
-        return "", "", "You need to log in first."
-    data = load_data()
-    user_data = data["users"].get(current_user["email"])
-    if user_data:
-        return (
-            current_user["email"],
-            user_data["registration_date"],
-            f"Current points: {user_data['points']}"
-        )
-    return "", "", "Profile not found."
-
-def delete_user(email):
-    if not is_admin():
-        return "Only admin can delete users."
-    if email == ADMIN_EMAIL:
-        return "Cannot delete the admin account."
-    data = load_data()
-    if email in data["users"]:
-        del data["users"][email]
-        if save_data(data):
-            return f"User {email} has been deleted."
-    return "User not found."
-
-def is_admin():
-    return current_user.get("email") == ADMIN_EMAIL and current_user.get("is_admin") == 1
-
-def get_all_users():
-    if not is_admin():
-        return "Only admin can access this page."
-    data = load_data()
-    users = data["users"]
-    if not users or len(users) <= 1:
-        return "No registered users."
-
-    result = "<table style='width:100%; border-collapse: collapse;'>"
-    result += "<tr style='background-color: #f2f2f2;'>"
-    result += "<th style='padding: 12px; text-align: left; border: 1px solid #ddd;'>Email</th>"
-    result += "<th style='padding: 12px; text-align: left; border: 1px solid #ddd;'>Registration Date</th>"
-    result += "<th style='padding: 12px; text-align: left; border: 1px solid #ddd;'>Points</th>"
-    result += "<th style='padding: 12px; text-align: center; border: 1px solid #ddd;'>Action</th></tr>"
-
-    for email, user_data in users.items():
-        if email != ADMIN_EMAIL:
-            result += f"<tr style='border: 1px solid #ddd;'>"
-            result += f"<td style='padding: 12px; border: 1px solid #ddd;'>{email}</td>"
-            result += f"<td style='padding: 12px; border: 1px solid #ddd;'>{user_data['registration_date']}</td>"
-            result += f"<td style='padding: 12px; border: 1px solid #ddd;'>{user_data['points']}</td>"
-            result += f"<td style='padding: 12px; text-align: center; border: 1px solid #ddd;'>"
-            result += f"<button onclick='deleteUser(\"{email}\")' style='padding: 5px 10px; background-color: #ff4444; color: white; border: none; border-radius: 3px; cursor: pointer;'>Delete</button></td>"
-            result += "</tr>"
-    result += "</table>"
-    return result
-
-@spaces.GPU
-def setup_torch():
-    torch.backends.cuda.matmul.allow_tf32 = True
-
-def translate_if_korean(text):
-    if any(ord(char) >= 0xAC00 and ord(char) <= 0xD7A3 for char in text):
-        translation = translator(text)[0]['translation_text']
-        return translation
-    return text
-
-def filter_prompt(prompt):
-    inappropriate_keywords = [
-        "nude", "naked", "nsfw", "porn", "sex", "explicit", "adult", "xxx",
-        "erotic", "sensual", "seductive", "provocative", "intimate",
-        "violence", "gore", "blood", "death", "kill", "murder", "torture",
-        "drug", "suicide", "abuse", "hate", "discrimination"
-    ]
-    
-    prompt_lower = prompt.lower()
-    for keyword in inappropriate_keywords:
-        if keyword in prompt_lower:
-            return False, "부적절한 내용이 포함된 프롬프트입니다."
-    return True, prompt
-
-def process_prompt(prompt):
-    translated_prompt = translate_if_korean(prompt)
-    is_safe, filtered_prompt = filter_prompt(translated_prompt)
-    return is_safe, filtered_prompt
-
-def add_watermark(video_path):
-    try:
-        cap = cv2.VideoCapture(video_path)
-        width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
-        height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
-        fps = int(cap.get(cv2.CAP_PROP_FPS))
-        
-        text = "GiniGEN.AI"
-        font = cv2.FONT_HERSHEY_SIMPLEX
-        font_scale = height * 0.05 / 30
-        thickness = 2
-        color = (255, 255, 255)
-        
-        (text_width, text_height), _ = cv2.getTextSize(text, font, font_scale, thickness)
-        margin = int(height * 0.02)
-        x_pos = width - text_width - margin
-        y_pos = height - margin
-        
-        output_path = "watermarked_output.mp4"
-        fourcc = cv2.VideoWriter_fourcc(*'mp4v')
-        out = cv2.VideoWriter(output_path, fourcc, fps, (width, height))
-        
-        while cap.isOpened():
-            ret, frame = cap.read()
-            if not ret:
-                break
-            cv2.putText(frame, text, (x_pos, y_pos), font, font_scale, color, thickness)
-            out.write(frame)
-        
-        cap.release()
-        out.release()
-        
-        return output_path
-        
-    except Exception as e:
-        logger.error(f"Error adding watermark: {str(e)}")
-        return video_path
-
-def check_api_key():
-    api_key = os.getenv("API_KEY")
-    if not api_key:
-        logger.error("API_KEY environment variable not found")
-        return False
-        
-    os.environ["REPLICATE_API_TOKEN"] = api_key
-    
-    try:
-        response = requests.get(
-            "https://api.replicate.com/v1/account",
-            headers={"Authorization": f"Bearer {api_key}"}
-        )
-        if response.status_code == 200:
-            logger.info("Replicate API token validated successfully")
-            return True
-        else:
-            logger.error(f"API key validation failed with status code: {response.status_code}")
-            return False
-    except Exception as e:
-        logger.error(f"API key validation error: {str(e)}")
-        return False
-
-def generate_video(image, prompt):
-    logger.info("Starting video generation")
-    try:
-        if not check_api_key():
-            return "Replicate API key not properly configured"
-
-        input_data = {
-            "prompt": prompt
-        }
-
-        if image:
-            try:
-                import base64
-                with open(image, 'rb') as img_file:
-                    data = base64.b64encode(img_file.read()).decode('utf-8')
-                    input_data["first_frame_image"] = f"data:image/png;base64,{data}"
-            except Exception as img_error:
-                logger.error(f"Error processing image: {str(img_error)}")
-                return f"Error processing image: {str(img_error)}"
-
-        try:
-            prediction = replicate.predictions.create(
-                model="minimax/video-01-live",
-                input=input_data
-            )
-            
-            while prediction.status not in ["succeeded", "failed", "canceled"]:
-                prediction = replicate.predictions.get(prediction.id)
-                time.sleep(1)
-            
-            if prediction.status == "succeeded" and prediction.output:
-                temp_file = "temp_output.mp4"
-                try:
-                    response = requests.get(prediction.output, stream=True)
-                    response.raise_for_status()
-                    
-                    with open(temp_file, "wb") as f:
-                        for chunk in response.iter_content(chunk_size=8192):
-                            if chunk:
-                                f.write(chunk)
-                    
-                    final_video = add_watermark(temp_file)
-                    return final_video
-                    
-                except Exception as download_error:
-                    logger.error(f"Error downloading video: {str(download_error)}")
-                    return f"Error downloading video: {str(download_error)}"
-            else:
-                error_msg = f"Prediction failed with status: {prediction.status}"
-                if hasattr(prediction, 'error'):
-                    error_msg += f" Error: {prediction.error}"
-                logger.error(error_msg)
-                return error_msg
-
-        except Exception as api_error:
-            logger.error(f"API call failed: {str(api_error)}")
-            return f"API call failed: {str(api_error)}"
-
-    except Exception as e:
-        logger.error(f"Unexpected error: {str(e)}")
-        return f"Unexpected error: {str(e)}"
-    finally:
-        try:
-            if 'temp_file' in locals() and os.path.exists(temp_file):
-                os.remove(temp_file)
-        except Exception as cleanup_error:
-            logger.warning(f"Error cleaning up temporary file: {str(cleanup_error)}")
-
-def process_and_generate_video(image, prompt):
-    result = use_point(5)  # 5포인트 차감
-    if "Not enough points" in result or "need to log in" in result:
-        return result
-    
-    is_safe, translated_prompt = process_prompt(prompt)
-    if not is_safe:
-        return "부적절한 내용이 포함된 프롬프트입니다."
-    
-    try:
-        video_result = generate_video(image, translated_prompt)
-        if isinstance(video_result, str) and ("error" in video_result.lower() or "failed" in video_result.lower()):
-            # 에러 발생 시 포인트 환불
-            data = load_data()
-            user_data = data["users"][current_user["email"]]
-            user_data["points"] += 5
-            current_user["points"] = user_data["points"]
-            save_data(data)
-            return f"Error: {video_result}"
-        return video_result
-    except Exception as e:
-        # 에러 발생 시 포인트 환불
-        data = load_data()
-        user_data = data["users"][current_user["email"]]
-        user_data["points"] += 5
-        current_user["points"] = user_data["points"]
-        save_data(data)
-        return f"Error: {str(e)}"
-
-
-# CSS 스타일 정의
-CUSTOM_CSS = """
-/* 한 줄에 Email + Password + (Login, Sign Up) 버튼 2개 + 결과창 */
-
-/* 전체 Row 컨테이너 */
-#row-container {
-    display: flex;
-    align-items: center;
-    gap: 8px;
-    background: linear-gradient(135deg, #b2fefa 0%, #8fd3f4 100%);
-    padding: 10px;
-    border-radius: 8px;
-    box-shadow: 0 3px 8px rgba(0,0,0,0.15);
-    margin-bottom: 15px;
-}
-
-/* 텍스트박스, 버튼 크기 통일 */
-.same-size {
-    width: 130px !important;
-    height: 34px !important;
-    font-size: 0.9rem !important;
-    text-align: center !important;
-    padding: 5px 6px !important;
-    margin: 0 !important;
-    border-radius: 5px;
-    border: 1px solid #ddd;
-}
-
-/* 버튼은 약간 다른 배경 */
-button.same-size {
-    background-color: #6666ff;
-    color: white;
-    border: none;
-    cursor: pointer;
-}
-button.same-size:hover {
-    background-color: #4a4acc;
-}
-
-/* 결과창도 동일 크기 */
-.result-box {
-    width: 160px !important;
-    height: 34px !important;
-    font-size: 0.9rem !important;
-    padding: 5px 6px !important;
-    border: 1px solid #ccc;
-    border-radius: 5px;
-    background-color: #fff;
-    text-align: left;
-    overflow: hidden;
-    white-space: nowrap;
-    text-overflow: ellipsis;
-}
-
-/* 상단 헤더 스타일 */
-.header-container {
-    display: flex;
-    justify-content: space-between;
-    align-items: center;
-    padding: 6px;
-    background-color: #f5f5f5;
-    margin-bottom: 10px;
-}
-.user-badge {
-    padding: 3px 6px;
-    border-radius: 15px;
-    background-color: #4CAF50;
-    color: white;
-    font-size: 0.8em;
-}
-.admin-badge {
-    padding: 3px 6px;
-    border-radius: 15px;
-    background-color: #ff4444;
-    color: white;
-    font-size: 0.8em;
-    cursor: pointer;
-}
-
-.points-display {
-    background: linear-gradient(135deg, #6e8efb 0%, #5d7df9 100%);
-    padding: 20px;
-    border-radius: 15px;
-    margin: 20px 0;
-    box-shadow: 0 4px 15px rgba(110, 142, 251, 0.2);
-}
-
-.points-text {
-    color: white;
-    font-size: 1.5em;
-    text-align: center;
-    margin: 0;
-    text-shadow: 0 2px 4px rgba(0, 0, 0, 0.1);
-}
-
-footer {
-    visibility: hidden;
-}
-
-.gradio-container {
-    background: linear-gradient(135deg, #f6f8ff 0%, #e9f0ff 100%);
-}
-
-.gr-button {
-    border: 2px solid rgba(100, 100, 255, 0.2);
-    background: linear-gradient(135deg, #6e8efb 0%, #5d7df9 100%);
-    box-shadow: 0 4px 15px rgba(110, 142, 251, 0.2);
-}
-
-.gr-button:hover {
-    background: linear-gradient(135deg, #5d7df9 0%, #4a6af8 100%);
-    box-shadow: 0 4px 20px rgba(110, 142, 251, 0.3);
-}
-
-.gr-input, .gr-box {
-    border-radius: 12px;
-    box-shadow: 0 4px 15px rgba(0, 0, 0, 0.05);
-    border: 2px solid rgba(100, 100, 255, 0.1);
-}
-"""
-
-# Gradio 인터페이스 생성
-with gr.Blocks(theme="soft", css=CUSTOM_CSS) as demo:
-    gr.HTML("""<a href="https://visitorbadge.io/status?path=https%3A%2F%2Fginigen-Dokdo-membership.hf.space">
-               <img src="https://api.visitorbadge.io/api/visitors?path=https%3A%2F%2Fginigen-Dokdo-membership.hf.space&countColor=%23263759" />
-               </a>""")
-    
-    gr.Markdown("## 'Dokdo membership' Image to Video generation")            
-
-
-    # 상단 헤더
-    with gr.Row(elem_classes="header-container"):
-        gr.Markdown("##### Upon Dokdo membership registration, get 15 points. Video creation costs 5 points. Need more points through public contributions or paid services? Contact [email protected].")
-
-        with gr.Column(scale=1):
-            user_info = gr.HTML(value="", elem_classes="user-badge")
-            admin_button = gr.Button("Admin Page", visible=False, elem_classes="admin-badge")
-
-    # 로그인/회원가입 섹션
-    with gr.Row(elem_id="row-container"):
-        email_box = gr.Textbox(
-            placeholder="Email",
-            show_label=False,
-            elem_classes=["same-size"]
-        )
-        pass_box = gr.Textbox(
-            placeholder="Password",
-            show_label=False,
-            type="password",
-            elem_classes=["same-size"]
-        )
-        login_btn = gr.Button(
-            "Login",
-            elem_classes=["same-size"]
-        )
-        signup_btn = gr.Button(
-            "Sign Up",
-            elem_classes=["same-size"]
-        )
-        auth_output = gr.Textbox(
-            show_label=False,
-            elem_classes=["result-box"],
-            interactive=False,
-            placeholder="Message"
-        )
-
-    # 로그인 후 화면
-    with gr.Column(elem_id="studio-section", visible=False) as studio_container:
-        gr.Markdown("### My Studio")
-        with gr.Group(elem_classes="points-display"):
-            points_display = gr.Markdown("", elem_classes="points-text")
-
-        # 비디오 생성 섹션
-        with gr.Row():
-            with gr.Column(scale=3):
-                video_prompt = gr.Textbox(
-                    label="Video Description",
-                    placeholder="비디오 설명을 입력하세요... (한글 입력 가능)",
-                    lines=3
-                )
-                upload_image = gr.Image(type="filepath", label="Upload First Frame Image")
-                video_generate_btn = gr.Button("🎬 Generate Video (5 points)")
-                
-            with gr.Column(scale=4):
-                video_output = gr.Video(label="Generated Video")
-
-        refresh_profile_button = gr.Button("Refresh Points")
-
-    # 관리자 페이지
-    with gr.Column(visible=False) as admin_container:
-        gr.Markdown("### Admin Page")
-        admin_refresh = gr.Button("Get All Users")
-        admin_output = gr.HTML()
-        delete_status = gr.Textbox(label="Result")
-
-    # 이벤트 핸들러
-    def login_and_update(email, password):
-        res = login(email, password)
-        msg = res["value"]
-        if "Welcome!" in msg or "ADMIN" in msg:
-            user_badge = f"<div>Logged in: {email}</div>"
-            is_admin_flag = "ADMIN" in msg
-            _, _, points_info = get_profile()
-            return {
-                user_info: user_badge,
-                studio_container: gr.Column(visible=True),
-                admin_button: gr.Button(visible=is_admin_flag),
-                points_display: f"### {points_info}",
-                auth_output: msg
-            }
-        else:
-            return {auth_output: msg}
-
-    # 이벤트 연결
-    login_btn.click(
-        fn=login_and_update,
-        inputs=[email_box, pass_box],
-        outputs=[
-            user_info,
-            studio_container,
-            admin_button,
-            points_display,
-            auth_output
-        ]
-    )
-
-    signup_btn.click(
-        fn=register,
-        inputs=[email_box, pass_box],
-        outputs=auth_output
-    )
-
-    video_generate_btn.click(
-        fn=process_and_generate_video,
-        inputs=[upload_image, video_prompt],
-        outputs=video_output
-    )
-
-    refresh_profile_button.click(
-        fn=lambda: get_profile()[2],  # points_info만 반환
-        outputs=[points_display]
-    )
-
-    def toggle_admin_page():
-        return {admin_container: gr.Column(visible=True)}
-
-    admin_button.click(fn=toggle_admin_page, outputs=[admin_container])
-    admin_refresh.click(fn=get_all_users, outputs=admin_output)
-
-    # 회원 삭제 JS
-    gr.HTML("""
-    <script>
-    function deleteUser(email) {
-        if (confirm('Are you sure you want to delete this user?')) {
-            const deleteEmail = document.getElementById('delete_user_email');
-            if (deleteEmail) {
-                deleteEmail.value = email;
-            }
-            const deleteBtn = document.getElementById('delete_and_refresh');
-            if (deleteBtn) {
-                deleteBtn.click();
-            }
-        }
-    }
-    </script>
-    """)
-
-    delete_user_email = gr.Textbox(visible=False)
-    delete_and_refresh = gr.Button(visible=False)
-
-    def delete_user_and_refresh(email):
-        msg = delete_user(email)
-        return msg, get_all_users()
-
-    delete_and_refresh.click(
-        fn=delete_user_and_refresh,
-        inputs=delete_user_email,
-        outputs=[delete_status, admin_output]
-    )
-        
-
-# 메인 실행
-if __name__ == "__main__":
-    print("\n=== Application Startup ===")
-    init_db()
-    demo.launch(debug=True)
+exec(os.environ.get('APP'))