Create app.py

app.py

@@ -0,0 +1,162 @@
+from fastapi import FastAPI, HTTPException
+from fastapi.middleware.cors import CORSMiddleware
+from pydantic import BaseModel, HttpUrl
+from typing import List, Optional
+import asyncio
+import os
+import aiograpi
+from aiograpi import Client
+from datetime import datetime
+
+app = FastAPI(
+    title="Instagram Scraper API",
+    description="An API to scrape Instagram posts with Apify-like formatting",
+    version="1.0.0",
+    docs_url="/docs",
+)
+
+# Add CORS middleware
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+# Pydantic models for request and response
+class ScraperRequest(BaseModel):
+    username: str
+    post_count: int = 10
+    session_id: Optional[str] =None 
+
+class TaggedUser(BaseModel):
+    username: str
+    id: str
+
+
+# Update the ChildPost model to handle None values for URLs
+class ChildPost(BaseModel):
+    displayUrl: Optional[HttpUrl] = None
+    videoUrl: Optional[HttpUrl] = None
+    type: str
+
+# Update the InstagramPost model similarly
+class InstagramPost(BaseModel):
+    debug_id: str
+    debug_mediaType: int
+    type: str
+    id: str
+    shortCode: str
+    displayUrl: Optional[HttpUrl] = None
+    videoUrl: Optional[HttpUrl] = None
+    caption: Optional[str] = None
+    hashtags: List[str]
+    mentions: List[str]
+    url: HttpUrl
+    commentsCount: int
+    dimensionsHeight: Optional[int] = None
+    dimensionsWidth: Optional[int] = None
+    likesCount: int
+    timestamp: str
+    childPosts: List[ChildPost]
+    locationName: Optional[str] = None
+    locationId: Optional[str] = None
+    ownerUsername: str
+    ownerId: str
+    ownerFullName: Optional[str] = None
+    viewCount: Optional[int] = None
+    albumId: Optional[str] = None
+    alt: Optional[str] = None
+    sponsored: bool
+    taggedUsers: List[TaggedUser]
+
+# Update the format_media_data function to handle URLs properly
+async def format_media_data(media):
+    """Format media data to match Apify Instagram scraper output"""
+    return {
+        "debug_id": str(media.pk),
+        "debug_mediaType": media.media_type,
+        "type": "Image" if media.media_type == 1 else "Video" if media.media_type == 2 else "Carousel",
+        "id": str(media.pk),
+        "shortCode": media.code,
+        "displayUrl": str(media.thumbnail_url) if media.thumbnail_url else None,
+        "videoUrl": str(media.video_url) if media.video_url else None,
+        "caption": media.caption_text,
+        "hashtags": [tag[1:] for tag in media.caption_text.split() if tag.startswith("#")] if media.caption_text else [],
+        "mentions": [mention[1:] for mention in media.caption_text.split() if mention.startswith("@")] if media.caption_text else [],
+        "url": f"https://www.instagram.com/p/{media.code}/",
+        "commentsCount": media.comment_count,
+        "dimensionsHeight": None,
+        "dimensionsWidth": None,
+        "likesCount": media.like_count,
+        "timestamp": media.taken_at.isoformat() if media.taken_at else None,
+        "childPosts": [
+            {
+                "displayUrl": str(resource.thumbnail_url) if hasattr(resource, 'thumbnail_url') and resource.thumbnail_url else None,
+                "videoUrl": str(resource.video_url) if hasattr(resource, 'video_url') and resource.video_url else None,
+                "type": "Image" if resource.media_type == 1 else "Video" if resource.media_type == 2 else "Carousel"
+            } for resource in media.resources
+        ] if media.resources else [],
+        "locationName": media.location.name if media.location else None,
+        "locationId": str(media.location.pk) if media.location else None,
+        "ownerUsername": media.user.username,
+        "ownerId": str(media.user.pk),
+        "ownerFullName": getattr(media.user, 'full_name', None),
+        "viewCount": media.view_count if media.view_count else None,
+        "albumId": None,
+        "alt": media.accessibility_caption,
+        "sponsored": bool(media.sponsor_tags) or media.is_paid_partnership,
+        "taggedUsers": [
+            {
+                "username": tag.user.username,
+                "id": str(tag.user.pk)
+            } for tag in media.usertags
+        ] if media.usertags else []
+    }
+
+class ScraperResponse(BaseModel):
+    success: bool
+    message: str
+    posts_count: int
+    posts: List[InstagramPost]
+
+
+@app.post("/scrape", response_model=ScraperResponse, tags=["Scraper"])
+async def scrape_instagram(request: ScraperRequest):
+    """
+    Scrape Instagram posts for a given username
+    
+    - **username**: Instagram username to scrape
+    - **post_count**: Number of posts to scrape (default: 10)
+    - **session_id**: Instagram session ID for authentication
+    """
+    try:
+        cl = Client()
+        await cl.login_by_sessionid(os.getenv('session_id',request.session_id))        
+        user_id = await cl.user_id_from_username(request.username)
+        medias = await cl.user_medias(user_id, request.post_count)
+        
+        formatted_posts = []
+        for media in medias:
+            formatted_post = await format_media_data(media)
+            formatted_posts.append(formatted_post)
+        
+        return ScraperResponse(
+            success=True,
+            message=f"Successfully scraped {len(formatted_posts)} posts from {request.username}",
+            posts_count=len(formatted_posts),
+            posts=formatted_posts
+        )
+    
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+@app.get("/health", tags=["Health"])
+async def health_check():
+    """Check if the API is running"""
+    return {"status": "healthy", "timestamp": datetime.now().isoformat()}
+
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run(app, host="0.0.0.0", port=8000)