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
class Pipe:
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.docx_path = "cc_ro_template_cv.docx"
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
self.generate_docx()
if self.recipients and self.resume_content:
self.send_cv_via_email(self.docx_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, docx_path):
body = "This is the cv generated by our assistent."
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 DOCX file
with open(docx_path, 'rb') as docx_file:
docx_attachment = MIMEApplication(docx_file.read(), _subtype='vnd.openxmlformats-officedocument.wordprocessingml.document')
docx_attachment.add_header('Content-Disposition', 'attachment', filename='file.docx')
msg.attach(docx_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)
json_data = json.loads(json_str)
return json_data
else:
return None
def generate_docx(self):
document = Document()
section = document.sections[0]
self.add_name(document)
self.add_role(document)
self.add_seniority_level(document)
self.add_exrcutive_summary(document)
self.add_skills(document)
self.add_work_experience(document)
self.add_education(document)
self.add_certifications(document)
document.save("cc_ro_template_cv.docx")
def add_name(self, document):
name_paragraph = document.add_paragraph()
name_paragraph_run = name_paragraph.add_run()
name_paragraph_run.text = self.resume_content["name"]["last_name"].upper() + ", " + self.resume_content["name"]["first_name"]
name_paragraph_run.bold = True
name_paragraph.alignment = WD_ALIGN_PARAGRAPH.CENTER
name_paragraph_run.font.name = 'Arial'
name_paragraph_run.font.size = Pt(20)
def add_role(self, document):
role_paragraph = document.add_paragraph()
role_run = role_paragraph.add_run()
role_run.text = self.resume_content["role"]
role_paragraph.alignment = WD_ALIGN_PARAGRAPH.CENTER
role_run.font.name = 'Arial'
role_run.font.size = Pt(16)
def add_seniority_level(self, document):
seniority_level_paragraph = document.add_paragraph()
seniority_level_run = seniority_level_paragraph.add_run()
seniority_level_run.text = self.resume_content["seniority_level"] + " Level\n\n"
seniority_level_run.bold = True
seniority_level_paragraph.alignment = WD_ALIGN_PARAGRAPH.CENTER
seniority_level_run.font.name = 'Arial'
seniority_level_run.font.size = Pt(14)
def add_exrcutive_summary(self, document):
title_paragraph = document.add_paragraph()
title_run = title_paragraph.add_run()
title_run.text = "Executive Summary"
title_paragraph.alignment = WD_ALIGN_PARAGRAPH.LEFT
title_run.font.name = 'Arial'
title_run.font.size = Pt(18)
job_profile_paragraph = document.add_paragraph()
subtitle1_run = job_profile_paragraph.add_run()
subtitle1_run.text = "About the Job Profile\n"
job_profile_paragraph.alignment = WD_ALIGN_PARAGRAPH.LEFT
subtitle1_run.font.name = 'Arial'
subtitle1_run.font.size = Pt(16)
job_profile_run = job_profile_paragraph.add_run()
job_profile_run.text = self.resume_content["executive_summary"]["about_job_profile"]
job_profile_run.font.name = 'Arial'
job_profile_run.font.size = Pt(12)
about_professional_paragraph = document.add_paragraph()
subtitle2_run = about_professional_paragraph.add_run()
subtitle2_run.text = "About the Pofessional\n"
about_professional_paragraph.alignment = WD_ALIGN_PARAGRAPH.LEFT
subtitle2_run.font.name = 'Arial'
subtitle2_run.font.size = Pt(16)
about_professional_run = about_professional_paragraph.add_run()
about_professional_run.text = self.resume_content["executive_summary"]["about_professional"] + "\n\n"
about_professional_run.font.name = 'Arial'
about_professional_run.font.size = Pt(12)
def add_skills(self, document):
skills_paragraph = document.add_paragraph()
title_run = skills_paragraph.add_run("Professoinal Skills\n")
skills_paragraph.alignment = WD_ALIGN_PARAGRAPH.LEFT
title_run.font.name = 'Arial'
title_run.font.size = Pt(18)
skill_run = skills_paragraph.add_run()
text = ""
professional_skills = self.resume_content["professional_skills"]
for skill in professional_skills:
skill_name = skill["skill"]
skill_rating = skill["rating"]
text += f"{skill_name}: {skill_rating}/5\n"
skill_run.text = text + "\n"
skill_run.font.name = 'Arial'
skill_run.font.size = Pt(12)
def add_work_experience(self, document):
if self.resume_content["work_experience"]:
work_experience_paragraph = document.add_paragraph()
title_run = work_experience_paragraph.add_run("Work Experience")
work_experience_paragraph.alignment = WD_ALIGN_PARAGRAPH.LEFT
title_run.font.name = 'Arial'
title_run.font.size = Pt(18)
work_experience = self.resume_content["work_experience"]
for experience in work_experience:
# Add role and company name
role_company_paragraph = document.add_paragraph()
role_run = role_company_paragraph.add_run(f"{experience['role']}")
role_run.font.name = 'Arial'
role_run.font.size = Pt(14)
role_run.bold = True
role_company_paragraph.alignment = WD_ALIGN_PARAGRAPH.LEFT
company_run = role_company_paragraph.add_run(f" at {experience['company_name']}")
company_run.font.name = 'Arial'
company_run.font.size = Pt(12)
company_run.bold = False
# Add time interval
time_paragraph = document.add_paragraph()
time_run = time_paragraph.add_run(f"{experience['time_interval']['start']} - {experience['time_interval']['end']}\n")
time_run.font.name = 'Arial'
time_run.font.size = Pt(12)
time_paragraph.alignment = WD_ALIGN_PARAGRAPH.LEFT
# Add responsibilities
responsibilities_paragraph = document.add_paragraph()
responsibilities_run = responsibilities_paragraph.add_run("Responsibilities:\n" + experience['responsibilities'] + "\n")
responsibilities_run.font.name = 'Arial'
responsibilities_run.font.size = Pt(12)
# Add skills/tools
skills_tools_paragraph = document.add_paragraph()
skills_text = "Skills and Tools: " + ", ".join(experience['skills_tools'])
skills_run = skills_tools_paragraph.add_run(skills_text + "\n")
skills_run.font.name = 'Arial'
skills_run.font.size = Pt(12)
# Add space between experiences
document.add_paragraph("\n")
def add_education(self, document):
if self.resume_content["education"]:
education_paragraph = document.add_paragraph()
title_run = education_paragraph.add_run("Education")
education_paragraph.alignment = WD_ALIGN_PARAGRAPH.LEFT
title_run.font.name = 'Arial'
title_run.font.size = Pt(18)
education_list = self.resume_content["education"]
for education in education_list:
education_paragraph = document.add_paragraph()
level_topic_run = education_paragraph.add_run(f"{education['level']} - {education['topic']}")
level_topic_run.font.size = Pt(14)
level_topic_run.bold = True
level_topic_run.font.name = 'Arial'
education_paragraph.alignment = WD_ALIGN_PARAGRAPH.LEFT
institution_run = education_paragraph.add_run(f" from {education['institution']}")
institution_run.font.name = 'Arial'
institution_run.font.size = Pt(12)
institution_run.bold = False
# Add time interval
time_paragraph = document.add_paragraph()
time_run = time_paragraph.add_run(f"{education['time_interval']['start']} - {education['time_interval']['end']}\n")
time_run.font.name = 'Arial'
time_run.font.size = Pt(12)
time_paragraph.alignment = WD_ALIGN_PARAGRAPH.LEFT
# Add description
description_paragraph = document.add_paragraph()
description_run = description_paragraph.add_run(education['description'])
description_run.font.name = 'Arial'
description_run.font.size = Pt(12)
# Add space between entries
document.add_paragraph("\n")
def add_certifications(self, document):
if self.resume_content["certifications"]:
certifications_paragraph = document.add_paragraph()
title_run = certifications_paragraph.add_run("Certifiations")
certifications_paragraph.alignment = WD_ALIGN_PARAGRAPH.LEFT
title_run.font.name = 'Arial'
title_run.font.size = Pt(18)
certifications_list = self.resume_content["certifications"]
for certification in certifications_list:
certification_paragraph = document.add_paragraph()
# Create a run for the certification title, set to bold
title_run = certification_paragraph.add_run(f"{certification['title']}")
title_run.font.name = 'Arial'
title_run.font.size = Pt(14)
title_run.bold = True
# Create a run for " from " and the issuing organization, set to regular weight
organization_run = certification_paragraph.add_run(f" from {certification['organization']}")
organization_run.font.name = 'Arial'
organization_run.font.size = Pt(12)
organization_run.bold = False
certification_paragraph.alignment = WD_ALIGN_PARAGRAPH.LEFT
# Add date
date_paragraph = document.add_paragraph()
date_run = date_paragraph.add_run(f"{certification['date']}\n")
date_run.font.name = 'Arial'
date_run.font.size = Pt(12)
date_paragraph.alignment = WD_ALIGN_PARAGRAPH.LEFT
# Add description
description_paragraph = document.add_paragraph()
description_run = description_paragraph.add_run(certification['description'])
description_run.font.name = 'Arial'
description_run.font.size = Pt(12)
# Add space between certifications
document.add_paragraph("\n")
