Commit

app.py

@@ -0,0 +1,208 @@
+# api/main.py
+from fastapi import FastAPI, HTTPException, Depends
+from pydantic import BaseModel
+from typing import List, Dict
+import logging
+import requests
+from io import BytesIO
+from PIL import Image
+import pytesseract
+from textblob import TextBlob
+import pandas as pd
+import joblib
+from sqlalchemy.orm import Session
+from utils.database import init_db, save_to_db, fetch_posts_from_db, get_db
+from utils.instaloader_utils import fetch_user_posts, fetch_competitors_data
+import torch
+from torchvision import transforms
+from transformers import ResNetForImageClassification
+
+# Set up logging
+logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
+
+# Initialize FastAPI app
+app = FastAPI()
+
+# Initialize database
+init_db()
+
+# Load models
+viral_model = joblib.load("api/models/viral_potential_model.pkl")
+engagement_model = joblib.load("api/models/engagement_rate_model.pkl")
+promotion_model = joblib.load("api/models/promotion_strategy_model.pkl")
+
+class UserRequest(BaseModel):
+    username: str
+
+class AnalyzePostRequest(BaseModel):
+    caption: str
+    hashtags: str
+    image_url: str
+
+@app.post("/fetch-posts")
+async def fetch_posts(user: UserRequest):
+    """
+    Fetch posts from a given Instagram profile (public data only).
+    """
+    username = user.username
+    logging.info(f"Fetching posts for user: {username}")
+
+    try:
+        # Fetch user's posts
+        user_posts = fetch_user_posts(username)
+        if not user_posts:
+            raise HTTPException(status_code=404, detail="No posts found for the user.")
+
+        # Fetch competitors' posts
+        competitors_posts = fetch_competitors_data(username)
+
+        # Combine user and competitor data
+        all_posts = user_posts + competitors_posts
+
+        # Save data to the database
+        save_to_db(all_posts)
+
+        return {"status": "success", "data": all_posts}
+    except Exception as e:
+        logging.error(f"Error fetching posts: {e}")
+        raise HTTPException(status_code=500, detail=str(e))
+
+@app.post("/analyze")
+async def analyze(user: UserRequest, db: Session = Depends(get_db)):
+    """
+    Analyze user and competitor data.
+    """
+    username = user.username
+    logging.info(f"Analyzing data for user: {username}")
+
+    try:
+        # Fetch data from the database
+        user_posts = fetch_posts_from_db(username)
+        if not user_posts:
+            raise HTTPException(status_code=404, detail="No posts found for the user.")
+
+        # Perform analysis (e.g., viral potential, engagement rate, etc.)
+        analysis_results = {
+            "viral_potential": predict_viral_potential(user_posts),
+            "top_hashtags": recommend_hashtags(user_posts),
+            "engagement_stats": {
+                "mean_likes": sum(post['likes'] for post in user_posts) / len(user_posts),
+                "mean_comments": sum(post['comments'] for post in user_posts) / len(user_posts)
+            }
+        }
+
+        return {"status": "success", "results": analysis_results}
+    except Exception as e:
+        logging.error(f"Error analyzing data: {e}")
+        raise HTTPException(status_code=500, detail=str(e))
+
+@app.post("/analyze-post")
+async def analyze_post(post: AnalyzePostRequest, db: Session = Depends(get_db)):
+    """
+    Analyze a single post (caption, hashtags, and image).
+    """
+    try:
+        # Download and analyze the image
+        response = requests.get(post.image_url)
+        response.raise_for_status()
+        image = Image.open(BytesIO(response.content))
+
+        # Extract text from the image
+        extracted_text = extract_text_from_image(image)
+
+        # Analyze the image content
+        image_analysis = analyze_image(image)
+
+        # Preprocess input for models
+        features = {
+            'caption_length': len(post.caption),
+            'hashtag_count': len(post.hashtags.split(",")),
+            'sentiment': TextBlob(post.caption).sentiment.polarity
+        }
+        features_df = pd.DataFrame([features])
+
+        # Make predictions
+        viral_score = viral_model.predict_proba(features_df)[0][1]
+        engagement_rate = engagement_model.predict(features_df)[0]
+        promote = promotion_model.predict(features_df)[0]
+
+        # Save post to database
+        post_data = {
+            "caption": post.caption,
+            "hashtags": post.hashtags,
+            "image_url": post.image_url,
+            "engagement_rate": engagement_rate,
+            "viral_score": viral_score,
+            "promote": bool(promote)
+        }
+        save_to_db([post_data])
+
+        return {
+            "extracted_text": extracted_text,
+            "image_analysis": image_analysis,
+            "viral_score": viral_score,
+            "engagement_rate": engagement_rate,
+            "promote": bool(promote)
+        }
+    except Exception as e:
+        logging.error(f"Error analyzing post: {e}")
+        raise HTTPException(status_code=500, detail=str(e))
+
+# Image processing functions
+def resize_image(image, max_size=(800, 600)):
+    """Resize an image to the specified maximum size."""
+    image.thumbnail(max_size)
+    return image
+
+def extract_text_from_image(image):
+    """Extract text from an image using OCR."""
+    try:
+        image = resize_image(image)
+        text = pytesseract.image_to_string(image)
+        return text
+    except Exception as e:
+        logging.error(f"Error extracting text from image: {e}")
+        return ""
+
+def analyze_image(image):
+    """Analyze image content using a pre-trained model."""
+    try:
+        preprocess = transforms.Compose([
+            transforms.Resize(256),
+            transforms.CenterCrop(224),
+            transforms.ToTensor(),
+            transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]),
+        ])
+        image_tensor = preprocess(image).unsqueeze(0)
+        
+        # Load ResNet model
+        model = ResNetForImageClassification.from_pretrained("microsoft/resnet-50")
+        model.eval()
+        
+        with torch.no_grad():
+            output = model(image_tensor)
+        return output.logits.tolist()  # Return the logits as a list
+    except Exception as e:
+        logging.error(f"Error analyzing image: {e}")
+        return None
+
+# Helper functions
+def predict_viral_potential(posts: List[Dict]) -> List[Dict]:
+    """
+    Predict viral potential for posts.
+    """
+    # Placeholder for viral potential prediction logic
+    return [{"caption": post["caption"], "viral_score": 0.8} for post in posts]
+
+def recommend_hashtags(posts: List[Dict]) -> List[str]:
+    """
+    Recommend trending hashtags.
+    """
+    hashtags = [hashtag for post in posts for hashtag in post['hashtags']]
+    hashtag_counts = Counter(hashtags)
+    return [hashtag for hashtag, _ in hashtag_counts.most_common(10)]
+
+# Run the API
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run(app, host="0.0.0.0", port=8000)

endpoints/analytics.py

@@ -0,0 +1,20 @@
+from fastapi import APIRouter, HTTPException
+from api.utils.preprocessing import preprocess_data
+from api.utils.visualization import generate_engagement_heatmap
+from api.utils.logging_utils import logger
+
+router = APIRouter()
+
+@router.post("/analyze")
+async def analyze(username: str):
+    try:
+        # Preprocess data
+        data = preprocess_data(username)
+
+        # Generate engagement heatmap
+        heatmap = generate_engagement_heatmap(data)
+
+        return {"status": "success", "heatmap": heatmap}
+    except Exception as e:
+        logger.error(f"Error analyzing data: {e}")
+        raise HTTPException(status_code=500, detail=str(e))

endpoints/database.py

@@ -0,0 +1,18 @@
+from sqlalchemy.orm import Session
+from models import Post, SessionLocal
+
+# Get database session
+def get_db():
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()
+
+# Save post to database
+def save_post(db: Session, post_data: dict):
+    post = Post(**post_data)
+    db.add(post)
+    db.commit()
+    db.refresh(post)
+    return post

endpoints/models.py

@@ -0,0 +1,28 @@
+from sqlalchemy import create_engine, Column, Integer, String, Float, DateTime
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.orm import sessionmaker
+
+# Database URL (e.g., SQLite, PostgreSQL, MySQL)
+DATABASE_URL = "sqlite:///./instagram_ai.db"
+
+# Create engine and session
+engine = create_engine(DATABASE_URL)
+SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+
+# Base class for models
+Base = declarative_base()
+
+# Define Post model
+class Post(Base):
+    __tablename__ = "posts"
+    id = Column(Integer, primary_key=True, index=True)
+    caption = Column(String, nullable=False)
+    hashtags = Column(String)
+    image_url = Column(String)
+    posting_time = Column(DateTime)
+    engagement_rate = Column(Float)
+    viral_score = Column(Float)
+    promote = Column(Integer)  # 1 for promote, 0 for don't promote
+
+# Create tables
+Base.metadata.create_all(bind=engine)

endpoints/posts.py

@@ -0,0 +1,63 @@
+from fastapi import APIRouter, Depends, HTTPException
+from sqlalchemy.orm import Session
+from database import get_db, save_post
+from utils.preprocessing import preprocess_data
+from utils.image_processing import extract_text_from_image, analyze_image
+import joblib
+import requests
+from PIL import Image
+from io import BytesIO
+from textblob import TextBlob
+import pandas as pd
+
+# Load models
+viral_model = joblib.load("models/viral_potential_model.pkl")
+engagement_model = joblib.load("models/engagement_rate_model.pkl")
+promotion_model = joblib.load("models/promotion_strategy_model.pkl")
+
+router = APIRouter()
+
+# Endpoint to analyze and save a post
+@router.post("/analyze-post")
+async def analyze_post(caption: str, hashtags: str, image_url: str, db: Session = Depends(get_db)):
+    try:
+        # Download and analyze the image
+        response = requests.get(image_url)
+        response.raise_for_status()
+        image = Image.open(BytesIO(response.content))
+        extracted_text = extract_text_from_image(image)
+        image_analysis = analyze_image(image)
+
+        # Preprocess input for models
+        features = {
+            'caption_length': len(caption),
+            'hashtag_count': len(hashtags.split(",")),
+            'sentiment': TextBlob(caption).sentiment.polarity
+        }
+        features_df = pd.DataFrame([features])
+
+        # Make predictions
+        viral_score = viral_model.predict_proba(features_df)[0][1]
+        engagement_rate = engagement_model.predict(features_df)[0]
+        promote = promotion_model.predict(features_df)[0]
+
+        # Save post to database
+        post_data = {
+            "caption": caption,
+            "hashtags": hashtags,
+            "image_url": image_url,
+            "engagement_rate": engagement_rate,
+            "viral_score": viral_score,
+            "promote": promote
+        }
+        save_post(db, post_data)
+
+        return {
+            "extracted_text": extracted_text,
+            "image_analysis": image_analysis,
+            "viral_score": viral_score,
+            "engagement_rate": engagement_rate,
+            "promote": bool(promote)
+        }
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))

models/engagement_rate_model.pkl

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:aafaee190ed53d6859751ab011e123134648137afcf840f57ae58e8d1e97445a
+size 73122

models/promotion_strategy_model.pkl

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:1d9f980263c17b1c8f05732ede25d5235caf5138f8d00620d457762ae0da21ec
+size 1247

models/prophet_model.pkl

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:359ea47c7383c48dcaee07b947b8c6ef4f77694a1f2d39dc194861eee9b3e84c
+size 178763

models/viral_potential_model.pkl

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:497dd50e7db8b9b7419121babc96a01efd5fbdb35a365971a289f9bba7c8d784
+size 48185

requirements.txt

@@ -0,0 +1,25 @@
+pandas
+numpy
+scikit-learn
+xgboost
+statsmodels
+tensorflow
+textblob
+imbalanced-learn
+matplotlib
+uvicorn
+prophet
+seaborn
+pytesseract
+torch
+torchvision
+Pillow
+python-dotenv
+transformers
+python-dotenv
+requests
+datasets
+plotly
+fastapi
+huggingface_hub
+kaggle

schemas.js

@@ -0,0 +1,11 @@
+const googleTrends = require('google-trends-api');
+
+const keyword = process.argv[2];
+
+googleTrends.interestOverTime({ keyword: keyword, startTime: new Date('2023-01-01') })
+  .then(function(results) {
+    console.log(results);
+  })
+  .catch(function(err) {
+    console.error(err);
+  });

utils/database.py

@@ -0,0 +1,114 @@
+# api/utils/database.py
+import sqlite3
+import json
+from typing import List, Dict
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker, Session
+
+# SQLite database setup
+DATABASE = "instagram_ai.db"
+
+def init_db():
+    """
+    Initialize the SQLite database.
+    """
+    conn = sqlite3.connect(DATABASE)
+    cursor = conn.cursor()
+    cursor.execute('''
+        CREATE TABLE IF NOT EXISTS posts (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            username TEXT NOT NULL,
+            caption TEXT,
+            hashtags TEXT,
+            likes INTEGER,
+            comments INTEGER,
+            date TEXT,
+            image_url TEXT UNIQUE,  # Ensure image_url is unique
+            engagement_rate REAL,
+            viral_score REAL,
+            promote BOOLEAN
+        )
+    ''')
+    conn.commit()
+    conn.close()
+
+def post_exists(image_url: str) -> bool:
+    """
+    Check if a post with the given image_url already exists in the database.
+    """
+    conn = sqlite3.connect(DATABASE)
+    cursor = conn.cursor()
+    cursor.execute('SELECT id FROM posts WHERE image_url = ?', (image_url,))
+    result = cursor.fetchone()
+    conn.close()
+    return result is not None
+
+def save_to_db(data: List[Dict]):
+    """
+    Save data to the SQLite database, avoiding duplicates.
+    """
+    conn = sqlite3.connect(DATABASE)
+    cursor = conn.cursor()
+    for post in data:
+        # Check if the post already exists
+        if not post_exists(post.get('image_url')):
+            cursor.execute('''
+                INSERT INTO posts (username, caption, hashtags, likes, comments, date, image_url, engagement_rate, viral_score, promote)
+                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+            ''', (
+                post.get('username', ''),
+                post.get('caption', ''),
+                json.dumps(post.get('hashtags', [])),  # Store hashtags as JSON
+                post.get('likes', 0),
+                post.get('comments', 0),
+                post.get('date', ''),  # Date is already a string
+                post.get('image_url', ''),
+                post.get('engagement_rate', 0.0),
+                post.get('viral_score', 0.0),
+                post.get('promote', False)
+            ))
+    conn.commit()
+    conn.close()
+    print(f"Data saved to database: {DATABASE}")
+
+def fetch_posts_from_db(username: str) -> List[Dict]:
+    """
+    Fetch posts from the database for a given username.
+    """
+    conn = sqlite3.connect(DATABASE)
+    cursor = conn.cursor()
+    cursor.execute('SELECT * FROM posts WHERE username = ?', (username,))
+    rows = cursor.fetchall()
+    conn.close()
+
+    # Convert rows to a list of dictionaries
+    posts = []
+    for row in rows:
+        posts.append({
+            "username": row[1],
+            "caption": row[2],
+            "hashtags": json.loads(row[3]),  # Convert JSON back to list
+            "likes": row[4],
+            "comments": row[5],
+            "date": row[6],  # Date is already a string
+            "image_url": row[7],
+            "engagement_rate": row[8],
+            "viral_score": row[9],
+            "promote": bool(row[10])
+        })
+    return posts
+
+# SQLAlchemy setup for dependency injection
+SQLALCHEMY_DATABASE_URL = f"sqlite:///{DATABASE}"
+engine = create_engine(SQLALCHEMY_DATABASE_URL)
+SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+
+def get_db():
+    """
+    Dependency to get a database session.
+    """
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()

utils/image_processing.py

@@ -0,0 +1,43 @@
+from PIL import Image
+import pytesseract
+import torch
+from torchvision import transforms
+from transformers import ResNetForImageClassification
+import logging
+
+def resize_image(image, max_size=(800, 600)):
+    """Resize an image to the specified maximum size."""
+    image.thumbnail(max_size)
+    return image
+
+def extract_text_from_image(image):
+    """Extract text from an image using OCR."""
+    try:
+        image = resize_image(image)
+        text = pytesseract.image_to_string(image)
+        return text
+    except Exception as e:
+        logging.error(f"Error extracting text from image: {e}")
+        return ""
+
+def analyze_image(image):
+    """Analyze image content using a pre-trained model."""
+    try:
+        preprocess = transforms.Compose([
+            transforms.Resize(256),
+            transforms.CenterCrop(224),
+            transforms.ToTensor(),
+            transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]),
+        ])
+        image_tensor = preprocess(image).unsqueeze(0)
+        
+        # Load ResNet model
+        model = ResNetForImageClassification.from_pretrained("microsoft/resnet-50")
+        model.eval()
+        
+        with torch.no_grad():
+            output = model(image_tensor)
+        return output
+    except Exception as e:
+        logging.error(f"Error analyzing image: {e}")
+        return None

utils/instaloader_utils.py

@@ -0,0 +1,71 @@
+# api/utils/instaloader_utils.py
+import instaloader
+from typing import List, Dict
+import requests
+
+# Initialize Instaloader
+L = instaloader.Instaloader()
+
+def fetch_user_posts(username: str, max_posts: int = 50) -> List[Dict]:
+    """
+    Fetch posts from a given Instagram profile (public data only).
+    """
+    try:
+        profile = instaloader.Profile.from_username(L.context, username)
+        posts = []
+        for post in profile.get_posts():
+            posts.append({
+                "username": username,
+                "caption": post.caption,
+                "hashtags": post.caption_hashtags,
+                "likes": post.likes,
+                "comments": post.comments,
+                "date": post.date_utc.isoformat(),
+                "image_url": post.url
+            })
+            if len(posts) >= max_posts:  # Limit the number of posts
+                break
+        return posts
+    except Exception as e:
+        print(f"Error fetching posts for {username}: {e}")
+        return []
+
+def find_similar_accounts(username: str, rapidapi_key: str) -> List[str]:
+    """
+    Fetch similar accounts using the RapidAPI endpoint.
+    """
+    url = "https://instagram-scraper-api2.p.rapidapi.com/v1/similar_accounts"
+    querystring = {"username_or_id_or_url": username}
+
+    headers = {
+        "x-rapidapi-host": "instagram-scraper-api2.p.rapidapi.com",
+        "x-rapidapi-key": "d14b901fa8mshdafabd10d36f007p1ff602jsn91766325646f"
+    }
+
+    try:
+        response = requests.get(url, headers=headers, params=querystring)
+        response.raise_for_status()  # Raise an error for bad status codes
+        data = response.json()
+
+        # Extract similar accounts from the API response
+        if data.get("status") == "success":
+            return data.get("data", {}).get("similar_accounts", [])
+        else:
+            print(f"Error fetching similar accounts: {data.get('message')}")
+            return []
+    except requests.exceptions.RequestException as e:
+        print(f"API request failed: {e}")
+        return []
+
+def fetch_competitors_data(username: str, rapidapi_key: str, max_posts: int = 50) -> List[Dict]:
+    """
+    Fetch data for similar accounts (competitors) using the RapidAPI endpoint.
+    """
+    similar_accounts = find_similar_accounts(username, rapidapi_key)
+    all_posts = []
+    for account in similar_accounts:
+        print(f"Fetching posts for competitor: {account}")
+        competitor_posts = fetch_user_posts(account, max_posts)
+        all_posts.extend(competitor_posts)
+    return all_posts
+

utils/logging_utils.py

@@ -0,0 +1,5 @@
+import logging
+
+def setup_logging():
+    """Set up logging configuration."""
+    logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

utils/preprocessing.py

@@ -0,0 +1,43 @@
+import pandas as pd
+from textblob import TextBlob
+from sklearn.preprocessing import LabelEncoder
+import logging
+
+def preprocess_data(df):
+    """Preprocess the input DataFrame."""
+    # Ensure required columns exist
+    required_columns = ['likes', 'comments', 'shares', 'posting_time', 'caption', 'hashtags']
+    missing_columns = [col for col in required_columns if col not in df.columns]
+    
+    if missing_columns:
+        logging.warning(f"Missing required columns: {missing_columns}")
+        for col in missing_columns:
+            if col in ['likes', 'comments', 'shares']:
+                df[col] = 0  # Fill with default value (integer)
+            elif col == 'caption':
+                df[col] = ''  # Fill with default value (empty string)
+            elif col == 'hashtags':
+                df[col] = [[] for _ in range(len(df))]  # Fill with default value (list of empty lists)
+    
+    # Convert posting_time to datetime
+    df['posting_time'] = pd.to_datetime(df['posting_time'], format='%Y-%m-%d %H:%M:%S', errors='coerce')
+    df = df[df['posting_time'].notna()]
+    
+    # Calculate engagement rate
+    df['engagement_rate'] = df['likes'] + df['comments'] + df['shares']
+    
+    # Calculate caption length and hashtag count
+    df['caption_length'] = df['caption'].apply(len)
+    df['hashtag_count'] = df['hashtags'].apply(len)
+    
+    # Calculate sentiment
+    df['caption_sentiment'] = df['caption'].apply(lambda x: TextBlob(x).sentiment.polarity)
+    df['sentiment'] = df['caption_sentiment']
+    
+    # Encode categorical columns
+    if 'content_type' in df.columns and 'media_type' in df.columns:
+        label_encoder = LabelEncoder()
+        df['content_type_encoded'] = label_encoder.fit_transform(df['content_type'])
+        df['media_type_encoded'] = label_encoder.fit_transform(df['media_type'])
+    
+    return df

utils/visualization.py

@@ -0,0 +1,20 @@
+import matplotlib.pyplot as plt
+import seaborn as sns
+
+def plot_engagement_heatmap(engagement_by_hour):
+    """Plot engagement heatmap by time of day."""
+    plt.figure(figsize=(10, 6))
+    sns.heatmap(engagement_by_hour.pivot_table(index='hour', values='engagement_rate'), annot=True, cmap='YlGnBu')
+    plt.title('Engagement Heatmap by Time of Day')
+    plt.xlabel('Engagement Rate')
+    plt.ylabel('Hour of Day')
+    plt.show()
+
+def plot_engagement_over_time(engagement_summary):
+    """Plot engagement rate over time."""
+    plt.figure(figsize=(10, 6))
+    plt.plot(engagement_summary['posting_time'], engagement_summary['engagement_rate'])
+    plt.title('Engagement Rate Over Time')
+    plt.xlabel('Time')
+    plt.ylabel('Engagement Rate')
+    plt.show()