Spaces:

kavehtaheri
/

ocrlight3

Sleeping

App Files Files Community

kavehtaheri commited on Jul 22, 2025

Commit

0824ca5

verified ·

1 Parent(s): 0d83276

Update app.py

Browse files

Files changed (1) hide show

app.py +236 -294

app.py CHANGED Viewed

@@ -1,307 +1,249 @@
-import gradio as gr
-import easyocr
-from PIL import Image
-import numpy as np
-import google.generativeai as genai
-import time
-from gradio_client import Client, handle_file
-import requests # <-- ADDED for API calls
-import io       # <-- ADDED to handle image bytes
-# --- Configuration ---
-# Gemini API key
-GEMINI_API_KEY = "AIzaSyAKI92YawOKQ1-HRLmvaryMEWk_y4alJgA" # Replace with your actual Gemini API Key
-# One-API Token for Instagram Downloader
-# IMPORTANT: Replace with your token from one-api.ir
-ONE_API_TOKEN = "268976:66f4f58a2a905"
-# URL to your background image hosted on your Hugging Face Space.
-BACKGROUND_IMAGE_URL = "1.jpg"
-# Global reader - initialize once
-reader = None
-def initialize_reader():
-    """Initialize EasyOCR reader"""
-    global reader
-    if reader is None:
-        print("Loading EasyOCR model...")
-        reader = easyocr.Reader(['en'], gpu=False, verbose=False)
-        print("EasyOCR model loaded successfully!")
-    return reader
-# --- NEW FUNCTION TO DOWNLOAD FROM INSTAGRAM ---
 def download_instagram_image(url):
     """
-    Downloads the first image from an Instagram post URL using One-API.
-    Returns a PIL Image object or an error message string.
     """
-    if not url or not url.strip():
-        return "Please enter an Instagram URL."
-    if ONE_API_TOKEN == "YOUR_ONE_API_TOKEN_HERE":
-        print("ERROR: ONE_API_TOKEN is not set.")
-        return "ERROR: The Instagram downloader API key is not configured on the server."
-    api_endpoint = f"https://one-api.ir/instagram/?token={ONE_API_TOKEN}&action=post&url={url.strip()}"
-    print(f"DEBUG: Calling One-API for URL: {url.strip()}")
-    try:
-        response = requests.get(api_endpoint, timeout=30) # 30-second timeout
-        response.raise_for_status()  # Raise an exception for bad status codes (4xx or 5xx)
-        data = response.json()
-        if data.get("status") != 200 or not data.get("result"):
-            error_message = data.get("message", "Unknown API error.")
-            print(f"ERROR: One-API failed. Message: {error_message}")
-            return f"API Error: {error_message}"
-        # Find the first image URL in the media list
-        media_list = data["result"].get("media", [])
-        image_url = None
-        for item in media_list:
-            if item.get("type") == "image":
-                image_url = item.get("url")
-                break
-        if not image_url:
-            print("ERROR: No image found in the Instagram post media.")
-            return "No image found in this Instagram post. It might be a video-only post."
-        print(f"DEBUG: Found image URL: {image_url[:50]}...")
-        # Download the image content
-        image_response = requests.get(image_url, timeout=30)
-        image_response.raise_for_status()
-        # Open the image from the downloaded bytes and return a PIL Image
-        image = Image.open(io.BytesIO(image_response.content))
-        print("DEBUG: Instagram image downloaded and converted to PIL Image successfully.")
-        return image
-    except requests.exceptions.RequestException as e:
-        print(f"ERROR: Network error while contacting One-API or downloading image: {e}")
-        return f"Network Error: Could not retrieve data. {e}"
-    except Exception as e:
-        print(f"ERROR: An unexpected error occurred in download_instagram_image: {e}")
-        return f"An unexpected error occurred: {str(e)}"
-def extract_text_from_quote(image):
-    """Extract text from quote image using EasyOCR"""
-    if image is None:
-        return "Please upload an image first.", "Words: 0"
-    try:
-        reader = initialize_reader()
-        img_array = np.array(image)
-        results = reader.readtext(img_array, paragraph=True)
-        if results:
-            text_parts = [result[1].strip() for result in results if len(result) >= 2 and result[1].strip()]
-            if text_parts:
-                extracted_text = ' '.join(text_parts)
-                word_count = f"Words: {len(extracted_text.split())}"
-                return extracted_text, word_count
-        return "No text detected in the image.", "Words: 0"
-    except Exception as e:
-        return f"Error processing image: {str(e)}", "Words: 0"
-def translate_extracted(text, lang):
-    """Translate the extracted English text using Gemini API"""
-    if not text or "No text" in text or "Error" in text or "Please upload" in text:
-        return "No valid text to translate."
     try:
-        print(f"DEBUG: API Key loaded (first 5 chars: {GEMINI_API_KEY[:5]}...). Starting translation to {lang}")
-        genai.configure(api_key=GEMINI_API_KEY)
-        for attempt in range(3):
-            try:
-                model = genai.GenerativeModel('gemini-1.5-flash')
-                prompt = f"""
-                You are a cool, chill translator with a fun and warm personality, inspired by Persian Twitter style.
-                Your translations should be natural, slangy, and relatable. Use colloquial words and contractions.
-                No emojis, keep it RTL-friendly. Be concise but preserve the emotional depth.
-                Maintain correct grammar, even with slang. Avoid literal translations.
-                Translate the following English quote into this style.
-                English Quote: "{text}"
-                Format your output for an image overlay: Break the Persian text into short, visually appealing lines.
-                Output ONLY the translated Persian text.
-                """
-                response = model.generate_content(prompt)
-                translated = response.text.strip()
-                print(f"DEBUG: Translation successful on attempt {attempt+1}: {translated[:50]}...")
-                return translated
-            except Exception as inner_e:
-                print(f"DEBUG: Attempt {attempt+1} failed: {str(inner_e)}. Retrying in 2s...")
-                time.sleep(2)
-        raise Exception("All translation retries failed.")
-    except Exception as e:
-        error_msg = f"Error translating: {str(e)}. Check network access to Google API."
-        print(f"DEBUG: Translation failed: {str(e)}")
-        return error_msg
-def overlay_text_on_image(translated_text):
-    """
-    Sends translated text to the 'textoverimage1' Space and gets the resulting image.
-    """
-    if not translated_text or "Error" in translated_text or "No valid text" in translated_text:
-        print("DEBUG: Skipping image overlay due to invalid translated text.")
-        return None
-    try:
-        print("DEBUG: Initializing client for 'kavehtaheri/textoverimage1'")
-        client = Client("kavehtaheri/textoverimage1")
-        print(f"DEBUG: Sending data to API endpoint '/overlay_text_on_image'")
-        print(f"DEBUG: Persian Text: {translated_text}")
-        print(f"DEBUG: Image URL: {BACKGROUND_IMAGE_URL}")
-        result_image_path = client.predict(
-            persian_text=translated_text,
-            url="",
-            upload=handle_file(BACKGROUND_IMAGE_URL),
-            username="",
-            text_color="Black",
-            api_name="/overlay_text_on_image"
-        )
-        print(f"DEBUG: Received image path from API: {result_image_path}")
-        return result_image_path
-    except Exception as e:
-        print(f"ERROR: Could not get image from 'textoverimage1' space. Error: {e}")
-        return None
-def clear_all():
-    """Clear all inputs and outputs"""
-    return None, "", "Your extracted quote will appear here...", "Words: 0", "Translation will appear here...", None, True
-# --- Main Processing Functions ---
-def process_everything(image, lang):
-    """The core pipeline: OCR -> Translate -> Overlay. Accepts a PIL image."""
-    # Check if the input is valid before starting
-    if image is None:
-        return "Please provide an image first.", "Words: 0", "Translation failed: No image provided.", None
-    text, wc = extract_text_from_quote(image)
-    if "No text" in text or "Error" in text:
-        return text, wc, "Translation failed: No text extracted.", None
-    translated = translate_extracted(text, lang)
-    final_image = overlay_text_on_image(translated)
-    return text, wc, translated, final_image
-def process_from_url(url, lang):
-    """Handler for the Instagram URL workflow."""
-    gr.Info("Downloading image from Instagram...")
-    download_result = download_instagram_image(url)
-    # Check if download_instagram_image returned an error string
-    if isinstance(download_result, str):
-        # The download failed, return the error message to the text output
-        gr.Error(f"Download Failed: {download_result}")
-        return download_result, "Words: 0", "Process failed.", None, None
-    # If successful, it returned a PIL image, so we can process it.
-    gr.Info("Image downloaded! Starting OCR and Translation...")
-    text, wc, translated, final_image = process_everything(download_result, lang)
-    # We also return the downloaded image to populate the input image box
-    return text, wc, translated, final_image, download_result
-# --- Gradio Interface ---
-with gr.Blocks(title="Quote OCR & Overlay", theme=gr.themes.Soft()) as demo:
-    gr.Markdown("# 📝 Quote Text Extractor & Image Generator")
-    gr.Markdown("Extract text from an image, translate it, and overlay it onto a new background. Choose your input method below.")
-    with gr.Row():
-        # --- INPUT COLUMN ---
-        with gr.Column(scale=1):
-            with gr.Tabs():
-                with gr.Tab("1. Upload Image"):
-                    image_input = gr.Image(label="Upload Quote Image", type="pil", sources=["upload", "clipboard"])
-                    auto_process_cb = gr.Checkbox(label="Auto-Process After Upload", value=True)
-                with gr.Tab("1. Download from Instagram URL"):
-                    insta_url_input = gr.Textbox(label="Instagram Post URL", placeholder="Paste a link like https://www.instagram.com/p/C0123ABCD.../")
-                    insta_process_btn = gr.Button("Download & Process", variant="primary")
-            target_lang = gr.Dropdown(
-                label="Target Language",
-                choices=["persian(farsi)"],
-                value="persian(farsi)",
-                interactive=False
-            )
-            with gr.Row():
-                clear_btn = gr.Button("Clear All", variant="secondary")
-                extract_btn = gr.Button("Process Uploaded Image", variant="primary")
-        # --- OUTPUTS COLUMN ---
-        with gr.Column(scale=2):
-            text_output = gr.Textbox(label="2. Extracted English Text", placeholder="Extracted text appears here...", lines=4, show_copy_button=True)
-            word_count = gr.Textbox(label="Word Count", interactive=False, max_lines=1)
-            translated_output = gr.Textbox(label="3. Translated Persian Text", placeholder="Persian translation appears here...", lines=4, show_copy_button=True)
-            gr.Markdown("---")
-            final_image_output = gr.Image(label="4. Final Image with Text Overlay", type="filepath")
-    # --- Event Handlers ---
-    # 1. For the "Process Uploaded Image" button
-    extract_btn.click(
-        fn=process_everything,
-        inputs=[image_input, target_lang],
-        outputs=[text_output, word_count, translated_output, final_image_output]
-    )
-    # 2. For auto-processing when an image is uploaded
-    def auto_process_wrapper(image, lang, is_enabled):
-        if is_enabled:
-            return process_everything(image, lang)
-        # If auto-process is off, just do OCR and stop
-        text, wc = extract_text_from_quote(image)
-        return text, wc, "Translation will appear here...", None
-    image_input.change(
-        fn=auto_process_wrapper,
-        inputs=[image_input, target_lang, auto_process_cb],
-        outputs=[text_output, word_count, translated_output, final_image_output]
-    )
-    # 3. For the "Download & Process" button in the Instagram tab
-    insta_process_btn.click(
-        fn=process_from_url,
-        inputs=[insta_url_input, target_lang],
-        # The outputs now include the image_input to show the downloaded image
-        outputs=[text_output, word_count, translated_output, final_image_output, image_input]
-    )
-    # 4. For the Clear button
-    clear_btn.click(
-        fn=clear_all,
-        outputs=[image_input, insta_url_input, text_output, word_count, translated_output, final_image_output, auto_process_cb]
-    )
-    gr.Markdown("### 💡 How It Works:\n1. **Upload an image** OR **paste an Instagram URL**.\n2. The app automatically extracts the English text using OCR.\n3. The text is translated to a casual, modern Persian.\n4. The Persian text is sent to a second app which overlays it on a background image.\n5. The final image is displayed.")
-if __name__ == "__main__":
-    # Add these to your requirements.txt file:
-    # gradio
-    # easyocr
-    # pillow
-    # numpy
-    # google-generativeai
-    # gradio_client
-    # requests
-    demo.launch(debug=True)

+import os
+import requests
+from flask import Flask, request, jsonify, render_template_string, send_from_directory
+from urllib.parse import urlparse, parse_qs
+import re # Import the regular expressions module
+app = Flask(__name__)
+# Load API token from environment variable
+API_TOKEN = os.environ.get('ONE_API_TOKEN')
+if not API_TOKEN:
+    raise ValueError("No ONE_API_TOKEN set for Flask application")
+# --- HTML Templates ---
+# Main page template
+INDEX_TEMPLATE = """
+<!DOCTYPE html>
+<html lang="en">
+<head>
+    <meta charset="UTF-8">
+    <meta name="viewport" content="width=device-width, initial-scale=1.0">
+    <title>Multi-Downloader Service</title>
+    <style>
+        body { font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, Helvetica, Arial, sans-serif; line-height: 1.6; background-color: #f4f4f9; color: #333; margin: 0; padding: 20px; }
+        .container { max-width: 800px; margin: 40px auto; background: #fff; padding: 30px; border-radius: 8px; box-shadow: 0 4px 10px rgba(0,0,0,0.1); }
+        h1 { color: #5a4bcf; text-align: center; }
+        .form-group { margin-bottom: 20px; }
+        label { display: block; font-weight: bold; margin-bottom: 8px; }
+        input[type="text"], input[type="url"] { width: 100%; padding: 12px; border: 1px solid #ccc; border-radius: 4px; box-sizing: border-box; }
+        button { display: block; width: 100%; background-color: #5a4bcf; color: white; padding: 14px 20px; margin-top: 10px; border: none; border-radius: 4px; cursor: pointer; font-size: 16px; font-weight: bold; }
+        button:hover { background-color: #4a3abc; }
+        #response { margin-top: 25px; padding: 15px; border-radius: 4px; background-color: #e9ecef; border: 1px solid #ced4da; white-space: pre-wrap; word-wrap: break-word; }
+        .loader { display: none; margin: 20px auto; border: 5px solid #f3f3f3; border-top: 5px solid #5a4bcf; border-radius: 50%; width: 50px; height: 50px; animation: spin 1s linear infinite; }
+        @keyframes spin { 0% { transform: rotate(0deg); } 100% { transform: rotate(360deg); } }
+    </style>
+</head>
+<body>
+    <div class="container">
+        <h1>Universal Downloader</h1>
+        <form id="download-form">
+            <div class="form-group">
+                <label for="url">Enter URL:</label>
+                <input type="url" id="url" name="url" placeholder="Enter Instagram, YouTube, etc. URL" required>
+            </div>
+            <button type="submit">Download</button>
+        </form>
+        <div class="loader" id="loader"></div>
+        <div id="response"></div>
+    </div>
+    <script>
+        document.getElementById('download-form').addEventListener('submit', function(event) {
+            event.preventDefault();
+            const url = document.getElementById('url').value;
+            const responseDiv = document.getElementById('response');
+            const loader = document.getElementById('loader');
+            responseDiv.textContent = '';
+            loader.style.display = 'block';
+            fetch('/download', {
+                method: 'POST',
+                headers: {
+                    'Content-Type': 'application/json'
+                },
+                body: JSON.stringify({ url: url })
+            })
+            .then(response => response.json())
+            .then(data => {
+                loader.style.display = 'none';
+                let content = '<h3>Result:</h3>';
+                if (data.error) {
+                    content += `<p style="color: red;">Error: ${data.error}</p>`;
+                } else {
+                    content += `<p><strong>Status:</strong> ${data.status}</p>`;
+                    if (data.filename) {
+                        content += `<p><strong>File:</strong> <a href="/uploads/${data.filename}" target="_blank">${data.filename}</a></p>`;
+                    }
+                    if (data.details) {
+                        content += '<pre>' + JSON.stringify(data.details, null, 2) + '</pre>';
+                    }
+                }
+                responseDiv.innerHTML = content;
+            })
+            .catch(error => {
+                loader.style.display = 'none';
+                responseDiv.innerHTML = `<p style="color: red;"><strong>An unexpected error occurred:</strong> ${error}</p>`;
+            });
+        });
+    </script>
+</body>
+</html>
+"""
+# --- Helper Functions ---
+def get_youtube_video_id(url):
+    """Extracts the YouTube video ID from a URL."""
+    parsed_url = urlparse(url)
+    if parsed_url.hostname == 'youtu.be':
+        return parsed_url.path[1:]
+    if parsed_url.hostname in ('www.youtube.com', 'youtube.com'):
+        if parsed_url.path == '/watch':
+            query_params = parse_qs(parsed_url.query)
+            return query_params.get('v', [None])[0]
+        if parsed_url.path.startswith('/embed/'):
+            return parsed_url.path.split('/embed/')[1]
+        if parsed_url.path.startswith('/v/'):
+            return parsed_url.path.split('/v/')[1]
+    return None
+def download_file(url, local_filename):
+    """Downloads a file from a URL to a local path."""
+    with requests.get(url, stream=True) as r:
+        r.raise_for_status()
+        with open(local_filename, 'wb') as f:
+            for chunk in r.iter_content(chunk_size=8192):
+                f.write(chunk)
+    return local_filename
+# --- API Interaction Functions ---
+def download_youtube_video(video_id):
+    """Fetches YouTube video download link from one-api.ir."""
+    api_url = f"https://one-api.ir/youtube/?token={API_TOKEN}&action=download&id={video_id}"
+    response = requests.get(api_url)
+    if response.status_code != 200:
+        return {"error": f"API request failed with status {response.status_code}", "details": response.text}
+    data = response.json()
+    if data.get("status") == 200 and data.get("result"):
+        # Assuming the best quality is the first one
+        download_url = data["result"]["medias"][0]["url"]
+        filename = f"{data['result']['title']}.mp4"
+        # Ensure the 'uploads' directory exists
+        os.makedirs('uploads', exist_ok=True)
+        local_filepath = os.path.join('uploads', filename)
+        download_file(download_url, local_filepath)
+        return {"status": "Success", "filename": filename, "details": data}
+    else:
+        return {"error": "Failed to get download link from API", "details": data}
+# ========================================================================
+# MODIFIED FUNCTION
+# ========================================================================
 def download_instagram_image(url):
     """
+    Fetches Instagram post details from one-api.ir using the shortcode,
+    then downloads the first image.
     """
+    # 1. Extract the shortcode from the URL using regex
+    #    e.g., from "https://www.instagram.com/p/DMaqqN_RuqQ/" -> "DMaqqN_RuqQ"
+    match = re.search(r"/p/([^/]+)", url)
+    if not match:
+        return {"error": "Invalid Instagram post URL. Could not find shortcode."}
+    shortcode = match.group(1)
+    # 2. Define the new, correct API endpoint and parameters
+    api_url = "https://api.one-api.ir/instagram/v1/post/"
+    params = {"shortcode": shortcode}
+    # 3. Place the token in the headers, not the URL
+    headers = {"one-api-token": API_TOKEN}
     try:
+        # 4. Make the API request
+        response = requests.get(api_url, params=params, headers=headers)
+        # Check for non-200 status codes (like 403, 404, 500)
+        response.raise_for_status()
+        data = response.json()
+        # 5. Process the response and download the image
+        if data.get("status") == 200 and data.get("result"):
+            post_result = data["result"]
+            # Find the first media item that is an image
+            image_url = None
+            if post_result.get("media_type") == 1: # 1 usually means Image
+                image_url = post_result["media_url"]
+            elif post_result.get("media_type") == 8: # 8 usually means Carousel/Album
+                for item in post_result.get("carousel_media", []):
+                    if item.get("media_type") == 1:
+                        image_url = item["media_url"]
+                        break
+            if not image_url:
+                return {"error": "No downloadable image found in the post.", "details": data}
+            # Create a filename and download
+            filename = f"instagram_{shortcode}.jpg"
+            os.makedirs('uploads', exist_ok=True)
+            local_filepath = os.path.join('uploads', filename)
+            download_file(image_url, local_filepath)
+            return {"status": "Success", "filename": filename, "details": data}
+        else:
+            # Handle API-level errors (e.g., status: 404 - not found)
+            return {"error": data.get("message", "Failed to get data from API"), "details": data}
+    except requests.exceptions.HTTPError as http_err:
+        # This will catch the 403 Forbidden error and others
+        return {"error": f"HTTP Error: {http_err}", "details": http_err.response.text}
+    except requests.exceptions.RequestException as req_err:
+        # Handle other request errors like network issues
+        return {"error": f"Request Exception: {req_err}"}
+# ========================================================================
+# END OF MODIFIED FUNCTION
+# ========================================================================
+# --- Flask Routes ---
+@app.route('/')
+def index():
+    """Serves the main HTML page."""
+    return render_template_string(INDEX_TEMPLATE)
+@app.route('/download', methods=['POST'])
+def download():
+    """Handles the download request by dispatching to the correct function."""
+    data = request.get_json()
+    url = data.get('url')
+    if not url:
+        return jsonify({"error": "URL is required"}), 400
+    hostname = urlparse(url).hostname
+    if 'youtube.com' in hostname or 'youtu.be' in hostname:
+        video_id = get_youtube_video_id(url)
+        if not video_id:
+            return jsonify({"error": "Invalid YouTube URL"}), 400
+        result = download_youtube_video(video_id)
+    elif 'instagram.com' in hostname:
+        result = download_instagram_image(url)
+    else:
+        return jsonify({"error": "Unsupported URL. Please use a YouTube or Instagram link."}), 400
+    return jsonify(result)
+@app.route('/uploads/<filename>')
+def uploaded_file(filename):
+    """Serves downloaded files from the 'uploads' directory."""
+    return send_from_directory('uploads', filename)
+if __name__ == '__main__':
+    app.run(host='0.0.0.0', port=5000)