Spaces:
Sleeping
Sleeping
from fastapi import FastAPI, HTTPException, BackgroundTasks, Depends, Header, status, Query | |
from fastapi.middleware.cors import CORSMiddleware | |
from fastapi.security import APIKeyHeader | |
from pydantic import BaseModel, EmailStr, Field | |
from typing import Optional, Tuple | |
from enum import Enum | |
import os | |
import base64 | |
import pickle | |
import pandas as pd | |
from dotenv import load_dotenv | |
from langchain_openai import ChatOpenAI | |
from langchain.schema import HumanMessage, SystemMessage | |
from email.mime.text import MIMEText | |
from google.auth.transport.requests import Request | |
from google.oauth2.credentials import Credentials | |
from google_auth_oauthlib.flow import InstalledAppFlow | |
from googleapiclient.discovery import build | |
from googleapiclient.errors import HttpError | |
from datetime import datetime | |
import json | |
from langfuse import Langfuse | |
from langfuse.langchain import CallbackHandler | |
load_dotenv() | |
langfuse = Langfuse( | |
secret_key=os.getenv("SECRET_KEY"), | |
public_key=os.getenv("PUBLIC_KEY"), | |
host=os.getenv("APP_HOST") | |
) | |
langfuse_handler = CallbackHandler() | |
# Load environment variables (not needed on Hugging Face, but harmless) | |
# ------------------------------------------ | |
# Security Configuration | |
# ------------------------------------------ | |
API_PASSWORD = os.getenv("API_PASSWORD", "Synapse@2025") # Can be overridden by env var | |
api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False) | |
def verify_api_key(api_key: Optional[str] = Depends(api_key_header)) -> str: | |
""" | |
Verify the API key/password provided in the request header. | |
""" | |
if api_key is None: | |
raise HTTPException( | |
status_code=status.HTTP_401_UNAUTHORIZED, | |
detail="API Key required. Please provide X-API-Key header.", | |
headers={"WWW-Authenticate": "ApiKey"}, | |
) | |
if api_key != API_PASSWORD: | |
raise HTTPException( | |
status_code=status.HTTP_401_UNAUTHORIZED, | |
detail="Invalid API Key", | |
headers={"WWW-Authenticate": "ApiKey"}, | |
) | |
return api_key | |
# ------------------------------------------ | |
# Helper: Write GOOGLE_CREDENTIALS_JSON to file if needed | |
# ------------------------------------------ | |
def ensure_credentials_file(): | |
credentials_env = os.getenv("GOOGLE_CREDENTIALS_JSON") | |
credentials_path = "credentials_SYNAPSE.json" | |
if not os.path.exists(credentials_path): | |
if not credentials_env: | |
raise Exception("GOOGLE_CREDENTIALS_JSON not found in environment variables.") | |
try: | |
parsed_json = json.loads(credentials_env) | |
except json.JSONDecodeError: | |
raise Exception("Invalid JSON in GOOGLE_CREDENTIALS_JSON") | |
with open(credentials_path, "w") as f: | |
json.dump(parsed_json, f, indent=2) | |
return credentials_path | |
# ------------------------------------------ | |
# FastAPI app | |
# ------------------------------------------ | |
app = FastAPI(title="Recruitment Message Generator API", version="1.0.0") | |
app.add_middleware( | |
CORSMiddleware, | |
allow_origins=["*"], | |
allow_credentials=True, | |
allow_methods=["*"], | |
allow_headers=["*"], | |
) | |
SCOPES = ["https://www.googleapis.com/auth/gmail.send"] | |
openai_api_key = os.getenv("OPENAI_API_KEY") | |
# ------------------------------------------ | |
# Enums and Models | |
# ------------------------------------------ | |
class MessageType(str, Enum): | |
OUTREACH = "outreach" | |
INTRODUCTORY = "introductory" | |
FOLLOWUP = "followup" | |
class GenerateMessageRequest(BaseModel): | |
job_evaluation: Optional[str] = Field(None, description="(Optional) Recruiter's evaluation of candidate for the job") | |
sender_email: EmailStr | |
reply_to_email: Optional[EmailStr] = Field(None, description="Recruiter's email for reply-to header") | |
recipient_email: EmailStr | |
candidate_name: str | |
current_role: str | |
current_company: str | |
company_name: str | |
role: str | |
recruiter_name: str | |
organisation: str | |
message_type: MessageType | |
send_email: bool = False | |
past_conversation: Optional[str] = Field(None, description="(Optional) Previous messages with candidate") | |
job_location: Optional[str] = Field(None, description="(Optional) Job location (e.g., 'Sterling, VA')") | |
company_blurb: Optional[str] = Field(None, description="(Optional) Brief description of company mission/industry") | |
culture: Optional[str] = Field(None, description="(Optional) Company culture description (e.g., 'fast-moving, tight-knit team')") | |
job_responsibilities: Optional[str] = Field(None, description="(Optional) Key responsibilities and duties for the role") | |
candidate_experience: Optional[str] = Field(None, description="(Optional) Specific skills and experience of the candidate") | |
class FeedbackRequest(BaseModel): | |
message: str | |
feedback: str | |
class AuthenticateRequest(BaseModel): | |
email: EmailStr | |
class AuthenticateResponse(BaseModel): | |
success: bool | |
message: str | |
error: Optional[str] = None | |
class MessageResponse(BaseModel): | |
success: bool | |
message: str | |
email_sent: bool = False | |
email_subject: Optional[str] = None | |
error: Optional[str] = None | |
class SendMessageRequest(BaseModel): | |
subject: str | |
email_body: str | |
sender_email: EmailStr | |
recipient_email: EmailStr | |
reply_to_email: Optional[EmailStr] = None | |
# ------------------------------------------ | |
# Gmail Helper Functions | |
# ------------------------------------------ | |
def get_token_file_path(email: str) -> str: | |
tokens_dir = "gmail_tokens" | |
if not os.path.exists(tokens_dir): | |
os.makedirs(tokens_dir) | |
safe_email = email.replace("@", "_at_").replace(".", "_dot_") | |
return os.path.join(tokens_dir, f"token_{safe_email}.pickle") | |
def check_user_token_exists(email: str) -> bool: | |
token_file = get_token_file_path(email) | |
return os.path.exists(token_file) | |
def load_user_credentials(email: str): | |
token_file = get_token_file_path(email) | |
if os.path.exists(token_file): | |
try: | |
with open(token_file, 'rb') as token: | |
creds = pickle.load(token) | |
return creds | |
except Exception: | |
if os.path.exists(token_file): | |
os.remove(token_file) | |
return None | |
def save_user_credentials(email: str, creds): | |
token_file = get_token_file_path(email) | |
with open(token_file, 'wb') as token: | |
pickle.dump(creds, token) | |
def create_new_credentials(email: str): | |
credentials_path = ensure_credentials_file() | |
flow = InstalledAppFlow.from_client_secrets_file( | |
credentials_path, SCOPES | |
) | |
creds = flow.run_local_server(port=0) | |
save_user_credentials(email, creds) | |
return creds | |
def authenticate_gmail(email: str, create_if_missing: bool = False): | |
creds = load_user_credentials(email) | |
if creds: | |
if creds.expired and creds.refresh_token: | |
try: | |
creds.refresh(Request()) | |
save_user_credentials(email, creds) | |
except Exception: | |
if create_if_missing: | |
try: | |
creds = create_new_credentials(email) | |
except: | |
return None | |
else: | |
return None | |
elif not creds.valid: | |
creds = None | |
if not creds: | |
if create_if_missing: | |
try: | |
creds = create_new_credentials(email) | |
except: | |
return None | |
else: | |
return None | |
try: | |
service = build("gmail", "v1", credentials=creds) | |
return service | |
except Exception: | |
return None | |
from email.mime.text import MIMEText | |
def create_email_message(sender: str, to: str, subject: str, message_text: str, reply_to: Optional[str] = None, is_html: bool = False): | |
if is_html: | |
message = MIMEText(message_text, "html") | |
else: | |
message = MIMEText(message_text, "plain") | |
message["To"] = to | |
message["From"] = sender | |
message["Subject"] = subject | |
if reply_to: | |
message["Reply-To"] = reply_to | |
raw_message = base64.urlsafe_b64encode(message.as_bytes()).decode() | |
return {"raw": raw_message} | |
def send_gmail_message(service, user_id: str, message: dict): | |
try: | |
result = service.users().messages().send(userId=user_id, body=message).execute() | |
return result is not None | |
except HttpError: | |
return False | |
# ------------------------------------------ | |
# LLM (OpenAI) Message Generation Helpers | |
# ------------------------------------------ | |
def refine_message_with_feedback( | |
original_message: str, | |
feedback: str, | |
) -> Tuple[str, str]: | |
api_key = os.getenv("OPENAI_API_KEY") | |
llm = ChatOpenAI( | |
model="gpt-4o-mini", | |
temperature=0.7, | |
max_tokens=600, | |
openai_api_key=api_key | |
) | |
prompt = f""" | |
Please refine the following recruitment message based on the provided feedback: | |
ORIGINAL MESSAGE: | |
{original_message} | |
FEEDBACK: | |
{feedback} | |
Please provide your response in the following format: | |
SUBJECT: [Your subject line here] | |
BODY: | |
[Your refined email body content here] | |
Keep the same warm, conversational tone and intent as the original message, but incorporate the feedback to improve it. Make sure the message feels personal and approachable. | |
""" | |
try: | |
messages = [ | |
SystemMessage(content="You are a friendly and approachable recruitment specialist. Refine the given message based on feedback while maintaining a warm, conversational tone and the original intent."), | |
HumanMessage(content=prompt) | |
] | |
response = llm.invoke(messages, config={"callbacks": [langfuse_handler]}) | |
content = response.content.strip() | |
subject_line = "" | |
body_content = "" | |
lines = content.split('\n') | |
body_found = False | |
body_lines = [] | |
for line in lines: | |
line_stripped = line.strip() | |
# Handle both plain text and markdown formats | |
if line_stripped.startswith('SUBJECT:') or line_stripped.startswith('**SUBJECT:**'): | |
subject_line = line_stripped.replace('SUBJECT:', '').replace('**SUBJECT:**', '').replace('**', '').strip() | |
elif line_stripped.startswith('BODY:') or line_stripped.startswith('**BODY:**'): | |
body_found = True | |
elif body_found and line_stripped: | |
body_lines.append(line) | |
body_content = '\n'.join(body_lines).strip() | |
# Clean up any remaining markdown formatting from body | |
if body_content: | |
body_content = body_content.replace('**', '') | |
if not subject_line: | |
subject_line = "Recruitment Opportunity - Updated" | |
if not body_content: | |
body_content = content | |
return subject_line, body_content | |
except Exception as e: | |
raise HTTPException(status_code=500, detail=f"Error refining message: {str(e)}") | |
def generate_recruitment_message_with_subject( | |
msg_type: str, | |
company: str, | |
role_title: str, | |
recruiter: str, | |
org: str, | |
candidate: str, | |
current_pos: str, | |
evaluation: Optional[str] = None, | |
feedback: Optional[str] = None, | |
past_conversation: Optional[str] = None, | |
job_location: Optional[str] = None, | |
company_blurb: Optional[str] = None, | |
culture: Optional[str] = None, | |
job_responsibilities: Optional[str] = None, | |
candidate_experience: Optional[str] = None | |
) -> Tuple[str, str]: | |
api_key = os.getenv("OPENAI_API_KEY") | |
llm = ChatOpenAI( | |
model="gpt-4o-mini", | |
temperature=0.7, | |
max_tokens=600, | |
openai_api_key=api_key | |
) | |
# Outreach: Only request consent | |
if msg_type == "outreach": | |
prompt = f""" | |
Generate a warm and friendly initial outreach message to a candidate. | |
- Introduce yourself as {recruiter} from {org} in a casual, approachable way | |
- Mention you came across their profile and were impressed by their background | |
- Share that you're reaching out about an exciting opportunity ({role_title}) at {company} | |
- Ask if they'd be interested in learning more about this role | |
- Use a conversational tone - think of it as reaching out to a colleague or friend | |
- Do NOT discuss any job evaluation or judgment. | |
- Keep it concise but warm and engaging | |
- No placeholders like [Candidate Name] or [Role Title] in the output. | |
""" | |
else: | |
base_prompt = f""" | |
Generate a friendly and professional recruitment {msg_type} with the following details: | |
- Company hiring: {company} | |
- Role: {role_title} | |
- Recruiter: {recruiter} from {org} | |
- Candidate: {candidate} | |
- Candidate's current position: {current_pos} | |
""" | |
# Add optional additional details if provided | |
if job_location: | |
base_prompt += f"- Job location: {job_location}\n" | |
if company_blurb: | |
base_prompt += f"- Company description: {company_blurb}\n" | |
if culture: | |
base_prompt += f"- Company culture: {culture}\n" | |
if job_responsibilities: | |
base_prompt += f"- Job responsibilities: {job_responsibilities}\n" | |
if candidate_experience: | |
base_prompt += f"- Candidate's specific experience: {candidate_experience}\n" | |
if evaluation: | |
base_prompt += f"- Evaluation: {evaluation}\n" | |
prompt = base_prompt | |
if msg_type == "introductory": | |
prompt += """ | |
Create a friendly and engaging introductory message that: | |
- Thanks the candidate warmly for their response and interest | |
- Shares exciting details about the role and company culture | |
- Explains why you think this opportunity would be a great fit for their background | |
- Suggests next steps (like a casual chat or coffee meeting) in a relaxed way | |
- Uses a conversational, approachable tone while staying professional | |
- Shows genuine enthusiasm about the opportunity | |
- if in the Evaluation the verdict is rejected, Send a polite rejection message instead. | |
""" | |
else: # followup | |
prompt += """ | |
Create a friendly and follow-up message that: | |
- References your previous conversation in a warm, personal way | |
- Shows continued enthusiasm and interest in the candidate | |
- Addresses any concerns they might have in a supportive manner | |
- Shares additional exciting reasons why this role would be perfect for them | |
- Includes a relaxed but clear call to action | |
- Uses a conversational tone that feels like catching up with someone | |
""" | |
# Use feedback if provided | |
if feedback: | |
prompt += f"\n\nPlease modify the message based on this feedback: {feedback}" | |
# Use past conversation if provided | |
if past_conversation: | |
prompt += f""" | |
Use the following past conversation as context to inform your reply: | |
PAST CONVERSATION: | |
{past_conversation} | |
Write a reply message to the candidate, maintaining professionalism and continuity. | |
""" | |
prompt += """ | |
Please provide your response in the following format: | |
SUBJECT: [Your subject line here] | |
BODY: | |
[Your email body content here] | |
""" | |
try: | |
messages = [ | |
SystemMessage(content="You are a friendly and approachable recruitment specialist who writes warm, engaging messages. Generate both an email subject line and body content. Use a conversational tone that feels personal and genuine while maintaining professionalism. Follow the exact format requested."), | |
HumanMessage(content=prompt) | |
] | |
response = llm.invoke(messages, config={"callbacks": [langfuse_handler]}) | |
content = response.content.strip() | |
subject_line = "" | |
body_content = "" | |
lines = content.split('\n') | |
body_found = False | |
body_lines = [] | |
for line in lines: | |
line_stripped = line.strip() | |
# Handle both plain text and markdown formats | |
if line_stripped.startswith('SUBJECT:') or line_stripped.startswith('**SUBJECT:**'): | |
subject_line = line_stripped.replace('SUBJECT:', '').replace('**SUBJECT:**', '').replace('**', '').strip() | |
elif line_stripped.startswith('BODY:') or line_stripped.startswith('**BODY:**'): | |
body_found = True | |
elif body_found and line_stripped: | |
body_lines.append(line) | |
body_content = '\n'.join(body_lines).strip() | |
# Clean up any remaining markdown formatting from body | |
if body_content: | |
body_content = body_content.replace('**', '') | |
if not subject_line: | |
subject_line = f"Opportunity at {company} - {role_title}" | |
if not body_content: | |
body_content = content | |
return subject_line, body_content | |
except Exception as e: | |
raise HTTPException(status_code=500, detail=f"Error generating message: {str(e)}") | |
def format_email_html(body: str, sender_name: Optional[str]=None, sender_org: Optional[str]=None): | |
"""Wrap plain text body in an HTML template for better appearance.""" | |
import re | |
# Smart paragraphing and formatting | |
body = body.strip() | |
# Convert markdown-style formatting | |
body = re.sub(r'\*\*(.*?)\*\*', r'<strong>\1</strong>', body) # Bold text | |
body = re.sub(r'\*(.*?)\*', r'<em>\1</em>', body) # Italic text | |
# Smart paragraphing | |
body = re.sub(r"\n\s*\n", "</p><p>", body) # Double newlines => new paragraph | |
body = re.sub(r"\n", "<br>", body) # Single newlines => <br> | |
# Optional signature with better styling | |
signature = "" | |
if sender_name or sender_org: | |
signature = f""" | |
<div style="margin-top: 30px; padding-top: 20px; border-top: 1px solid #e0e0e0;"> | |
<p style="margin: 0; color: #666; font-size: 14px;">Best regards,</p> | |
""" | |
if sender_name: | |
signature += f'<p style="margin: 5px 0 0 0; font-weight: 600; color: #333; font-size: 16px;">{sender_name}</p>' | |
if sender_org: | |
signature += f'<p style="margin: 3px 0 0 0; color: #666; font-size: 14px;">{sender_org}</p>' | |
signature += "</div>" | |
html = f""" | |
<!DOCTYPE html> | |
<html> | |
<head> | |
<meta charset="UTF-8"> | |
<meta name="viewport" content="width=device-width, initial-scale=1.0"> | |
<title>Recruitment Message</title> | |
</head> | |
<body style=" | |
font-family: 'Segoe UI', Tahoma, Geneva, Verdana, sans-serif; | |
color: #333333; | |
line-height: 1.6; | |
max-width: 600px; | |
margin: 0 auto; | |
padding: 20px; | |
background-color: #f8f9fa; | |
"> | |
<div style=" | |
background-color: #ffffff; | |
border-radius: 8px; | |
box-shadow: 0 2px 10px rgba(0, 0, 0, 0.1); | |
padding: 30px; | |
border: 1px solid #e9ecef; | |
"> | |
<!-- Header --> | |
<div style=" | |
border-bottom: 2px solid #007bff; | |
padding-bottom: 15px; | |
margin-bottom: 25px; | |
"> | |
<h2 style=" | |
margin: 0; | |
color: #007bff; | |
font-size: 20px; | |
font-weight: 600; | |
">Professional Recruitment Message</h2> | |
</div> | |
<!-- Main Content --> | |
<div style=" | |
font-size: 15px; | |
color: #444; | |
line-height: 1.7; | |
"> | |
<p style="margin: 0 0 15px 0;">{body}</p> | |
</div> | |
<!-- Signature --> | |
{signature} | |
<!-- Footer --> | |
<div style=" | |
margin-top: 30px; | |
padding-top: 15px; | |
border-top: 1px solid #e0e0e0; | |
text-align: center; | |
font-size: 12px; | |
color: #888; | |
"> | |
<p style="margin: 0;">This message was generated by Synapse Recruitment Platform</p> | |
<p style="margin: 5px 0 0 0;">Please reply to this email for any inquiries</p> | |
</div> | |
</div> | |
</body> | |
</html> | |
""" | |
return html | |
# ------------------------------------------ | |
# FastAPI Endpoints | |
# ------------------------------------------ | |
async def root(): | |
"""Root endpoint - public access""" | |
return { | |
"message": "Recruitment Message Generator API", | |
"version": "1.0.0", | |
"authentication": "Required for all endpoints except / and /health", | |
"auth_header": "X-API-Key", | |
"endpoints": [ | |
"/generate-message", | |
"/refine-message", | |
"/authenticate", | |
"/send-message", | |
"/docs" | |
] | |
} | |
async def send_message( | |
subject: str = Query(..., description="Email subject"), | |
email_body: str = Query(..., description="Email body content"), | |
sender_email: str = Query(..., description="Sender's email address"), | |
recipient_email: str = Query(..., description="Recipient's email address"), | |
reply_to_email: Optional[str] = Query(None, description="Reply-to email address") | |
): | |
try: | |
# Authenticate sender | |
service = authenticate_gmail(sender_email) | |
if not service: | |
return MessageResponse( | |
success=False, | |
message="", | |
error="Gmail authentication failed" | |
) | |
formatted_html = format_email_html( | |
email_body | |
) | |
# Create the email | |
email_message = create_email_message( | |
sender=sender_email, | |
to=recipient_email, | |
subject=subject, | |
message_text=formatted_html, | |
reply_to=reply_to_email, | |
is_html=True | |
) | |
# Send the email | |
email_sent = send_gmail_message(service, "me", email_message) | |
if email_sent: | |
return MessageResponse( | |
success=True, | |
message="Email sent successfully.", | |
email_sent=True, | |
email_subject=subject | |
) | |
else: | |
return MessageResponse( | |
success=False, | |
message="", | |
email_sent=False, | |
email_subject=subject, | |
error="Failed to send via Gmail API" | |
) | |
except Exception as e: | |
return MessageResponse( | |
success=False, | |
message="", | |
error=str(e) | |
) | |
async def generate_message( | |
background_tasks: BackgroundTasks, | |
sender_email: str = Query(..., description="Sender's email address"), | |
recipient_email: str = Query(..., description="Recipient's email address"), | |
candidate_name: str = Query(..., description="Candidate's name"), | |
current_role: str = Query(..., description="Candidate's current role"), | |
current_company: str = Query(..., description="Candidate's current company"), | |
company_name: str = Query(..., description="Company name for the job"), | |
role: str = Query(..., description="Job role title"), | |
recruiter_name: str = Query(..., description="Recruiter's name"), | |
organisation: str = Query(..., description="Recruiting organisation"), | |
message_type: MessageType = Query(..., description="Type of message"), | |
job_evaluation: Optional[str] = Query(None, description="Recruiter's evaluation of candidate for the job"), | |
reply_to_email: Optional[str] = Query(None, description="Recruiter's email for reply-to header"), | |
send_email: bool = Query(False, description="Whether to send the email"), | |
past_conversation: Optional[str] = Query(None, description="Previous messages with candidate"), | |
job_location: Optional[str] = Query(None, description="Job location (e.g., 'Sterling, VA')"), | |
company_blurb: Optional[str] = Query(None, description="Brief description of company mission/industry"), | |
culture: Optional[str] = Query(None, description="Company culture description (e.g., 'fast-moving, tight-knit team')"), | |
job_responsibilities: Optional[str] = Query(None, description="Key responsibilities and duties for the role"), | |
candidate_experience: Optional[str] = Query(None, description="Specific skills and experience of the candidate") | |
): | |
try: | |
current_position = f"{current_role} at {current_company}" | |
email_subject, generated_message = generate_recruitment_message_with_subject( | |
msg_type=message_type.value.replace('followup', 'follow-up'), | |
company=company_name, | |
role_title=role, | |
recruiter=recruiter_name, | |
org=organisation, | |
candidate=candidate_name, | |
current_pos=current_position, | |
evaluation=job_evaluation, | |
past_conversation=past_conversation, | |
job_location=job_location, | |
company_blurb=company_blurb, | |
culture=culture, | |
job_responsibilities=job_responsibilities, | |
candidate_experience=candidate_experience | |
) | |
email_sent = False | |
if send_email: | |
registered_users = [] | |
if os.path.exists("registered_users.csv"): | |
df = pd.read_csv("registered_users.csv") | |
registered_users = df['email'].tolist() if 'email' in df.columns else [] | |
if sender_email.lower() not in [user.lower() for user in registered_users]: | |
return MessageResponse( | |
success=True, | |
message=generated_message, | |
email_sent=False, | |
email_subject=email_subject, | |
error="Email not sent: Sender email is not registered" | |
) | |
service = authenticate_gmail(sender_email) | |
if service: | |
formatted_html = format_email_html( | |
generated_message, | |
sender_name=recruiter_name, | |
sender_org=organisation | |
) | |
email_message = create_email_message( | |
sender=sender_email, | |
to=recipient_email, | |
subject=email_subject, | |
message_text=formatted_html, | |
reply_to=reply_to_email, | |
is_html=True | |
) | |
email_sent = send_gmail_message(service, "me", email_message) | |
if not email_sent: | |
return MessageResponse( | |
success=True, | |
message=generated_message, | |
email_sent=False, | |
email_subject=email_subject, | |
error="Email not sent: Failed to send via Gmail API" | |
) | |
else: | |
return MessageResponse( | |
success=True, | |
message=generated_message, | |
email_sent=False, | |
email_subject=email_subject, | |
error="Email not sent: Gmail authentication failed" | |
) | |
return MessageResponse( | |
success=True, | |
message=generated_message, | |
email_sent=email_sent, | |
email_subject=email_subject | |
) | |
except Exception as e: | |
return MessageResponse( | |
success=False, | |
message="", | |
error=str(e) | |
) | |
async def refine_message(request: FeedbackRequest): | |
try: | |
email_subject, refined_message = refine_message_with_feedback( | |
original_message=request.message, | |
feedback=request.feedback | |
) | |
return MessageResponse( | |
success=True, | |
message=refined_message, | |
email_sent=False, | |
email_subject=email_subject | |
) | |
except Exception as e: | |
return MessageResponse( | |
success=False, | |
message="", | |
error=str(e) | |
) | |
async def authenticate_user(request: AuthenticateRequest): | |
try: | |
if check_user_token_exists(request.email): | |
service = authenticate_gmail(request.email, create_if_missing=False) | |
if service: | |
return AuthenticateResponse( | |
success=True, | |
message="User already authenticated and token is valid" | |
) | |
else: | |
service = authenticate_gmail(request.email, create_if_missing=True) | |
if service: | |
return AuthenticateResponse( | |
success=True, | |
message="Token refreshed successfully" | |
) | |
else: | |
return AuthenticateResponse( | |
success=False, | |
message="Failed to refresh token", | |
error="Could not refresh existing token. Please check credentials.json" | |
) | |
else: | |
try: | |
creds = create_new_credentials(request.email) | |
if creds: | |
return AuthenticateResponse( | |
success=True, | |
message="Authentication successful. Token created and saved." | |
) | |
else: | |
return AuthenticateResponse( | |
success=False, | |
message="Authentication failed", | |
error="Failed to create credentials" | |
) | |
except Exception as e: | |
return AuthenticateResponse( | |
success=False, | |
message="Authentication failed", | |
error=str(e) | |
) | |
except Exception as e: | |
return AuthenticateResponse( | |
success=False, | |
message="Authentication error", | |
error=str(e) | |
) | |
async def health_check(): | |
"""Health check endpoint - public access""" | |
return {"status": "healthy", "timestamp": datetime.utcnow().isoformat()} | |
# Protected documentation endpoint | |
async def get_documentation(): | |
"""This will be handled by FastAPI's built-in docs""" | |
pass | |
if __name__ == "__main__": | |
import uvicorn | |
uvicorn.run(app, host="0.0.0.0", port=8000) | |
# Additional Infor: Job Location, Company Blurb, Culture, Job Responsibilities, Candidate Experience |