from typing import List, Optional from pydantic import BaseModel from docx import Document from docx.shared import Pt, Inches from docx.enum.text import WD_ALIGN_PARAGRAPH from sendgrid import SendGridAPIClient from sendgrid.helpers.mail import Mail, Attachment, FileContent, FileName, FileType, Disposition import re import os import base64 from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText from email.mime.application import MIMEApplication import smtplib import json from jinja2 import Template import pdfkit class Filter: class Valves(BaseModel): pipelines: List[str] = [] priority: int = 0 SENDGRID_API_KEY: str = "" PASSWORD: str="" def __init__(self): self.type = "filter" self.name = "Generate document & send email" self.valves = self.Valves( **{ "pipelines": ["*"], "SENDGRID_API_KEY": os.getenv("SENDGRID_API_KEY", ""), "PASSWORD": os.getenv("PASSWORD", "") } ) self.recipients = [] self.resume_content = None self.sender = "[email protected]" async def on_startup(self): print(f"on_startup:{__name__}") pass async def on_shutdown(self): print(f"on_shutdown:{__name__}") pass async def inlet(self, body: dict, user: Optional[dict] = None) -> dict: if body.get("title", False): return body user_message = self.get_last_user_message(body["messages"]) email_pattern = r'[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}' self.recipients = re.findall(email_pattern, user_message) return body async def outlet(self, body: dict, user: Optional[dict] = None) -> dict: print(f"outlet:{__name__}") assistent_response = self.get_last_assistent_message(body["messages"]) result = self.extract_resume_content(assistent_response) if result: self.resume_content = result output_path = self.resume_content["name"]["first_name"] + "_" + self.resume_content["name"]["last_name"] + ".pdf" html = self.generate_html() self.save_html_to_pdf(html, output_path) if self.recipients: self.send_cv_via_email(output_path) return body def get_last_assistent_message(self, messages): for message in list(reversed(messages)): if message["role"] == "assistant": return message["content"] def get_last_user_message(self, messages): for message in list(reversed(messages)): if message["role"] == "user": return message["content"] def send_cv_via_email(self, cv_path): body = "This is the CV generated by our assistant." msg = MIMEMultipart() msg['Subject'] = "CV CC RO Template" msg['From'] = self.sender msg['To'] = ', '.join(self.recipients) # Attach the body of the email msg.attach(MIMEText(body, 'plain')) # Attach the PDF file with open(cv_path, 'rb') as pdf_file: pdf_attachment = MIMEApplication(pdf_file.read(), _subtype='pdf') pdf_attachment.add_header('Content-Disposition', 'attachment', filename='file.pdf') msg.attach(pdf_attachment) # Send the email with smtplib.SMTP_SSL('smtp.gmail.com', 465) as smtp_server: smtp_server.login(self.sender, self.valves.PASSWORD) smtp_server.sendmail(self.sender, self.recipients, msg.as_string()) print("Message sent!") # def send_report_via_email(self, docx_path): # message = Mail( # from_email='[email protected]', # to_emails=self.recipients, # subject="Raport de protocol", # html_content="This is the protocol report generated by our assistent." # ) # # Read and encode the file as Base64 # with open(docx_path, 'rb') as f: # file_data = f.read() # encoded_file = base64.b64encode(file_data).decode() # # Create the attachment # attached_file = Attachment( # FileContent(encoded_file), # FileName(os.path.basename(docx_path)), # FileType('application/vnd.openxmlformats-officedocument.wordprocessingml.document'), # Disposition('attachment') # ) # # Add the attachment to the email # message.attachment = attached_file # # try: # # Send the email # sg = SendGridAPIClient(self.valves.SENDGRID_API_KEY) # response = sg.send(message) # print(f"Email sent! Status Code: {response.status_code}") # print(f"Response Body: {response.body}") # print(f"Response Headers: {response.headers}") # # except Exception as e: # # print(f"Error sending email: {str(e)}") def extract_resume_content(self, text): json_match = re.search(r'\{.*\}', text, re.DOTALL) if json_match: json_str = json_match.group(0) try: json_data = json.loads(json_str) except json.JSONDecodeError: return None # Top-level required fields and their expected types required_fields = { "name": dict, "role": str, "seniority_level": str, "executive_summary": dict, "professional_skills": list, "work_experience": list, "education": list, "certifications": list } # Validate presence and type of top-level fields for key, expected_type in required_fields.items(): if not isinstance(json_data.get(key), expected_type): return None # Validate nested dictionary structures name = json_data.get("name", {}) if not isinstance(name.get("last_name"), str) or not isinstance(name.get("first_name"), str): return None executive_summary = json_data.get("executive_summary", {}) if not isinstance(executive_summary.get("about_job_profile"), str) or not isinstance(executive_summary.get("about_professional"), str): return None # Validate list of skills, work experience, education, and certifications entries def validate_list_items(data_list, required_keys): for item in data_list: if not all(isinstance(item.get(k), t) for k, t in required_keys.items()): return False return True if not validate_list_items(json_data.get("professional_skills", []), {"skill": str, "rating": int}): return None if not validate_list_items(json_data.get("work_experience", []), { "role": str, "company_name": str, "responsibilities": str, "time_interval": dict, "skills_tools": list }) or not all(isinstance(w.get("time_interval", {}).get(k), str) for w in json_data["work_experience"] for k in ["start", "end"]): return None if not validate_list_items(json_data.get("education", []), { "level": str, "topic": str, "institution": str, "description": str, "time_interval": dict }) or not all(isinstance(e.get("time_interval", {}).get(k), str) for e in json_data["education"] for k in ["start", "end"]): return None if not validate_list_items(json_data.get("certifications", []), { "title": str, "organization": str, "date": str, "description": str }): return None return json_data return None def generate_html(self): html_template = """ <!DOCTYPE html> <html lang="en"> <style> /* Basic styling */ body { font-family: 'Roboto', sans-serif; } .progress-container { background-color: #e0e0e0; border-radius: 5px; height: 20px; width: 100%; overflow: hidden; margin-bottom: 15px; } .progress-bar { height: 100%; transition: width 0.3s ease-in-out; border-radius: 5px; } .progress-bar-20 { width: 20%; background-color: #6a0ffa; } .progress-bar-40 { width: 40%; background-color: #6a0ffa; } .progress-bar-60 { width: 60%; background-color: #6a0ffa; } .progress-bar-80 { width: 80%; background-color: #6a0ffa; } .progress-bar-100 { width: 100%; background-color: #6a0ffa; } .skill-label { font-size: 16px; font-weight: bold; color: #333; } .card { margin-bottom: 40px; /* Adjust this value for more or less spacing */ padding: 20px; border-radius: 8px; } </style> <div class="container" id="content"> <!-- Resume Header Card --> <div class="card" style="background-color: #151523; text-align: center; padding-bottom: 20px; padding-top: 0; padding-left: 0; padding-right: 0; margin: 20px 0;"> <div style="background-color: #6a0ffa; height: 10px; width: 100%; border-top-left-radius: 10px; border-top-right-radius: 10px;"></div> <h1 style="color: white; font-size: 1.75rem;">{{ employee_name }}</h1> <h2 style="color: white; font-size: 1.2rem;">{{ job_profile }}</h2> {% if seniority_level %} <h3 style="color: white;">{{ seniority_level }} Level</h3> {% endif %} </div> <!-- Executive Summary Section --> {% if about_job_profile or about_professional %} <div class="card" style="margin: 20px 0;"> <h2>Executive Summary</h2> {% if about_job_profile %} <h4>About the Job Profile</h4> <p>{{ about_job_profile }}</p> {% endif %} {% if about_professional %} <h4>About the Professional</h4> <p>{{ about_professional }}</p> {% endif %} </div> {% endif %} <!-- Professional Skills Section --> <div class="card" style="margin: 20px 0;"> <h2>Professional Skills</h2> {% for skill in professional_skills %} <div class="skill-item"> <span class="skill-label">{{ skill.skill }}</span> <div class="progress-container"> <div class="progress-bar {% if skill.rating == 1 %} progress-bar-20 {% elif skill.rating == 2 %} progress-bar-40 {% elif skill.rating == 3 %} progress-bar-60 {% elif skill.rating == 4 %} progress-bar-80 {% elif skill.rating == 5 %} progress-bar-100 {% endif %}"></div> </div> </div> {% endfor %} </div> <!-- Work Experience Section --> {% if work_experience %} <div class="card" style="margin: 20px 0;"> <h2>Work Experience</h2> {% for work in work_experience %} <div> <strong>{{ work.role }} at {{ work.company_name }}</strong><br> <small>{{ work.time_interval.start }} - {{ work.time_interval.end }}</small> <p style="font-weight: bold;">Responsabilities</p> <p>{{ work.responsibilities }}</p> <p style="font-weight: bold;">Skills and Tools:</p> {% for skill in work.skills_tools%} <p>{{ skill }}</p> {% endfor %} </div> <hr> {% endfor %} </div> {% endif %} <!-- Education Section --> {% if education %} <div class="card" style="margin: 20px 0;"> <h2>Education</h2> {% for edu in education %} <div> <strong>{{ edu.level }} from {{ edu.institution }}</strong><br> <small>{{ edu.time_interval.start }} - {{ edu.time_interval.end }}</small> <p>{{ edu.description }}</p> </div> <hr> {% endfor %} </div> {% endif %} <!-- Certifications Section --> {% if certifications %} <div class="card" style="margin: 20px 0;"> <h2>Certifications</h2> {% for cert in certifications %} <div> <strong>{{ cert.title }} from {{ cert.organization }}</strong><br> <small>{{ cert.date }}</small> <p>{{ cert.description }}</p> </div> <hr> {% endfor %} </div> {% endif %} </div> </html> """ # Prepare Jinja2 template template = Template(html_template) context = { "site_brand": "Your Brand", # Replace as necessary "employee_name": f"{self.resume_content['name']['last_name'].upper()}, {self.resume_content['name']['first_name']}", "job_profile": self.resume_content['role'], "seniority_level": self.resume_content['seniority_level'], "about_job_profile": self.resume_content['executive_summary']["about_job_profile"], "about_professional": self.resume_content['executive_summary']["about_professional"], "professional_skills": self.resume_content['professional_skills'], "work_experience": self.resume_content['work_experience'], "education": self.resume_content['education'], "certifications": self.resume_content['certifications'] } return template.render(context) def save_html_to_pdf(self, html_content, output_pdf_path): options = { 'encoding': 'UTF-8', 'page-size': 'A4', 'margin-top': '0.5in', 'margin-right': '0.5in', 'margin-bottom': '0.5in', 'margin-left': '0.5in' } pdfkit.from_string(html_content, output_pdf_path, options=options)