Spaces:
Runtime error
Runtime error
File size: 7,248 Bytes
e33a96e | 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 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 | from fastapi import APIRouter, Depends, HTTPException, UploadFile, File
from sqlalchemy.orm import Session
from typing import List
from app.database import get_db
from app.models import UploadedRFP, Question
from app.schemas import ProcessQuestionRequest, AIAnswerResponse, SimilarQuestion
from app.services.pdf_service import pdf_service
from app.services.similarity_service import similarity_service
from app.services.huggingface_service import huggingface_service
router = APIRouter(prefix="/api/rfp-processing", tags=["RFP Processing"])
@router.post("/upload")
async def upload_rfp(file: UploadFile = File(...), db: Session = Depends(get_db)):
"""Upload a new RFP PDF and extract text from first 3 pages"""
if not file.filename.endswith('.pdf'):
raise HTTPException(status_code=400, detail="Only PDF files are allowed")
uploaded_rfp = await pdf_service.save_and_extract_pdf(file, db, extract_pages=3)
return {
"id": uploaded_rfp.id,
"filename": uploaded_rfp.filename,
"uploaded_at": uploaded_rfp.uploaded_at,
"text_preview": uploaded_rfp.text_content[:500] + "..." if len(uploaded_rfp.text_content) > 500 else uploaded_rfp.text_content
}
@router.get("/uploaded-rfps")
def list_uploaded_rfps(skip: int = 0, limit: int = 50, db: Session = Depends(get_db)):
"""List all uploaded RFPs"""
rfps = db.query(UploadedRFP).offset(skip).limit(limit).all()
return [
{
"id": rfp.id,
"filename": rfp.filename,
"rfp_name": rfp.rfp_name if hasattr(rfp, 'rfp_name') and rfp.rfp_name else rfp.filename.rsplit('.', 1)[0],
"uploaded_at": rfp.uploaded_at
}
for rfp in rfps
]
@router.delete("/uploaded-rfps/{rfp_id}")
def delete_uploaded_rfp(rfp_id: int, db: Session = Depends(get_db)):
"""Delete an uploaded RFP"""
uploaded_rfp = db.query(UploadedRFP).filter(UploadedRFP.id == rfp_id).first()
if not uploaded_rfp:
raise HTTPException(status_code=404, detail="Uploaded RFP not found")
# Delete the file from disk if it exists
import os
if uploaded_rfp.file_path and os.path.exists(uploaded_rfp.file_path):
try:
os.remove(uploaded_rfp.file_path)
except Exception as e:
print(f"Error deleting file: {e}")
db.delete(uploaded_rfp)
db.commit()
return {"message": "Uploaded RFP deleted successfully"}
@router.post("/process-question", response_model=AIAnswerResponse)
async def process_question(request: ProcessQuestionRequest, db: Session = Depends(get_db)):
"""
Process a question against a new RFP:
1. Find similar questions from database
2. Get new RFP content (if provided)
3. Generate AI answer combining both
4. Provide gap analysis
"""
# Get uploaded RFP (optional)
uploaded_rfp = None
new_rfp_content = ""
if request.uploaded_rfp_id:
uploaded_rfp = db.query(UploadedRFP).filter(
UploadedRFP.id == request.uploaded_rfp_id
).first()
if not uploaded_rfp:
raise HTTPException(status_code=404, detail="Uploaded RFP not found")
new_rfp_content = uploaded_rfp.text_content
# Check for exact match first
exact_match = similarity_service.check_exact_match(db, request.question_text)
if exact_match:
# Return exact match answer
return AIAnswerResponse(
ai_answer=exact_match.answer_text,
gap_analysis="Exact match found in database. This is the stored answer from previous RFP." if not new_rfp_content else "Exact match found. No new information in current RFP.",
similar_questions=[
SimilarQuestion(
question_id=exact_match.id,
question_text=exact_match.question_text,
answer_text=exact_match.answer_text,
page_start=exact_match.page_start,
page_end=exact_match.page_end,
rfp_title=exact_match.rfp.title,
similarity_score=1.0
)
],
new_rfp_content=new_rfp_content[:1000] + "..." if new_rfp_content else "No new RFP uploaded"
)
# Find similar questions - get more matches for AI to consider
similar_questions = similarity_service.find_similar_questions(
db,
request.question_text,
top_k=10, # Get more answers for AI to reference
threshold=0.15 # Lower threshold to include more potential matches
)
# Format similar questions for response
similar_questions_list = [
SimilarQuestion(
question_id=q.id,
question_text=q.question_text,
answer_text=q.answer_text,
page_start=q.page_start,
page_end=q.page_end,
rfp_title=q.rfp.title,
similarity_score=score
)
for q, score in similar_questions
]
print(f"Found {len(similar_questions_list)} similar questions")
for sq in similar_questions_list:
print(f" - {sq.question_text} (score: {sq.similarity_score:.2f})")
# Prepare previous answers for AI
previous_answers = [
{
"question_text": q.question_text,
"answer_text": q.answer_text,
"page_start": q.page_start,
"page_end": q.page_end,
"rfp_title": q.rfp.title,
"similarity_score": score
}
for q, score in similar_questions
]
# Generate AI answer with HuggingFace
prompt = f"""Question: {request.question_text}
New RFP Content:
{new_rfp_content if new_rfp_content else "No new RFP content provided"}
Previous Similar Answers:
{chr(10).join([f"- {ans['answer_text']} (from {ans['rfp_title']}, score: {ans['similarity_score']:.2f})" for ans in previous_answers[:5]])}
{request.additional_context if request.additional_context else ""}
Provide a comprehensive answer and gap analysis."""
ai_answer = huggingface_service.generate_chat_response(prompt)
ai_response = {
'ai_answer': ai_answer,
'gap_analysis': "Generated using HuggingFace Mistral-7B model based on available context."
}
return AIAnswerResponse(
ai_answer=ai_response['ai_answer'],
gap_analysis=ai_response['gap_analysis'],
similar_questions=similar_questions_list,
new_rfp_content=new_rfp_content[:1000] + "..." if new_rfp_content else "No new RFP uploaded"
)
@router.get("/uploaded-rfps/{rfp_id}/content")
def get_rfp_content(rfp_id: int, db: Session = Depends(get_db)):
"""Get the full extracted content of an uploaded RFP"""
uploaded_rfp = db.query(UploadedRFP).filter(UploadedRFP.id == rfp_id).first()
if not uploaded_rfp:
raise HTTPException(status_code=404, detail="Uploaded RFP not found")
return {
"id": uploaded_rfp.id,
"filename": uploaded_rfp.filename,
"content": uploaded_rfp.text_content
}
|