app.py

import gradio as gr
import time
import os
import subprocess
import sys
from PIL import Image
import io
import base64
import logging
import traceback
from datetime import datetime

# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

# Create screenshots directory if it doesn't exist
os.makedirs('screenshots', exist_ok=True)

# Global variable to track browser installation
BROWSERS_INSTALLED = False

def download_playwright_browsers():
    """Download and install Playwright browsers"""
    global BROWSERS_INSTALLED
    
    if BROWSERS_INSTALLED:
        return "Playwright browsers already installed."
    
    try:
        print("Installing Playwright browsers...")
        
        # First install playwright itself if needed
        subprocess.run([sys.executable, "-m", "pip", "install", "playwright==1.39.0"], check=True)
        
        # Then install the browsers
        result = subprocess.run(
            [sys.executable, "-m", "playwright", "install", "chromium"], 
            check=True,
            capture_output=True, 
            text=True
        )
        
        print("Installation output:", result.stdout)
        if result.stderr:
            print("Installation errors:", result.stderr)
            
        BROWSERS_INSTALLED = True
        return "Playwright browsers installed successfully."
    except Exception as e:
        error_msg = f"Error installing browsers: {str(e)}"
        print(error_msg)
        return error_msg

def save_screenshot(name):
    """Save a placeholder screenshot if we can't get a real one"""
    try:
        timestamp = int(time.time())
        filename = f"screenshots/{name}_{timestamp}.png"
        
        # Create a simple colored image as a placeholder
        img = Image.new('RGB', (400, 300), color=(73, 109, 137))
        img.save(filename)
        
        logger.info(f"Saved placeholder screenshot: {filename}")
        return filename
    except Exception as e:
        logger.error(f"Error creating placeholder: {e}")
        return None

def login_and_like_posts(username, password, max_likes):
    """Main function to login to Instagram and like posts"""
    status_updates = ["Starting Instagram Auto-Liker..."]
    image_path = save_screenshot("start")
    
    # First ensure browsers are installed
    install_msg = download_playwright_browsers()
    status_updates.append(install_msg)
    
    try:
        # Only import playwright after ensuring it's installed
        from playwright.sync_api import sync_playwright
        
        status_updates.append("Launching browser...")
        
        with sync_playwright() as p:
            browser = p.chromium.launch(
                headless=True,
                args=['--no-sandbox', '--disable-dev-shm-usage']
            )
            
            # Create a context with realistic user agent
            context = browser.new_context(
                user_agent='Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
                viewport={'width': 1280, 'height': 720}
            )
            
            # Create a new page
            page = context.new_page()
            
            # First test browser by visiting a simple site
            status_updates.append("Testing browser connection...")
            page.goto("https://www.google.com")
            page.wait_for_load_state("networkidle")
            status_updates.append(f"Browser working. Title: {page.title()}")
            
            # Take test screenshot
            try:
                test_screenshot = f"screenshots/browser_test_{int(time.time())}.png"
                page.screenshot(path=test_screenshot)
                image_path = test_screenshot
                status_updates.append(f"Browser test screenshot saved")
            except Exception as e:
                status_updates.append(f"Error taking screenshot: {str(e)}")
            
            # Instagram login
            status_updates.append("Navigating to Instagram...")
            page.goto("https://www.instagram.com/")
            page.wait_for_load_state("networkidle")
            
            # Take screenshot of Instagram landing
            try:
                landing_screenshot = f"screenshots/instagram_landing_{int(time.time())}.png"
                page.screenshot(path=landing_screenshot)
                image_path = landing_screenshot
                status_updates.append("Instagram home page loaded")
            except Exception as e:
                status_updates.append(f"Error taking landing screenshot: {str(e)}")
            
            # Handle cookie consent if present
            try:
                if page.query_selector('text=Accept') or page.query_selector('text=Allow'):
                    status_updates.append("Handling cookie consent...")
                    
                    # Try different button text variants
                    for button_text in ["Accept", "Allow", "Accept All", "Allow All Cookies"]:
                        try:
                            button = page.query_selector(f'text={button_text}')
                            if button:
                                button.click()
                                status_updates.append(f"Clicked '{button_text}' button")
                                page.wait_for_timeout(2000)
                                break
                        except Exception as btn_error:
                            logger.warning(f"Error clicking {button_text}: {str(btn_error)}")
                            continue
            except Exception as e:
                status_updates.append(f"No cookie dialog or error: {str(e)}")
            
            # Wait for and fill login form
            status_updates.append("Looking for login form...")
            try:
                page.wait_for_selector('input[name="username"]', state="visible", timeout=10000)
                
                # Take pre-login screenshot
                try:
                    pre_login_screenshot = f"screenshots/pre_login_{int(time.time())}.png"
                    page.screenshot(path=pre_login_screenshot)
                    image_path = pre_login_screenshot
                except Exception as e:
                    status_updates.append(f"Error taking pre-login screenshot: {str(e)}")
                
                # Enter credentials
                status_updates.append(f"Entering username: {username}")
                page.fill('input[name="username"]', username)
                page.fill('input[name="password"]', password)
                status_updates.append("Credentials entered")
                
                # Click login button
                status_updates.append("Clicking login button...")
                page.click('button[type="submit"]')
                
                # Wait for navigation
                page.wait_for_load_state("networkidle")
                page.wait_for_timeout(5000)  # Additional wait for elements
                
                # Take post-login screenshot
                try:
                    post_login_screenshot = f"screenshots/post_login_{int(time.time())}.png"
                    page.screenshot(path=post_login_screenshot)
                    image_path = post_login_screenshot
                except Exception as e:
                    status_updates.append(f"Error taking post-login screenshot: {str(e)}")
                
                # Handle "Save Login Info" popup if it appears
                try:
                    not_now_button = page.query_selector('text="Not Now"')
                    if not_now_button:
                        not_now_button.click()
                        status_updates.append("Dismissed 'Save Login Info' popup")
                        page.wait_for_timeout(2000)
                except Exception as e:
                    status_updates.append(f"No 'Save Login Info' popup or error: {str(e)}")
                
                # Handle notifications popup if it appears
                try:
                    not_now_button = page.query_selector('text="Not Now"')
                    if not_now_button:
                        not_now_button.click()
                        status_updates.append("Dismissed notifications popup")
                        page.wait_for_timeout(2000)
                except Exception as e:
                    status_updates.append(f"No notifications popup or error: {str(e)}")
                
                # Verify we're on the feed
                try:
                    feed_screenshot = f"screenshots/feed_{int(time.time())}.png"
                    page.screenshot(path=feed_screenshot)
                    image_path = feed_screenshot
                    status_updates.append("Successfully logged in! On Instagram feed now.")
                except Exception as e:
                    status_updates.append(f"Error taking feed screenshot: {str(e)}")
                
                # Start liking posts
                status_updates.append(f"Starting to like posts (target: {max_likes})...")
                
                # Like posts
                likes_count = 0
                scroll_count = 0
                max_scrolls = 30
                
                # Try to find posts
                try:
                    page.wait_for_selector('article', timeout=10000)
                    status_updates.append("Found posts on feed")
                    
                    while likes_count < max_likes and scroll_count < max_scrolls:
                        # Find like buttons
                        like_buttons = page.query_selector_all('article svg[aria-label="Like"]')
                        
                        status_updates.append(f"Found {len(like_buttons)} like buttons on scroll {scroll_count}")
                        
                        if len(like_buttons) == 0 and scroll_count > 5:
                            status_updates.append("No more like buttons found. Stopping.")
                            break
                        
                        # Click like buttons
                        for i, button in enumerate(like_buttons):
                            if likes_count >= max_likes:
                                break
                            
                            try:
                                # Scroll to button
                                button.scroll_into_view_if_needed()
                                page.wait_for_timeout(500)
                                
                                # Click like
                                button.click()
                                likes_count += 1
                                status_updates.append(f"Liked post {likes_count}/{max_likes}")
                                
                                # Wait between likes
                                page.wait_for_timeout(2000)
                            except Exception as e:
                                status_updates.append(f"Error liking post {i+1}: {str(e)}")
                                continue
                        
                        # Scroll down to load more
                        page.evaluate("window.scrollBy(0, 1000)")
                        status_updates.append(f"Scrolled down to load more posts")
                        page.wait_for_timeout(3000)
                        scroll_count += 1
                    
                    # Final status
                    final_message = f"Finished! Liked {likes_count} posts."
                    status_updates.append(final_message)
                    
                except Exception as e:
                    status_updates.append(f"Error finding or liking posts: {str(e)}")
                
            except Exception as e:
                status_updates.append(f"Error with login form: {str(e)}")
            
            # Close the browser
            browser.close()
            status_updates.append("Browser closed")
            
            return "\n".join(status_updates), image_path
    except Exception as e:
        error_message = f"Error: {str(e)}"
        logger.error(error_message)
        status_updates.append(error_message)
        status_updates.append(traceback.format_exc())
        
        return "\n".join(status_updates), image_path

# Simple interface for testing
def test_functionality():
    """Simple function to test if the app is working"""
    return "Instagram Auto-Liker is operational", save_screenshot("test")

# Gradio Interface
def create_interface():
    with gr.Blocks(title="Instagram Auto-Liker") as app:
        gr.Markdown("# Instagram Auto-Liker")
        gr.Markdown("Enter your Instagram credentials and the number of posts to like.")
        
        with gr.Row():
            with gr.Column(scale=1):
                username = gr.Textbox(label="Instagram Username")
                password = gr.Textbox(label="Instagram Password", type="password")
                max_likes = gr.Slider(minimum=1, maximum=100, value=10, step=1, label="Number of Posts to Like")
                submit_btn = gr.Button("Start Liking Posts")
                test_btn = gr.Button("Test App Function")
            
            with gr.Column(scale=2):
                status_output = gr.Textbox(label="Status Log", lines=15)
                image_output = gr.Image(label="Latest Screenshot", type="filepath")
        
        submit_btn.click(
            fn=login_and_like_posts,
            inputs=[username, password, max_likes],
            outputs=[status_output, image_output]
        )
        
        test_btn.click(
            fn=test_functionality,
            inputs=[],
            outputs=[status_output, image_output]
        )
    
    return app

# Make sure we install browsers first at module level
try:
    download_playwright_browsers()
except Exception as e:
    print(f"Warning: Failed to install browsers at startup: {e}")

# Launch the app
if __name__ == "__main__":
    print("Creating interface...")
    app = create_interface()
    print("Launching interface...")
    app.launch(server_name="0.0.0.0", server_port=7860, share=True)
    print("Interface launched.")