import configparser import pymysql import pymysql.cursors import bcrypt import jwt import datetime from fastapi import FastAPI, Depends, HTTPException, status from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm import uvicorn from fastapi.middleware.cors import CORSMiddleware config = configparser.ConfigParser() config.read("config.ini") DB_HOST = config["DATABASE"]["MYSQL_HOST"] DB_PORT = int(config["DATABASE"]["MYSQL_PORT"]) DB_USER = config["DATABASE"]["MYSQL_USER"] DB_PASSWORD = config["DATABASE"]["MYSQL_PASSWORD"] DB_NAME = config["DATABASE"]["MYSQL_DB"] SECRET_KEY = "ef7bd031dc150392a94e4aae97ee3e25cafa7ae0b6080964a91bba445d3bd8ac" ALGORITHM = "HS256" app = FastAPI() app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"],allow_headers=["*"]) # OAuth2 scheme for authentication oauth2_scheme = OAuth2PasswordBearer(tokenUrl="login") # Database Connection def get_db(): return pymysql.connect( host=DB_HOST, port=DB_PORT, user=DB_USER, password=DB_PASSWORD, database=DB_NAME, cursorclass=pymysql.cursors.DictCursor ) def verify_token(token: str): try: payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) return payload except jwt.ExpiredSignatureError: raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Token expired") except jwt.InvalidTokenError: raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token") def clean_and_format(value): if isinstance(value, str): value = value.replace(",", "") # Remove commas if present return f"{int(float(value)):,}" def parse_number(value): if isinstance(value, (int, float)): return value try: value = str(value).lower().strip() if value.endswith('k'): return float(value[:-1]) * 1_000 elif value.endswith('m'): return float(value[:-1]) * 1_000_000 elif value.replace('.', '', 1).isdigit(): return float(value) except: pass return 0 @app.post("/login") def login(form_data: OAuth2PasswordRequestForm = Depends()): db = get_db() cursor = db.cursor() cursor.execute("SELECT * FROM users WHERE UserEmail=%s", (form_data.username,)) user = cursor.fetchone() if not user or not bcrypt.checkpw(form_data.password.encode(), user["UserPass"].encode()): raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid credentials") cursor.execute("SELECT * FROM brands WHERE UserId=%s", (user["UserId"],)) brand = cursor.fetchone() if not brand: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Brand not found") token_payload = {"brand_id": brand["BrandId"], "exp": datetime.datetime.utcnow() + datetime.timedelta(hours=1)} token = jwt.encode(token_payload, SECRET_KEY, algorithm=ALGORITHM) response_data = { "code": 200, "status": "Success", "data" : {"access_token": token, "token_type": "bearer","expire_in": 3600,"user_type":"super_admin","brand_name": brand['BrandName']}} return response_data @app.get("/brand/campaigns") def list_campaigns(token: str = Depends(oauth2_scheme)): brand = verify_token(token) brand_id = brand["brand_id"] db = get_db() cursor = db.cursor() cursor.execute(""" SELECT COUNT(DISTINCT ci.CampaignInfoId) AS total_campaigns, IFNULL(SUM(ci.Budget), 0) AS total_budget, IFNULL(SUM(cfb.DepositAmount), 0) AS total_paid, (IFNULL(SUM(ci.Budget), 0) - IFNULL(SUM(cfb.DepositAmount), 0)) AS pending_amount FROM campaigninfo ci INNER JOIN cashflowbrand cfb ON ci.CampaignInfoId = cfb.CampaignInfoId WHERE ci.BrandId = %s """, (brand_id,)) result = cursor.fetchone() formatted_result = { "total_campaigns": result["total_campaigns"], "total_budget": f"{result['total_budget']:,.0f}", "total_paid": f"{result['total_paid']:,.0f}", "pending_amount": f"{result['pending_amount']:,.0f}" } cursor.execute(""" SELECT CampaignInfoId as CamapignId, CampaignName, Budget, StartDate, EndDate,InfluencerCategory,CampaignType, Audience,SocialMediaPlatforms FROM campaigninfo WHERE BrandId=%s """, (brand_id,)) campaigns = cursor.fetchall() db.close() for campaign in campaigns: if isinstance(campaign["Budget"], str): campaign["Budget"] = int(campaign["Budget"]) # Convert string to int campaign["Budget"] = f"{campaign['Budget']:,.0f}" response_data = { "code": 200, "status": "Success", "data": {"kpis":formatted_result,"campaigns": campaigns}} return response_data @app.get("/brand/campaigns/{campaign_id}") def get_campaign_details(campaign_id: int, token: str = Depends(oauth2_scheme)): brand = verify_token(token) brand_id = brand["brand_id"] db = get_db() cursor = db.cursor() # Validate campaign ownership cursor.execute("SELECT * FROM campaigninfo WHERE CampaignInfoId=%s AND BrandId=%s", (campaign_id, brand_id)) campaign = cursor.fetchone() if not campaign: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Campaign not found or unauthorized") if isinstance(campaign["Budget"], str): campaign["Budget"] = int(campaign["Budget"]) # Convert string to int campaign["Budget"] = f"{campaign['Budget']:,.0f}" # Get influencers for this campaign cursor.execute(""" SELECT inf.InfluencerID, CONCAT(inf.FirstName, ' ', inf.LastName) AS 'Name', inf.Category,inf.Profilepic, IFNULL(NULLIF(infs.YouTubeFollowers, ''), 0) AS YouTubeFollowers, IFNULL(NULLIF(infs.FacebookFollowers, ''), 0) AS FacebookFollowers, IFNULL(NULLIF(infs.InstagramFollowers, ''), 0) AS InstagramFollowers, IFNULL(NULLIF(infs.TiktokFollowers, ''), 0) AS TiktokFollowers, IFNULL(NULLIF(infs.LinkedInFollowers, ''), 0) AS LinkedInFollowers, IFNULL(NULLIF(infs.TwitterFollowers, ''), 0) AS TwitterFollowers FROM campiagnexecution ce INNER JOIN influencers inf ON inf.InfluencerID = ce.InfluencerID INNER JOIN influencersocialmedia infs ON infs.InfluencerID = inf.InfluencerID WHERE ce.CampaignInfoId = %s """, (campaign_id,)) influencers = cursor.fetchall() for influencer in influencers: for key in ["YouTubeFollowers", "FacebookFollowers", "InstagramFollowers", "TiktokFollowers", "LinkedInFollowers", "TwitterFollowers"]: if isinstance(influencer[key], str): # If the value is a string influencer[key] = influencer[key].replace(",", "") # Remove commas value = parse_number(influencer[key]) influencer[key] = f"{int(value):,}" # Get financials cursor.execute(""" SELECT IFNULL(SUM(ci.Budget), 0) AS total_budget, IFNULL(SUM(cfb.DepositAmount), 0) AS total_paid, (IFNULL(SUM(ci.Budget), 0) - IFNULL(SUM(cfb.DepositAmount), 0)) AS pending_amount FROM campaigninfo ci INNER JOIN cashflowbrand cfb ON ci.CampaignInfoId = cfb.CampaignInfoId WHERE ci.BrandId = %s AND ci.CampaignInfoId = %s """, (brand_id,campaign_id)) financials = cursor.fetchone() financials['total_budget'] = clean_and_format(financials['total_budget']) financials['total_paid'] = clean_and_format(financials['total_paid']) financials['pending_amount'] = clean_and_format(financials['pending_amount']) # Get campaign details and Instagram URLs for each influencer influencer_posts = [] for influencer in influencers: influencer_id = influencer["InfluencerID"] # Get campaign details and Instagram URLs cursor.execute(""" SELECT CampaignDetailId, InstagramUrl FROM campaigndetail cd WHERE CampaignInfoId = %s AND InfluencerID = %s """, (campaign_id, influencer_id)) campaign_detail = cursor.fetchone() if campaign_detail and campaign_detail["InstagramUrl"]: campaign_detail_id = campaign_detail["CampaignDetailId"] instagram_urls = campaign_detail["InstagramUrl"].split(",") if campaign_detail["InstagramUrl"] else [] # Get insights for the campaign detail cursor.execute(""" SELECT ins.Reactions, ins.Comments, ins.Shares, ins.Views FROM insights ins WHERE ins.CampaignDetailId = %s """, (campaign_detail_id,)) insights = cursor.fetchall() # Prepare post data by combining URLs with insights posts_data = [] for i, url in enumerate(instagram_urls): post_data = { "link": url.strip(), "reactions": insights[i]["Reactions"] if i < len(insights) else 0, "comments": insights[i]["Comments"] if i < len(insights) else 0, "shares": insights[i]["Shares"] if i < len(insights) else 0, "views": insights[i]["Views"] if i < len(insights) else 0 } posts_data.append(post_data) # Add posts data to influencer influencer["campaign_detail_id"] = campaign_detail_id influencer["posts"] = posts_data data = { "campaign_name": campaign["CampaignName"], "campaign_type": campaign["CampaignType"], "total_budget": campaign["Budget"], "amount_paid": financials["total_paid"], "amount_pending": financials["pending_amount"], "influencers": influencers } response_data = { "code": 200, "status": "Success", "data": data} db.close() return response_data if __name__ == "__main__": uvicorn.run(app, host="127.0.0.1", port=8000)