ai-hackaton-backend/app/services/pdf_report_service.py
2025-09-10 23:55:56 +05:00

270 lines
10 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

import io
import os
from datetime import datetime
from jinja2 import Template
from app.core.s3 import s3_service
from app.models.interview_report import InterviewReport, RecommendationType
class PDFReportService:
"""Сервис для генерации и загрузки PDF отчетов в S3 хранилище"""
def __init__(self):
self.template_path = "templates/interview_report.html"
def _load_html_template(self) -> str:
"""Загружает HTML шаблон из файла"""
try:
with open(self.template_path, encoding="utf-8") as file:
return file.read()
except FileNotFoundError:
raise FileNotFoundError(f"HTML шаблон не найден: {self.template_path}")
def _format_concerns_field(self, concerns):
"""Форматирует поле concerns для отображения"""
if not concerns:
return ""
if isinstance(concerns, list):
return "; ".join(concerns)
elif isinstance(concerns, str):
return concerns
else:
return str(concerns)
def _format_list_field(self, field_value) -> str:
"""Форматирует поле со списком для отображения"""
if not field_value:
return "Не указаны"
if isinstance(field_value, list):
return "\n".join([f"{item}" for item in field_value])
elif isinstance(field_value, str):
return field_value
else:
return str(field_value)
def _get_score_class(self, score: int) -> str:
"""Возвращает CSS класс для цвета оценки"""
if score >= 90:
return "score-green" # STRONGLY_RECOMMEND
elif score >= 75:
return "score-green" # RECOMMEND
elif score >= 60:
return "score-orange" # CONSIDER
else:
return "score-red" # REJECT
def _format_recommendation(self, recommendation: RecommendationType) -> tuple:
"""Форматирует рекомендацию для отображения"""
if recommendation == RecommendationType.STRONGLY_RECOMMEND:
return ("Настоятельно рекомендуем", "recommend-button")
elif recommendation == RecommendationType.RECOMMEND:
return ("Рекомендуем", "recommend-button")
elif recommendation == RecommendationType.CONSIDER:
return ("К рассмотрению", "consider-button")
else: # REJECT
return ("Не рекомендуем", "reject-button")
async def generate_html_report(
self,
interview_report: InterviewReport,
candidate_name: str = None,
position: str = None,
resume_file_url: str = None,
) -> str:
"""
Генерирует HTML отчет на основе Jinja2 шаблона
Args:
interview_report: Данные отчета по интервью
candidate_name: Имя кандидата
position: Позиция
resume_file_url: URL резюме
Returns:
str: HTML содержимое отчета
"""
try:
# Загружаем HTML шаблон
html_template = self._load_html_template()
# Подготавливаем данные для шаблона
template_data = self._prepare_template_data(
interview_report,
candidate_name or "Не указано",
position or "Не указана",
resume_file_url,
)
# Рендерим HTML с данными
template = Template(html_template)
rendered_html = template.render(**template_data)
# Сохраняем debug.html для отладки
with open("debug.html", "w", encoding="utf-8") as f:
f.write(rendered_html)
return rendered_html
except Exception as e:
raise Exception(f"Ошибка при генерации HTML отчета: {str(e)}")
def _prepare_template_data(
self,
interview_report: InterviewReport,
candidate_name: str,
position: str,
resume_file_url: str = None,
) -> dict:
"""Подготавливает данные для HTML шаблона"""
# Используем переданные параметры
resume_url = resume_file_url or "#"
# Форматируем дату интервью
interview_date = "Не указана"
if (
interview_report.interview_session
and interview_report.interview_session.interview_start_time
):
interview_date = (
interview_report.interview_session.interview_start_time.strftime(
"%d.%m.%Y %H:%M"
)
)
# Общий балл и рекомендация
overall_score = interview_report.overall_score or 0
recommendation_text, recommendation_class = self._format_recommendation(
interview_report.recommendation
)
# Сильные стороны и области развития
strengths = (
self._format_list_field(interview_report.strengths)
if interview_report.strengths
else "Не указаны"
)
areas_for_development = (
self._format_list_field(interview_report.weaknesses)
if interview_report.weaknesses
else "Не указаны"
)
# Детальная оценка - всегда все критерии
evaluation_criteria = [
{
"name": "Технические навыки",
"score": interview_report.technical_skills_score or 0,
"score_class": self._get_score_class(
interview_report.technical_skills_score or 0
),
"justification": interview_report.technical_skills_justification or "",
"concerns": self._format_concerns_field(
interview_report.technical_skills_concerns
),
},
{
"name": "Релевантность опыта",
"score": interview_report.experience_relevance_score or 0,
"score_class": self._get_score_class(
interview_report.experience_relevance_score or 0
),
"justification": interview_report.experience_relevance_justification
or "",
"concerns": self._format_concerns_field(
interview_report.experience_relevance_concerns
),
},
{
"name": "Коммуникация",
"score": interview_report.communication_score or 0,
"score_class": self._get_score_class(
interview_report.communication_score or 0
),
"justification": interview_report.communication_justification or "",
"concerns": self._format_concerns_field(
interview_report.communication_concerns
),
},
{
"name": "Решение задач",
"score": interview_report.problem_solving_score or 0,
"score_class": self._get_score_class(
interview_report.problem_solving_score or 0
),
"justification": interview_report.problem_solving_justification or "",
"concerns": self._format_concerns_field(
interview_report.problem_solving_concerns
),
},
{
"name": "Культурное соответствие",
"score": interview_report.cultural_fit_score or 0,
"score_class": self._get_score_class(
interview_report.cultural_fit_score or 0
),
"justification": interview_report.cultural_fit_justification or "",
"concerns": self._format_concerns_field(
interview_report.cultural_fit_concerns
),
},
]
# Красные флаги
red_flags = interview_report.red_flags or []
# ID отчета
report_id = f"#{interview_report.id}" if interview_report.id else "#0"
# Дата генерации отчета
generation_date = datetime.now().strftime("%d.%m.%Y %H:%M")
return {
"report_id": report_id,
"candidate_name": candidate_name,
"position": position,
"interview_date": interview_date,
"overall_score": overall_score,
"recommendation_text": recommendation_text,
"recommendation_class": recommendation_class,
"strengths": strengths,
"areas_for_development": areas_for_development,
"evaluation_criteria": evaluation_criteria,
"red_flags": red_flags,
"resume_url": resume_url,
"generation_date": generation_date,
}
async def upload_pdf_to_s3(self, pdf_bytes: bytes, filename: str) -> str:
"""
Загружает PDF файл в S3 и возвращает публичную ссылку
Args:
pdf_bytes: PDF файл в виде байтов
filename: Имя файла
Returns:
str: Публичная ссылка на файл в S3
"""
try:
pdf_stream = io.BytesIO(pdf_bytes)
# Загружаем с публичным доступом
file_url = await s3_service.upload_file(
pdf_stream, filename, content_type="application/pdf", public=True
)
return file_url
except Exception as e:
raise Exception(f"Ошибка при загрузке PDF в S3: {str(e)}")
# Экземпляр сервиса
pdf_report_service = PDFReportService()