Spaces:
Sleeping
Sleeping
from fastapi import FastAPI, File, UploadFile, Form, HTTPException, Request, Depends | |
from fastapi.staticfiles import StaticFiles | |
from fastapi.responses import HTMLResponse | |
from fastapi.responses import JSONResponse | |
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials | |
from pydantic import BaseModel, HttpUrl | |
from typing import Optional | |
import uvicorn | |
from fastapi.middleware.cors import CORSMiddleware | |
from utils import read_file, fetch_job_description, optimize_resume_api | |
import logging | |
import os | |
import json | |
from fastapi.responses import JSONResponse | |
API_SECRET = os.getenv("API_SECRET") | |
# Set up logging | |
logging.basicConfig(level=logging.DEBUG) | |
logger = logging.getLogger(__name__) | |
app = FastAPI(title="Resume Optimizer API") | |
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"]) | |
security = HTTPBearer() | |
class OptimizationResponse(BaseModel): | |
optimized_resume: str | |
changes_made: str | |
metadata: dict | |
class ErrorResponse(BaseModel): | |
detail: str | |
def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)): | |
token = credentials.credentials | |
if token != API_SECRET: | |
raise HTTPException(status_code=401, detail="Invalid authentication token") | |
return token | |
import re | |
def extract_xml_content(text, tag): | |
pattern = f"<{tag}>(.*?)</{tag}>" | |
matches = re.findall(pattern, text, re.DOTALL) | |
return matches | |
async def optimize_resume( | |
resume: Optional[UploadFile] = File(None), | |
resumeText: Optional[str] = Form(None), | |
jobDescription: Optional[str] = Form(None), | |
jobDescriptionUrl: Optional[HttpUrl] = Form(None), | |
token: str = Depends(verify_token) | |
): | |
try: | |
# Input validation | |
if (resume is None) == (resumeText is None): | |
raise ValueError("Provide either resume file or resume text, but not both") | |
if (jobDescription is None) == (jobDescriptionUrl is None): | |
raise ValueError("Provide either job description text or URL, but not both") | |
# Process resume | |
if resume: | |
resume_content = read_file(resume) | |
else: | |
resume_content = resumeText | |
# Process job description | |
if jobDescription: | |
job_description_content = jobDescription | |
else: | |
job_description_content = fetch_job_description(jobDescriptionUrl) | |
# Perform optimization | |
output = optimize_resume_api(resume_content, job_description_content) | |
optimized_resume = extract_xml_content(output, "optimized_resume") | |
changes_made = extract_xml_content(output, "changes_made") | |
# Prepare the response | |
response_data = OptimizationResponse( | |
optimized_resume=optimized_resume[0] if optimized_resume else "", | |
changes_made=changes_made[0] if changes_made else "", | |
metadata={ | |
"original_resume": resume_content, | |
"original_jd": job_description_content | |
} | |
) | |
return response_data | |
except ValueError as ve: | |
logger.error(f"ValueError occurred: {str(ve)}", exc_info=True) | |
raise HTTPException(status_code=400, detail=str(ve)) | |
except Exception as e: | |
logger.error(f"Unexpected error occurred: {str(e)}", exc_info=True) | |
raise HTTPException(status_code=500, detail=f"An unexpected error occurred: {str(e)}") | |
# Mount the static directory | |
app.mount("/static", StaticFiles(directory="static"), name="static") | |
# Add a route for serving the HTML file | |
async def serve_landing_page(): | |
with open("static/landing-page.html", "r") as file: | |
content = file.read() | |
return HTMLResponse(content=content) | |
# Add a route for serving the HTML file | |
async def serve_resume_optimizer(): | |
with open("static/resume-optimizer.html", "r") as file: | |
content = file.read() | |
return HTMLResponse(content=content) | |
async def serve_resume_optimizer(): | |
with open("static/followup-agent.html", "r") as file: | |
content = file.read() | |
return HTMLResponse(content=content) | |
async def serve_resume_optimizer(): | |
with open("static/digiyatra.html", "r") as file: | |
content = file.read() | |
return HTMLResponse(content=content) | |
async def serve_resume_optimizer(): | |
with open("static/auth.html", "r") as file: | |
content = file.read() | |
return HTMLResponse(content=content) | |
async def serve_resume_optimizer(): | |
with open("static/auth-success.html", "r") as file: | |
content = file.read() | |
return HTMLResponse(content=content) | |
async def serve_resume_optimizer(): | |
with open("static/v0/app-selector.html", "r") as file: | |
content = file.read() | |
return HTMLResponse(content=content) | |
async def serve_resume_optimizer(): | |
with open("static/multi-agent.html", "r") as file: | |
content = file.read() | |
return HTMLResponse(content=content) | |
async def serve_resume_optimizer(): | |
with open("static/medium-report.html", "r") as file: | |
content = file.read() | |
return HTMLResponse(content=content) | |
async def serve_iresearcher(): | |
with open("static/iresearcher.html", "r") as file: | |
content = file.read() | |
return HTMLResponse(content=content) | |
async def serve_iresearcher(): | |
with open("static/graph-maker.html", "r") as file: | |
content = file.read() | |
return HTMLResponse(content=content) | |
async def serve_indic_tts(): | |
with open("static/indic-text-to-speech.html", "r") as file: | |
content = file.read() | |
return HTMLResponse(content=content) | |
async def presentation(): | |
with open("static/presentation-agent.html", "r") as file: | |
content = file.read() | |
return HTMLResponse(content=content) | |
async def presentation(): | |
with open("static/mult-agent-auth.html", "r") as file: | |
content = file.read() | |
return HTMLResponse(content=content) | |
async def presentation(): | |
with open("static/login.html", "r") as file: | |
content = file.read() | |
return HTMLResponse(content=content) | |
async def serve_resume_optimizer(): | |
with open("static/file-convert.html", "r") as file: | |
content = file.read() | |
return HTMLResponse(content=content) | |
class EmailCheck(BaseModel): | |
email: str | |
ALLOWLIST = ["[email protected]", "[email protected]"] # Your allowlist | |
async def check_allowlist(email_check: EmailCheck): | |
print(f"Checking email: {email_check.email}") | |
try: | |
if email_check.email in ALLOWLIST: | |
print("Email is in allowlist") | |
try: | |
# Try to read the JSON file with UTF-8 encoding | |
with open("app_urls.json", "r", encoding="utf-8") as file: | |
app_data = json.load(file) | |
except UnicodeDecodeError: | |
# If UTF-8 fails, try with ISO-8859-1 encoding | |
with open("app_urls.json", "r", encoding="iso-8859-1") as file: | |
app_data = json.load(file) | |
return { | |
"allowed": True, | |
"app_data": app_data | |
} | |
else: | |
print("Email is not in allowlist") | |
return {"allowed": False} | |
except Exception as e: | |
print(f"Error: {e}") | |
raise HTTPException(status_code=500, detail=str(e)) | |
async def http_exception_handler(request: Request, exc: HTTPException): | |
return JSONResponse(status_code=exc.status_code, content={"detail": exc.detail}) | |
if __name__ == "__main__": | |
uvicorn.run(app, host="0.0.0.0", port=8000, debug=True) |