""" Excel Job Application Tracker Service Creates and manages job application tracking spreadsheets """ import os import json import logging from typing import Dict, Any, List, Optional from datetime import datetime import requests from pathlib import Path from models.schemas import JobPosting, ResumeDraft, CoverLetterDraft, OrchestrationResult logger = logging.getLogger(__name__) class ExcelTracker: """Create and manage job application tracking spreadsheets""" def __init__(self): self.mcp_server_url = os.getenv("EXCEL_MCP_URL", "http://localhost:3002") self.tracker_path = "job_applications_tracker.xlsx" def create_tracker(self, results: List[OrchestrationResult]) -> str: """Create comprehensive job application tracker""" try: # Try MCP server first if self._use_mcp_server(results): logger.info(f"Excel tracker created via MCP: {self.tracker_path}") return self.tracker_path # Fallback to openpyxl return self._create_with_openpyxl(results) except Exception as e: logger.error(f"Error creating Excel tracker: {e}") return None def update_application_status( self, job_id: str, status: str, notes: str = None, interview_date: str = None ) -> bool: """Update status of a job application""" try: # Try MCP server response = self._call_mcp_tool("update_row", { "file_path": self.tracker_path, "sheet": "Applications", "job_id": job_id, "status": status, "notes": notes, "interview_date": interview_date, "last_updated": datetime.now().strftime("%Y-%m-%d") }) return response.get("success", False) except Exception as e: logger.error(f"Error updating application status: {e}") return False def _use_mcp_server(self, results: List[OrchestrationResult]) -> bool: """Try to use MCP server for Excel generation""" try: # Create workbook response = self._call_mcp_tool("create_workbook", {}) if not response.get("success"): return False # Create sheets sheets = [ "Applications", "Keywords", "Companies", "Statistics", "Timeline" ] for sheet in sheets: self._call_mcp_tool("add_sheet", {"name": sheet}) # Applications sheet self._setup_applications_sheet(results) # Keywords sheet self._setup_keywords_sheet(results) # Companies sheet self._setup_companies_sheet(results) # Statistics sheet self._setup_statistics_sheet(results) # Timeline sheet self._setup_timeline_sheet(results) # Save workbook self._call_mcp_tool("save_workbook", { "file_path": self.tracker_path }) return True except Exception as e: logger.error(f"MCP server error: {e}") return False def _setup_applications_sheet(self, results: List[OrchestrationResult]): """Setup main applications tracking sheet""" # Headers headers = [ "Job ID", "Company", "Position", "Location", "Date Applied", "Status", "Match Score", "Salary Range", "Resume Version", "Cover Letter", "Keywords Matched", "Interview Date", "Notes", "URL", "Last Updated" ] self._call_mcp_tool("add_headers", { "sheet": "Applications", "headers": headers }) # Add data for i, result in enumerate(results): job = result.job resume = result.resume # Calculate match score match_score = self._calculate_match_score(job, resume) # Extract keywords matched keywords_matched = ", ".join(result.keywords[:10]) if hasattr(result, 'keywords') else "" row_data = [ job.id or f"JOB_{i+1}", job.company, job.title, job.location or "Remote", datetime.now().strftime("%Y-%m-%d"), "Applied", f"{match_score}%", job.salary or "Not specified", f"v{i+1}", "Yes", keywords_matched, "", # Interview date "", # Notes job.url or "", datetime.now().strftime("%Y-%m-%d %H:%M") ] self._call_mcp_tool("add_row", { "sheet": "Applications", "data": row_data }) # Add formatting self._call_mcp_tool("format_table", { "sheet": "Applications", "style": "professional", "freeze_panes": "A2", "auto_filter": True }) def _setup_keywords_sheet(self, results: List[OrchestrationResult]): """Setup keywords analysis sheet""" headers = ["Keyword", "Frequency", "Companies", "Positions"] self._call_mcp_tool("add_headers", { "sheet": "Keywords", "headers": headers }) # Aggregate keywords keyword_data = {} for result in results: if hasattr(result, 'keywords'): for keyword in result.keywords: if keyword not in keyword_data: keyword_data[keyword] = { "frequency": 0, "companies": set(), "positions": set() } keyword_data[keyword]["frequency"] += 1 keyword_data[keyword]["companies"].add(result.job.company) keyword_data[keyword]["positions"].add(result.job.title) # Add keyword rows for keyword, data in sorted(keyword_data.items(), key=lambda x: x[1]["frequency"], reverse=True): row_data = [ keyword, data["frequency"], ", ".join(list(data["companies"])[:5]), ", ".join(list(data["positions"])[:5]) ] self._call_mcp_tool("add_row", { "sheet": "Keywords", "data": row_data }) # Add chart self._call_mcp_tool("add_chart", { "sheet": "Keywords", "chart_type": "bar", "title": "Top Keywords", "data_range": "A1:B21" # Top 20 keywords }) def _setup_companies_sheet(self, results: List[OrchestrationResult]): """Setup companies overview sheet""" headers = ["Company", "Positions Applied", "Average Match", "Status", "Notes"] self._call_mcp_tool("add_headers", { "sheet": "Companies", "headers": headers }) # Aggregate by company company_data = {} for result in results: company = result.job.company if company not in company_data: company_data[company] = { "positions": [], "matches": [] } company_data[company]["positions"].append(result.job.title) match_score = self._calculate_match_score(result.job, result.resume) company_data[company]["matches"].append(match_score) # Add company rows for company, data in company_data.items(): avg_match = sum(data["matches"]) / len(data["matches"]) row_data = [ company, len(data["positions"]), f"{avg_match:.1f}%", "Active", ", ".join(data["positions"][:3]) ] self._call_mcp_tool("add_row", { "sheet": "Companies", "data": row_data }) def _setup_statistics_sheet(self, results: List[OrchestrationResult]): """Setup statistics dashboard sheet""" stats = [ ["Total Applications", len(results)], ["Unique Companies", len(set(r.job.company for r in results))], ["Average Match Score", f"{self._calculate_average_match(results):.1f}%"], ["Documents Generated", len(results) * 2], # Resume + Cover Letter ["Date Range", f"{datetime.now().strftime('%Y-%m-%d')}"], ] for stat in stats: self._call_mcp_tool("add_row", { "sheet": "Statistics", "data": stat }) # Add dashboard charts self._call_mcp_tool("add_chart", { "sheet": "Statistics", "chart_type": "pie", "title": "Application Status", "data_range": "A1:B5" }) def _setup_timeline_sheet(self, results: List[OrchestrationResult]): """Setup application timeline sheet""" headers = ["Date", "Company", "Position", "Action", "Status"] self._call_mcp_tool("add_headers", { "sheet": "Timeline", "headers": headers }) # Add timeline entries for result in results: row_data = [ datetime.now().strftime("%Y-%m-%d"), result.job.company, result.job.title, "Applied", "Pending" ] self._call_mcp_tool("add_row", { "sheet": "Timeline", "data": row_data }) def _create_with_openpyxl(self, results: List[OrchestrationResult]) -> str: """Create tracker using openpyxl as fallback""" try: from openpyxl import Workbook from openpyxl.styles import Font, PatternFill, Alignment from openpyxl.utils import get_column_letter wb = Workbook() # Applications sheet ws = wb.active ws.title = "Applications" # Headers headers = [ "Job ID", "Company", "Position", "Location", "Date Applied", "Status", "Match Score", "Salary", "Resume", "Cover Letter", "Keywords", "Interview", "Notes", "URL", "Updated" ] for col, header in enumerate(headers, 1): cell = ws.cell(row=1, column=col, value=header) cell.font = Font(bold=True) cell.fill = PatternFill(start_color="366092", end_color="366092", fill_type="solid") cell.font = Font(color="FFFFFF", bold=True) # Add data for row, result in enumerate(results, 2): job = result.job ws.cell(row=row, column=1, value=job.id or f"JOB_{row-1}") ws.cell(row=row, column=2, value=job.company) ws.cell(row=row, column=3, value=job.title) ws.cell(row=row, column=4, value=job.location or "Remote") ws.cell(row=row, column=5, value=datetime.now().strftime("%Y-%m-%d")) ws.cell(row=row, column=6, value="Applied") ws.cell(row=row, column=7, value=f"{self._calculate_match_score(job, result.resume)}%") ws.cell(row=row, column=8, value=job.salary or "Not specified") ws.cell(row=row, column=9, value=f"v{row-1}") ws.cell(row=row, column=10, value="Yes") ws.cell(row=row, column=14, value=job.url or "") ws.cell(row=row, column=15, value=datetime.now().strftime("%Y-%m-%d %H:%M")) # Auto-adjust column widths for column in ws.columns: max_length = 0 column_letter = get_column_letter(column[0].column) for cell in column: try: if len(str(cell.value)) > max_length: max_length = len(str(cell.value)) except: pass adjusted_width = min(max_length + 2, 50) ws.column_dimensions[column_letter].width = adjusted_width # Save workbook wb.save(self.tracker_path) logger.info(f"Excel tracker created: {self.tracker_path}") return self.tracker_path except ImportError: logger.error("openpyxl not installed") return None def _calculate_match_score(self, job: JobPosting, resume: ResumeDraft) -> float: """Calculate job match score""" try: # Simple scoring based on keyword matches job_text = f"{job.title} {job.description}".lower() resume_text = resume.text.lower() matches = 0 total_keywords = 0 # Check for common keywords important_keywords = job_text.split()[:50] # Top 50 words for keyword in important_keywords: if len(keyword) > 3 and keyword in resume_text: matches += 1 total_keywords += 1 if total_keywords > 0: return round((matches / total_keywords) * 100, 1) return 50.0 except: return 50.0 def _calculate_average_match(self, results: List[OrchestrationResult]) -> float: """Calculate average match score across all applications""" if not results: return 0.0 scores = [self._calculate_match_score(r.job, r.resume) for r in results] return sum(scores) / len(scores) def _call_mcp_tool(self, tool_name: str, params: Dict[str, Any]) -> Dict[str, Any]: """Call MCP server tool""" try: response = requests.post( f"{self.mcp_server_url}/tools/{tool_name}", json=params, timeout=30 ) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: logger.error(f"MCP server call failed: {e}") return {"success": False, "error": str(e)}