Spaces:
Runtime error
Runtime error
File size: 2,943 Bytes
b3712d9 434bd0b b8e9f2b 434bd0b 15f88b7 b8e9f2b cdb0794 0376f20 cdb0794 0376f20 cdb0794 15f88b7 cdb0794 0376f20 cdb0794 0376f20 cdb0794 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 |
from fastapi import FastAPI, File, UploadFile
from pydantic import BaseModel
from typing import List
from pathlib import Path
import shutil
from langchain_docling import DoclingLoader
from langchain_docling.loader import ExportType
app = FastAPI()
resumes = []
jobs = []
UPLOAD_DIR = Path("uploads")
UPLOAD_DIR.mkdir(exist_ok=True)
@app.post("/upload")
async def upload_file(file: UploadFile = File(...)):
file_path = UPLOAD_DIR / file.filename
with file_path.open("wb") as buffer:
shutil.copyfileobj(file.file, buffer)
# result = process_with_langchain(file_path)
loader = DoclingLoader(file_path=FILE_PATH, export_type=ExportType.MARKDOWN)
docs = loader.load()
# docs = docs.model_dump()
return {
"code":201,
"message":"Request was successful.",
"data": docs[0].model_dump()
}
# return {"filename": file.filename, "path": str(file_path), "status": "uploaded"}
# class InputResume(BaseModel):
# content: str
# @app.post("/suggest/")
# async def suggestion(data: InputResume):
# return {
# "code":201,
# "message":"Request was successful.",
# "data": InputResume.model_dump_json()
# }
from ranker import rank_resume
from embeddings import rank_jobs
# Function to wrap the existing rank_resume
def process_input(job_description, resumes):
print("[JOB DESC]", job_description)
print("[RESUMES]", resumes)
resumes = [r for r in resumes if r and r.strip() != ""] # Remove empty
if not job_description.strip() or not resumes:
return "Please provide both job description and at least one resume."
return rank_resume(job_description, resumes)[1]
def process_input_suggestion(resume, job_descriptions):
# print("[JOB DESC]", job_description)
# print("[RESUMES]", resumes)
# resumes = [r for r in resumes if r and r.strip() != ""] # Remove empty
# if not job_description.strip() or not resumes:
# return "Please provide both resume and at least one job description."
return rank_jobs(job_descriptions, resume)[1]
# results = zip(*rank_jobs(resumes, job_description))
# formatted_output = ""
# for i, (resume, score) in enumerate(results, 1):
# formatted_output += f"Job #{i}:\nScore: {score:.2f}\nJob Description Snippet: {resume[:200]}...\n\n-------\n\n"
# return formatted_output
app.get("/")
def read_root():
return {"message": "Hello, World!"}
class InputData(BaseModel):
resumes: List[str]
job_description: str
class InputData2(BaseModel):
job_descriptions: List[str]
resume: str
@app.post("/rank/")
async def process_data(data: InputData):
return dict(scores=process_input(data.job_description, data.resumes))
@app.post("/suggest/")
async def suggestion(data: InputData2):
return {
"scores":process_input_suggestion(data.resume, data.job_descriptions)
}
|