Spaces:
Sleeping
Sleeping
| import base64 | |
| import re | |
| import tempfile | |
| from enum import Enum | |
| import httpx | |
| import pypandoc | |
| from docx import Document | |
| from docx.shared import Inches, Pt | |
| from fastapi import FastAPI, HTTPException | |
| from pydantic import BaseModel | |
| from src.utils import EmailClient | |
| app = FastAPI() | |
| pypandoc.download_pandoc() | |
| TEMPLATE_PATH = "src/template.docx" | |
| class Email(str, Enum): | |
| sukhwinder = "sukhwinder@sifars.com" | |
| munish = "munish@sifars.com" | |
| jatin = "jatin@sifars.com" | |
| class ProposalRequest(BaseModel): | |
| project_name: str | |
| project_overview: str | |
| project_introduction: str | |
| about_us: str | |
| background_of_the_project: str | |
| scope_of_work: str | |
| technical_approach: str | |
| deployment_and_integration: str | |
| timeline: str | |
| team_structure: str | |
| platform: str | |
| licensing_model: str | |
| risk_analysis: str | |
| past_projects: str | |
| architecture_diagram: str | |
| recipient_email: Email | |
| async def generate_proposal(data: dict): | |
| doc = Document(TEMPLATE_PATH) | |
| email_to_name = { | |
| "sukhwinder@sifars.com": "Sukhwinder Singh Sehgal", | |
| "munish@sifars.com": "Munish Kumar", | |
| "jatin@sifars.com": "Jatin Sethi", | |
| } | |
| data["name"] = email_to_name.get(data.get("recipient_email", ""), "").title() | |
| mermaid_code = data.get("architecture_diagram", "") | |
| architecture_diagram_path = None | |
| mermaid_code = ( | |
| mermaid_code.replace("```mermaid\n", "").replace("```", "").replace("\n", ";") | |
| ) | |
| if mermaid_code.strip(): | |
| graphbytes = mermaid_code.encode("utf8") | |
| base64_bytes = base64.urlsafe_b64encode(graphbytes) | |
| base64_str = base64_bytes.decode("ascii") | |
| url = f"https://mermaid.ink/img/{base64_str}" | |
| async with httpx.AsyncClient() as client: | |
| response = await client.get(url) | |
| response.raise_for_status() | |
| with tempfile.NamedTemporaryFile(delete=False, suffix=".png") as temp_file: | |
| temp_file.write(response.content) | |
| architecture_diagram_path = temp_file.name | |
| for para in doc.paragraphs: | |
| for key, value in data.items(): | |
| if f"{{{key}}}" in para.text: | |
| if key == "architecture_diagram": | |
| if architecture_diagram_path: | |
| para.clear() | |
| run = para.add_run() | |
| run.add_picture( | |
| architecture_diagram_path, width=Inches(6) | |
| ) | |
| else: | |
| clean_value = re.sub(r"[*#`]", "", value) | |
| para.text = para.text.replace(f"{{{key}}}", clean_value) | |
| for run in para.runs: | |
| run.font.size = Pt(14) | |
| run.font.name = "Arial" | |
| with tempfile.NamedTemporaryFile(suffix=".docx") as temp_file: | |
| doc.save(temp_file.name) | |
| docx_file_path = temp_file.name | |
| attachments = [docx_file_path] | |
| recipient_email = data.get("recipient_email", "sukhwinder@sifars.com") | |
| subject = f"Proposal for {data.get('project_name')}" | |
| body = "Hello,\n\nPlease find the proposal attached.\n\nBest,\nProposal Bot" | |
| async with EmailClient() as email_client: | |
| await email_client.send_email( | |
| subject=subject, | |
| body=body, | |
| to_emails=[recipient_email], | |
| attachments=attachments, | |
| ) | |
| return {"message": "Proposal sent successfully"} | |
| return {"message": "Failed to send proposal"} | |
| async def check_health(): | |
| return {"response": "Service is healthy!"} | |
| async def generate_proposal_endpoint(request: ProposalRequest): | |
| try: | |
| proposal = await generate_proposal(data=request.model_dump()) | |
| return {"proposal": proposal} | |
| except Exception as e: | |
| raise HTTPException(status_code=500, detail=f"{e}") | |