Finalized File List:
1. main.py – Central FastAPI app integrating all modules.
2. database.py – Stores user data, chat records, and compatibility scores.
3. matchmaking.py – Handles user matching based on personality profiles and chat
history.
4. chat.py – Manages direct messages, group chats, and chat summaries.
5. user_management.py – Handles user actions like blocking, reporting, and
archiving chats.
6. utils.py – Contains helper functions like username generation.
7. user_onboarding.py – Manages user registration, onboarding, and profile setup.
8. ai_chatbot.py – Core AI chatbot logic for engaging, personality-driven responses.
9. next_question_prompting.py – Suggests next questions to keep users
engaged.
10. social_sharing.py – Enables sharing insights via social media.
11. push_notifications.py – Sends personalized notifications to users.
12. requirements.txt – Lists all necessary dependencies.
Main.py
from fastapi import FastAPI from chat import chat_router from matchmaking import
matchmaking_router from user_management import user_router from user_onboarding
import onboarding_router from ai_chatbot import chatbot_router from
next_question_prompting import question_prompt_router from social_sharing import
social_sharing_router from push_notifications import notifications_router
app = FastAPI()
Include different modules as routers
app.include_router(chat_router, prefix="/chat") app.include_router(matchmaking_router,
prefix="/matchmaking") app.include_router(user_router, prefix="/user")
app.include_router(onboarding_router, prefix="/onboarding")
app.include_router(chatbot_router, prefix="/ai_chat")
app.include_router(question_prompt_router, prefix="/next_question")
app.include_router(social_sharing_router, prefix="/social")
app.include_router(notifications_router, prefix="/notifications")
@app.get("/") def root(): return {"message": "Welcome to SnugHub AI Chatbot!"}
Dataase.py
Mock database for user profiles and chats
user_profiles = {} conversation_streaks = {} compatibility_scores = {} archived_chats = {}
blocked_users = {} chat_rooms = {}
Matchmaking.py
from fastapi import APIRouter from database import user_profiles, chat_rooms import
random
def calculate_compatibility(user1: str, user2: str) -> int: if not user1 or not user2 or user1
== user2: return 0
user1_data = user_profiles.get(user1, {})
user2_data = user_profiles.get(user2, {})
score = 0
if user1_data.get("zodiac") == user2_data.get("zodiac"):
score += 20
if user1_data.get("mbti") == user2_data.get("mbti"):
score += 30
chat_history = chat_rooms.get(f"room-{user1}-{user2}",
{}).get("messages", [])
score += len(chat_history) * 2 # Increase score based on chat
frequency
return score
matchmaking_router = APIRouter()
@matchmaking_router.get("/") def matchmaking(email: str): user_data =
user_profiles.get(email, {}) potential_matches = [ { "username": data.get("username",
"MysticUser"), "zodiac": data.get("zodiac", "Unknown"), "mbti": data.get("mbti",
"Unknown"), "compatibility_score": calculate_compatibility(email, data.get("email")) } for
_, data in user_profiles.items() ]
sorted_matches = sorted(potential_matches, key=lambda x:
x["compatibility_score"], reverse=True)
return {"message": "Match found!", "matched_users":
sorted_matches[:5]}
Chat.py
from fastapi import APIRouter from database import chat_rooms import datetime
chat_router = APIRouter()
def create_chat_room(users: list): room_id = f"room-{random.randint(1000, 9999)}"
chat_rooms[room_id] = {"users": users, "messages": []} return room_id
@chat_router.post("/start_dm/") def start_dm(user1: str, user2: str): room_id =
create_chat_room([user1, user2]) return {"message": "Private chat started!", "room_id":
room_id}
@chat_router.post("/start_group_chat/") def start_group_chat(users: list): room_id =
create_chat_room(users) return {"message": "Group chat created!", "room_id": room_id}
@chat_router.post("/send_message/") def send_message(room_id: str, sender: str,
message: str): if room_id not in chat_rooms: return {"message": "Invalid chat room."}
chat_rooms[room_id]["messages"].append({"sender": sender, "message": message,
"timestamp": datetime.datetime.utcnow().isoformat(), "read": False, "reactions": {},
"typing_status": ""}) return {"message": "Message sent!"}
@chat_router.get("/chat_summary/") def chat_summary(email: str, room_id: str):
messages = chat_rooms.get(room_id, {}).get("messages", []) if not messages: return
{"message": "No messages found in this chat."} summary =
generate_chat_summary(messages) return {"summary": summary}
def generate_chat_summary(messages): return f"This conversation covered topics like {',
'.join(set([msg['message'][:10] for msg in messages]))} and ended on a positive note."
User management.py
from fastapi import APIRouter from database import user_profiles, archived_chats,
blocked_users
user_router = APIRouter()
@user_router.post("/archive_chat/") def archive_chat(email: str, room_id: str):
archived_chats[email] = archived_chats.get(email, []) + [room_id] return {"message": "Chat
archived successfully."}
@user_router.post("/block_user/") def block_user(email: str, blocked_email: str):
blocked_users[email] = blocked_users.get(email, []) + [blocked_email] return {"message":
f"User {blocked_email} blocked successfully."}
@user_router.post("/report_user/") def report_user(email: str, reported_email: str, reason:
str): return {"message": f"User {reported_email} has been reported for: {reason}. Our team
will review the report."}
Utils.py
import random
def generate_username():
adjectives = ["Mystic", "Radiant", "Fearless", "Enchanted", "Serene", "Majestic",
"Ethereal", "Brilliant", "Luminous"]
animals = ["Phoenix", "Unicorn", "Tiger", "Wolf", "Dragon", "Owl", "Butterfly", "Hawk",
"Dolphin"]
return f"{random.choice(adjectives)}{random.choice(animals)}"
User Onboarding.py
from fastapi import APIRouter, HTTPException from database import user_profiles import
firebase_admin from firebase_admin import auth
onboarding_router = APIRouter()
@onboarding_router.post("/register/") def register_user(email: str, password: str): try: user
= auth.create_user(email=email, password=password) user_profiles[email] = {"email":
email, "personality_type": None, "mbti": None, "zodiac": None} return {"message": "User
registered successfully!", "user_id": user.uid} except Exception as e: raise
HTTPException(status_code=400, detail=str(e))
@onboarding_router.post("/set_profile/") def set_user_profile(email: str, zodiac: str =
None, mbti: str = None): if email not in user_profiles: raise
HTTPException(status_code=404, detail="User not found")
user_profiles[email]["zodiac"] = zodiac if zodiac else
user_profiles[email]["zodiac"]
user_profiles[email]["mbti"] = mbti if mbti else
user_profiles[email]["mbti"]
return {"message": "User profile updated successfully!"}
User Management.py
from fastapi import APIRouter from database import user_profiles, archived_chats,
blocked_users
user_router = APIRouter()
@user_router.post("/archive_chat/") def archive_chat(email: str, room_id: str):
archived_chats[email] = archived_chats.get(email, []) + [room_id] return {"message": "Chat
archived successfully."}
@user_router.post("/block_user/") def block_user(email: str, blocked_email: str):
blocked_users[email] = blocked_users.get(email, []) + [blocked_email] return {"message":
f"User {blocked_email} blocked successfully."}
@user_router.post("/report_user/") def report_user(email: str, reported_email: str, reason:
str): return {"message": f"User {reported_email} has been reported for: {reason}. Our team
will review the report."}
AI chatbot.py
from fastapi import APIRouter from database import user_profiles import openai
chatbot_router = APIRouter()
@chatbot_router.post("/chat/") def chat_with_bot(email: str, message: str): if email not in
user_profiles: return {"message": "User profile not found. Please complete onboarding."}
user_data = user_profiles[email]
context = f"Personality Type: {user_data.get('personality_type',
'Unknown')}, MBTI: {user_data.get('mbti', 'Unknown')}, Zodiac:
{user_data.get('zodiac', 'Unknown')}"
prompt = f"User: {message}\nBot ({context}):"
response = openai.Completion.create(
engine="davinci",
prompt=prompt,
max_tokens=150
)
return {"message": response["choices"][0]["text"].strip()}
Next Question Prompting.py
from fastapi import APIRouter import random
question_prompt_router = APIRouter()
next_question_prompts = [ "Have you ever noticed a pattern in your relationships?", "Do
you think this has affected your career choices too?", "Want to know what kind of energy
you attract the most?", "How do you handle conflicts in relationships?", "Do you think your
personality type influences your friendships?" ]
@question_prompt_router.get("/") def get_next_question(): return {"next_question":
random.choice(next_question_prompts)}
Social Sharing.py
from fastapi import APIRouter
social_sharing_router = APIRouter()
@social_sharing_router.post("/share/") def share_insight(email: str, insight: str):
shareable_text = f"I just got this freakishly accurate reading on SnugHub! \n'{insight}'\nTry it
yourself: [link]" return {"shareable_text": shareable_text}
Push Notifications.py
from fastapi import APIRouter import firebase_admin from firebase_admin import
messaging from database import user_profiles
notifications_router = APIRouter()
@notifications_router.post("/send_notification/") def send_notification(email: str,
message: str): if email not in user_profiles: return {"message": "User not found."}
notification = messaging.Message(
notification=messaging.Notification(
title="A new insight for you!",
body=message
),
token=user_profiles[email].get("fcm_token", "")
)
response = messaging.send(notification)
return {"message": "Notification sent!", "response": response}
Requirements.py
fastapi
uvicorn
firebase-admin
openai
