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)):,}" @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 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 influencer[key] = f"{int(influencer[key]):,}" # 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']) db.close() 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} return response_data if __name__ == "__main__": uvicorn.run(app, host="127.0.0.1", port=8000)