Spaces:
Running
Running
| from __future__ import annotations | |
| import json | |
| from pathlib import Path | |
| from typing import List, Dict, Any | |
| from app.models.resume import ResumeData, Experience | |
| from app.models.job import JobData | |
| from app.models.customization import CustomizationResult, Change, Intensity | |
| from app.models.score import ATSScore | |
| from app.models.analysis import BulletAnalysis, KeywordPlacement | |
| from app.services.ats_scorer import ATSScorer | |
| from app.services.bullet_analyzer import BulletAnalyzer | |
| from app.llm.factory import LLMFactory | |
| class ResumeCustomizer: | |
| def __init__(self): | |
| self.prompts_dir = Path(__file__).parent.parent.parent / "prompts" | |
| self.scorer = ATSScorer() | |
| self.bullet_analyzer = BulletAnalyzer() | |
| def _detect_changes(self, original: ResumeData, customized: ResumeData) -> List[Change]: | |
| """Compare original and customized to detect changes.""" | |
| changes = [] | |
| # Compare experience bullets | |
| for i, (orig_exp, cust_exp) in enumerate(zip(original.experience, customized.experience)): | |
| for j, (orig_bullet, cust_bullet) in enumerate(zip(orig_exp.bullets, cust_exp.bullets)): | |
| if orig_bullet != cust_bullet: | |
| changes.append(Change( | |
| type="modified", | |
| location=f"experience[{i}].bullets[{j}]", | |
| before=orig_bullet, | |
| after=cust_bullet, | |
| )) | |
| # Check for added bullets | |
| if len(cust_exp.bullets) > len(orig_exp.bullets): | |
| for j in range(len(orig_exp.bullets), len(cust_exp.bullets)): | |
| changes.append(Change( | |
| type="added", | |
| location=f"experience[{i}].bullets[{j}]", | |
| before="", | |
| after=cust_exp.bullets[j], | |
| )) | |
| # Compare skills | |
| orig_skills = set(original.skills) | |
| cust_skills = set(customized.skills) | |
| for skill in cust_skills - orig_skills: | |
| changes.append(Change( | |
| type="added", | |
| location="skills", | |
| before="", | |
| after=skill, | |
| )) | |
| # Compare summary | |
| if original.summary != customized.summary: | |
| changes.append(Change( | |
| type="modified", | |
| location="summary", | |
| before=original.summary, | |
| after=customized.summary, | |
| )) | |
| return changes | |
| async def customize( | |
| self, | |
| resume: ResumeData, | |
| job: JobData, | |
| intensity: Intensity = Intensity.MODERATE, | |
| ) -> CustomizationResult: | |
| """Customize resume for the target job.""" | |
| import logging | |
| logger = logging.getLogger(__name__) | |
| # Calculate original score | |
| original_score = await self.scorer.calculate(resume, job) | |
| # Analyze bullets BEFORE customization (optional feature) | |
| bullet_analysis: List[BulletAnalysis] = [] | |
| try: | |
| bullet_analysis = await self.bullet_analyzer.analyze_all_bullets(resume, job) | |
| except Exception as e: | |
| logger.warning(f"Bullet analysis failed (non-critical): {e}") | |
| # Prepare prompt | |
| prompt_template = (self.prompts_dir / "customize_resume.txt").read_text() | |
| resume_dict = resume.model_dump() | |
| del resume_dict["raw_text"] # Don't include raw text in prompt | |
| prompt = prompt_template.format( | |
| intensity=intensity.value, | |
| resume_json=json.dumps(resume_dict, indent=2), | |
| job_title=job.title, | |
| job_company=job.company, | |
| keywords_required=", ".join(job.keywords_required), | |
| keywords_preferred=", ".join(job.keywords_preferred), | |
| responsibilities="\n".join(f"- {r}" for r in job.responsibilities[:5]), | |
| missing_keywords=", ".join(original_score.missing_keywords[:10]), | |
| ) | |
| # Get customized resume from LLM | |
| llm = LLMFactory.get_smart() | |
| schema: Dict[str, Any] = resume_dict # Use original structure as schema | |
| customized_dict = await llm.complete_json(prompt, schema) | |
| # Preserve raw_text from original | |
| customized_dict["raw_text"] = resume.raw_text | |
| customized = ResumeData(**customized_dict) | |
| # Calculate new score | |
| customized_score = await self.scorer.calculate(customized, job) | |
| # Detect changes | |
| changes = self._detect_changes(resume, customized) | |
| # Update bullet analysis with customized versions (optional feature) | |
| try: | |
| if bullet_analysis: | |
| bullet_analysis = self.bullet_analyzer.update_with_customized( | |
| bullet_analysis, customized, job | |
| ) | |
| except Exception as e: | |
| logger.warning(f"Bullet analysis update failed (non-critical): {e}") | |
| # Check keyword quality (optional feature) | |
| keyword_quality: List[KeywordPlacement] = [] | |
| try: | |
| added_keywords = [ | |
| kw for kw in customized_score.matched_keywords | |
| if kw not in original_score.matched_keywords | |
| ] | |
| keyword_quality = self.scorer.check_keyword_quality( | |
| customized, job, added_keywords | |
| ) | |
| except Exception as e: | |
| logger.warning(f"Keyword quality check failed (non-critical): {e}") | |
| return CustomizationResult( | |
| original=resume, | |
| customized=customized, | |
| changes=changes, | |
| original_score=original_score, | |
| customized_score=customized_score, | |
| bullet_analysis=bullet_analysis, | |
| keyword_quality=keyword_quality, | |
| ) | |