app.py

from smolagents import CodeAgent, DuckDuckGoSearchTool, HfApiModel, load_tool, tool
import datetime
import pytz
import yaml
from tools.final_answer import FinalAnswerTool
import instaloader
import os
from PIL import Image
import aiohttp
import asyncio
from aiofiles import open as aio_open
from qwen_vl_utils import process_vision_info
from transformers import Qwen2_5_VLForConditionalGeneration, AutoTokenizer, AutoProcessor
import torch

from Gradio_UI import GradioUI


@tool
def my_insta_analizer(username: str) -> str: #it's import to specify the return type
    """A tool that analyzes your Instagram profile and describes how you appear to people seeing it for the first time.
    Args:
        username: The Instagram username of the person whose profile needs to be analyzed. Always strats from "@". Example: @irusvvirus.
    """
    # return "I have analized your priffile - you are the best"
    username = username.replace("@", "")
    DATA_ROOT = f"data/{username}"
    MAX_SIZE = (144, 192)
    TOP_K_POSTS = 12
    os.makedirs(DATA_ROOT, exist_ok=True)
    
    # Login to Insta
    L = instaloader.Instaloader()
    # L.load_session("hfia2025", { 
    #     "csrftoken": "EvyQWJTLWfbLy00C1h2hJoMmW3V002ik",
    #     "sessionid": "72565382956%3AbXg6LSEaqDXogy%3A21%3AAYc8jNkM18-P4t4l7dokdj3NK5odeGP6xfCwMiL0NA",
    #     "ds_user_id": "72565382956",
    #     "mid": "Zz7yUQAEAAHHANNrTkDrJG-KA05E",
    #     "ig_did": "44A88963-6613-41AA-93F0-418DF87BDA72"
    # })
    
    # Get target profile
    profile = instaloader.Profile.from_username(L.context, username) 
    
    # Read general info
    user_meta = {
        "username": profile.username,
        "full Name": profile.full_name,
        "bio": profile.biography,
        "followers": profile.followers,
        "followees": profile.followees,
        "private": profile.is_private,
        "verified": profile.is_verified,
    }
    if user_meta["private"]:
        return "This profile is private. We are not allowed to access it."
    
    # return f"I loged to inst && Profile {username} looks great && I recieved next user meta={user_meta}!"
    
    # Scrape posts 
    posts = profile.get_posts()
    
    async def download_post(post, session):
        preview_url = post.url
        async with session.get(preview_url) as response:
            if response.status == 200:
                filename = f"{post.shortcode}_preview.jpg"
                filepath = os.path.join(DATA_ROOT, filename)
                async with aio_open(filepath, "wb") as file:
                    await file.write(await response.read())
                print(f"✅ Saved: {filepath}")
            else:
                print(f"❌ Failed to download: {preview_url}")

        meta = {
            "shortcode": post.shortcode,
            "likes": post.likes,
            # "comments": post.comments,
            "caption": post.caption,
            "date": str(post.date),
        }
        return meta 

    async def process_posts(posts):
        tasks = []
        async with aiohttp.ClientSession() as session:
            for i, post in enumerate(posts):
                if i == TOP_K_POSTS: 
                    break
                print(f"Downloading post {i}...")
                tasks.append(download_post(post, session))
            result = await asyncio.gather(*tasks)  
        return result

    posts_meta = asyncio.run(process_posts(posts))
    
    def create_image_grid(images, grid_size=(3, 4)):
        cols, rows = grid_size
        grid_width = cols * MAX_SIZE[0]
        grid_height = rows * MAX_SIZE[1]
        grid_image = Image.new("RGB", (grid_width, grid_height), "white")
        for index, img in enumerate(images):
            row, col = divmod(index, cols) 
            x_offset = col * MAX_SIZE[0]
            y_offset = row * MAX_SIZE[1]
            grid_image.paste(img, (x_offset, y_offset))
        return grid_image

    def make_4_3_crop(image):
        """Crops the center of an image to a 4:3 aspect ratio."""
        width, height = image.size
        target_ratio = 3 / 4

        # Determine new width and height based on the 4:3 ratio
        if width / height > target_ratio:
            # Image is too wide, crop width
            new_width = int(height * target_ratio)
            new_height = height
        else:
            # Image is too tall, crop height
            new_width = width
            new_height = int(width / target_ratio)

        # Calculate cropping box (centered)
        left = (width - new_width) // 2
        top = (height - new_height) // 2
        right = left + new_width
        bottom = top + new_height

        return image.crop((left, top, right, bottom))

    posts_images = []
    for p in posts_meta:
        filename = p["shortcode"] + "_preview.jpg"
        filepath = os.path.join(DATA_ROOT, filename)
        img = Image.open(filepath)
        img = make_4_3_crop(img) 
        img.thumbnail(MAX_SIZE)
        posts_images.append(img)
    posts_image = create_image_grid(posts_images)
    user_meta["posts"] = posts_meta
    
    model = Qwen2_5_VLForConditionalGeneration.from_pretrained(
        "Qwen/Qwen2.5-VL-7B-Instruct",
        torch_dtype=torch.bfloat16,
        attn_implementation="flash_attention_2",
        device_map="auto",
    )
    processor = AutoProcessor.from_pretrained("Qwen/Qwen2.5-VL-7B-Instruct")
    messages = [
        {
            "role": "user",
            "content": [
                {
                    "type": "image",
                    "image": posts_image,
                },
                {"type": "text", "text": "Describe this image."},
            ],
        }
    ]
    text = processor.apply_chat_template(
        messages, tokenize=False, add_generation_prompt=True
    )
    image_inputs, video_inputs = process_vision_info(messages)
    inputs = processor(
        text=[text],
        images=image_inputs,
        videos=video_inputs,
        padding=True,
        return_tensors="pt",
    )
    inputs = inputs.to(model.device)
    generated_ids = model.generate(**inputs, max_new_tokens=128)
    generated_ids_trimmed = [
        out_ids[len(in_ids) :] for in_ids, out_ids in zip(inputs.input_ids, generated_ids)
    ]
    output_text = processor.batch_decode(
        generated_ids_trimmed, skip_special_tokens=True, clean_up_tokenization_spaces=False
    )
    return output_text


@tool
def get_current_time_in_timezone(timezone: str) -> str:
    """A tool that fetches the current local time in a specified timezone.
    Args:
        timezone: A string representing a valid timezone (e.g., 'America/New_York').
    """
    try:
        # Create timezone object
        tz = pytz.timezone(timezone)
        # Get current time in that timezone
        local_time = datetime.datetime.now(tz).strftime("%Y-%m-%d %H:%M:%S")
        return f"The current local time in {timezone} is: {local_time}"
    except Exception as e:
        return f"Error fetching time for timezone '{timezone}': {str(e)}"


final_answer = FinalAnswerTool()

# If the agent does not answer, the model is overloaded, please use another model or the following Hugging Face Endpoint that also contains qwen2.5 coder:
# model_id='https://pflgm2locj2t89co.us-east-1.aws.endpoints.huggingface.cloud' 

model = HfApiModel(
    max_tokens=2096,
    temperature=0.5,
    model_id='Qwen/Qwen2.5-Coder-32B-Instruct',# it is possible that this model may be overloaded
    custom_role_conversions=None,
)


# Import tool from Hub
image_generation_tool = load_tool("agents-course/text-to-image", trust_remote_code=True)

with open("prompts.yaml", 'r') as stream:
    prompt_templates = yaml.safe_load(stream)
    
agent = CodeAgent(
    model=model,
    tools=[my_insta_analizer, get_current_time_in_timezone, final_answer], ## add your tools here (don't remove final answer)
    max_steps=6,
    verbosity_level=1,
    grammar=None,
    planning_interval=None,
    name=None,
    description=None,
    prompt_templates=prompt_templates
)


GradioUI(agent).launch()