""" Web Interface for Job Application AI Agent This module provides a Flask web application for the job application AI agent. """ import os import logging import json import uuid import time import sys import importlib from datetime import datetime from flask import Flask, render_template, request, redirect, url_for, flash, send_file, session, jsonify import pandas as pd import zipfile import io from job_apply_ai.scraper.linkedin import LinkedInScraper from job_apply_ai.cv_modifier.cv_analyzer import CVAnalyzer, CVModifier, batch_process_jobs from job_apply_ai.utils.helpers import ensure_directory_exists # Configure logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) # Initialize Flask app app = Flask(__name__) app.secret_key = os.environ.get('SECRET_KEY', 'dev_key_for_testing') # Use a per-run session cookie name so stale browser sessions don't leak old state into new runs. app.config['SESSION_COOKIE_NAME'] = f"job_apply_ai_session_{int(time.time())}" # Keep runtime files local to the project unless overridden by env var. runtime_root = os.environ.get('JOB_APPLY_AI_DATA_DIR', os.path.join(os.getcwd(), '.runtime')) if not os.path.isabs(runtime_root): runtime_root = os.path.abspath(runtime_root) app.config['UPLOAD_FOLDER'] = os.path.join(runtime_root, 'uploads') ensure_directory_exists(app.config['UPLOAD_FOLDER']) # Create output directories app.config['CV_OUTPUT_DIR'] = os.path.join(app.config['UPLOAD_FOLDER'], 'cvs') app.config['JOBS_OUTPUT_DIR'] = os.path.join(app.config['UPLOAD_FOLDER'], 'jobs') app.config['STATE_DIR'] = os.path.join(app.config['UPLOAD_FOLDER'], 'session_state') ensure_directory_exists(app.config['CV_OUTPUT_DIR']) ensure_directory_exists(app.config['JOBS_OUTPUT_DIR']) ensure_directory_exists(app.config['STATE_DIR']) # Ensure session data is saved app.config['SESSION_TYPE'] = 'filesystem' SUPPORTED_TAILORING_MODES = {'local', 'api'} def _session_state_path(state_id): return os.path.join(app.config['STATE_DIR'], f"{state_id}.json") def _get_tailoring_mode(): """Resolve active tailoring mode from session, then environment.""" mode = (session.get('tailoring_mode') or '').strip().lower() if mode in SUPPORTED_TAILORING_MODES: return mode env_mode = (os.environ.get('CV_TAILORING_MODE', 'local') or 'local').strip().lower() if env_mode not in SUPPORTED_TAILORING_MODES: env_mode = 'local' return env_mode def _set_tailoring_mode(mode): mode = (mode or '').strip().lower() if mode in SUPPORTED_TAILORING_MODES: session['tailoring_mode'] = mode return mode return _get_tailoring_mode() def _clear_job_context(keep_cv_template=True): """Clear prior search/CV generation state for a fresh workflow.""" state_id = session.pop('jobs_state_id', None) if state_id: state_path = _session_state_path(state_id) if os.path.exists(state_path): try: os.remove(state_path) except OSError: pass for key in [ 'jobs_file', 'excel_filename', 'generated_cvs', 'successful_jobs', 'failed_jobs', 'current_cv', 'current_cv_filename', ]: session.pop(key, None) if not keep_cv_template: session.pop('cv_template', None) def _save_processed_jobs(processed_jobs): state_id = str(uuid.uuid4()) state_path = _session_state_path(state_id) with open(state_path, 'w', encoding='utf-8') as f: json.dump(processed_jobs, f, ensure_ascii=False) session['jobs_state_id'] = state_id return state_id def _load_processed_jobs(): state_id = session.get('jobs_state_id') if not state_id: return [] state_path = _session_state_path(state_id) if not os.path.exists(state_path): return [] try: with open(state_path, 'r', encoding='utf-8') as f: data = json.load(f) return data if isinstance(data, list) else [] except Exception as e: logger.error(f"Failed to load session job state: {str(e)}") return [] def _update_processed_job(job_id, updated_job): jobs = _load_processed_jobs() if 0 <= job_id < len(jobs): jobs[job_id] = updated_job _save_processed_jobs(jobs) def _build_professional_summary(job, matched_categories): """Build a concise, professional summary tailored to role and extracted skills.""" job_title = (job.get('title') or 'Professional').strip() company = (job.get('company') or 'your target company').strip() flat_skills = [] for skills in (matched_categories or {}).values(): for s in skills or []: skill = str(s).strip() if skill: flat_skills.append(skill) # Keep unique order and top priority skills only. deduped = [] seen = set() for skill in flat_skills: key = skill.lower() if key not in seen: seen.add(key) deduped.append(skill) top_skills = deduped[:5] if top_skills: skills_text = ", ".join(top_skills) return ( f"{job_title} professional with hands-on experience in {skills_text}. " f"Delivers reliable, scalable outcomes through strong collaboration, ownership, " f"and structured problem-solving. Ready to contribute immediate impact at {company}." ) return ( f"{job_title} professional focused on delivering measurable outcomes through " f"technical execution, collaboration, and continuous improvement. " f"Motivated to contribute meaningful impact at {company}." ) def _generate_cv_with_api_tailoring(job, cv_template, output_path): """Use API subproject updater to generate a tailored CV from the same UI flow.""" api_project_root = os.path.join(os.getcwd(), "Automatic CV and Cover Letter with API") if not os.path.exists(api_project_root): raise FileNotFoundError("API subproject folder not found: Automatic CV and Cover Letter with API") if api_project_root not in sys.path: sys.path.append(api_project_root) APIIntegration = importlib.import_module('src.utils.openai_integration').OpenAIIntegration DocumentUpdater = importlib.import_module('src.updaters.document_updater').DocumentUpdater cover_letter_template = os.environ.get( 'API_COVER_LETTER_TEMPLATE_PATH', os.path.join(api_project_root, 'data', 'Cover Letter_Imon .docx') ) if not os.path.isabs(cover_letter_template): cover_letter_template = os.path.abspath(cover_letter_template) if not os.path.exists(cover_letter_template): raise FileNotFoundError( f"Cover letter template not found for API mode: {cover_letter_template}. " "Set API_COVER_LETTER_TEMPLATE_PATH in .env." ) description = (job.get('description') or '').strip() if not description: raise ValueError("Job description is empty; cannot run API tailoring mode") llm_integration = APIIntegration() provider = getattr(llm_integration, 'provider', 'unknown') if provider in ('openai', 'grok', 'groq') and not llm_integration.is_api_key_set(): raise ValueError( f"API key not configured for provider '{provider}'. " "Set provider key in .env or notebook config." ) updater = DocumentUpdater(cv_template, cover_letter_template, llm_integration) updater.update_cv(description, output_path) return provider @app.route('/') def index(): """Render the home page.""" return render_template( 'index.html', tailoring_mode=_get_tailoring_mode(), llm_provider=(os.environ.get('LLM_PROVIDER', 'ollama') or 'ollama').lower() ) @app.route('/set_tailoring_mode', methods=['POST']) def set_tailoring_mode(): """Set tailoring mode from UI without restarting the server.""" selected_mode = request.form.get('tailoring_mode', 'local') mode = _set_tailoring_mode(selected_mode) flash(f"Tailoring mode set to: {mode}", 'info') return redirect(url_for('index')) @app.route('/search', methods=['GET', 'POST']) def search_jobs(): """Handle job search form and display results.""" if request.method == 'POST': _set_tailoring_mode(request.form.get('tailoring_mode', _get_tailoring_mode())) keyword = request.form.get('keyword', '') location = request.form.get('location', '') max_jobs = int(request.form.get('max_jobs', 10)) if not keyword or not location: flash('Please enter both job title and location', 'error') return redirect(url_for('index')) try: # Scrape jobs scraper = LinkedInScraper(headless=True) jobs = scraper.scrape_job_listings(keyword, location, max_jobs=max_jobs) if not jobs: flash('No jobs found. Try different search terms.', 'warning') return redirect(url_for('index')) # Fetch job descriptions for i, job in enumerate(jobs): logger.info(f"Fetching description for job {i+1}/{len(jobs)}: {job['title']}") title, company, description = scraper.fetch_job_description(job['link']) jobs[i]['description'] = description # Save to Excel for reference today_date = datetime.today().strftime("%Y-%m-%d") filename = f"linkedin_jobs_{today_date}.xlsx" filepath = os.path.join(app.config['JOBS_OUTPUT_DIR'], filename) df = pd.DataFrame(jobs) df.to_excel(filepath, index=False) session['jobs_file'] = filepath session['excel_filename'] = filename # Process job descriptions to extract skills analyzer = CVAnalyzer() processed_jobs = [] for job in jobs: if job.get('description'): matched_skills, matched_requirements, matched_categories = analyzer.extract_skills_from_description(job['description']) job['matched_skills'] = matched_skills job['matched_categories'] = matched_categories processed_jobs.append(job) _save_processed_jobs(processed_jobs) return render_template('job_list.html', jobs=processed_jobs, excel_file=filename, excel_path=filepath, tailoring_mode=_get_tailoring_mode(), llm_provider=(os.environ.get('LLM_PROVIDER', 'ollama') or 'ollama').lower()) except Exception as e: logger.error(f"Error during job search: {str(e)}") flash(f'An error occurred: {str(e)}', 'error') return redirect(url_for('index')) return redirect(url_for('index')) @app.route('/upload_cv', methods=['GET', 'POST']) def upload_cv(): """Handle CV template upload.""" if request.method == 'POST': _set_tailoring_mode(request.form.get('tailoring_mode', _get_tailoring_mode())) if 'cv_file' not in request.files: flash('No file part', 'error') return redirect(request.url) file = request.files['cv_file'] if file.filename == '': flash('No selected file', 'error') return redirect(request.url) if file and file.filename.lower().endswith('.docx'): # Starting with a new CV should reset previous job selections/results. _clear_job_context(keep_cv_template=False) filename = os.path.join(app.config['UPLOAD_FOLDER'], 'cv_template.docx') file.save(filename) session['cv_template'] = filename flash('CV template uploaded successfully', 'success') # Always return to home so user can enter fresh job query/location/max-jobs. return redirect(url_for('index')) else: flash('Please upload a .docx file', 'error') return redirect(request.url) return render_template('upload_cv.html') @app.route('/job_list') def job_list(): """Display the list of jobs with Make CV buttons.""" processed_jobs = _load_processed_jobs() excel_filename = session.get('excel_filename') excel_path = session.get('jobs_file') if not processed_jobs: flash('No jobs found. Please search for jobs first.', 'warning') return redirect(url_for('index')) return render_template('job_list.html', jobs=processed_jobs, excel_file=excel_filename, excel_path=excel_path, tailoring_mode=_get_tailoring_mode(), llm_provider=(os.environ.get('LLM_PROVIDER', 'ollama') or 'ollama').lower()) @app.route('/download_excel') def download_excel(): """Download the Excel file with job listings.""" jobs_file = session.get('jobs_file') if jobs_file and not os.path.isabs(jobs_file): jobs_file = os.path.abspath(jobs_file) if not jobs_file or not os.path.exists(jobs_file): flash('Excel file not found', 'error') return redirect(url_for('index')) return send_file(jobs_file, as_attachment=True) @app.route('/make_cv/') def make_cv(job_id): """Generate a CV for a specific job.""" processed_jobs = _load_processed_jobs() cv_template = session.get('cv_template') if not processed_jobs or job_id >= len(processed_jobs): flash('Job not found', 'error') return redirect(url_for('job_list')) if not cv_template: flash('Please upload a CV template first', 'error') return redirect(url_for('upload_cv')) job = processed_jobs[job_id] tailoring_mode = _get_tailoring_mode() try: today_date = datetime.today().strftime("%Y-%m-%d") safe_company = job['company'].replace(' ', '_') safe_title = job['title'].replace(' ', '_') output_filename = f"CV_{today_date}_{safe_company}_{safe_title}.docx" output_path = os.path.join(app.config['CV_OUTPUT_DIR'], output_filename) if tailoring_mode == 'api': provider = _generate_cv_with_api_tailoring(job, cv_template, output_path) session['current_cv'] = output_path session['current_cv_filename'] = output_filename flash(f"CV generated successfully using API mode ({provider})", 'success') return render_template('cv_success.html', job=job, cv_filename=output_filename, matched_categories=job.get('matched_categories', {})) # Create CV modifier modifier = CVModifier(cv_template) # Get matched categories matched_categories = job.get('matched_categories', {}) if not matched_categories: # Try to extract skills again with the job title for context analyzer = CVAnalyzer() description = job.get('description', '') if description: # Add job title to the document's user_data for better skill extraction doc = analyzer.nlp(description) doc.user_data['job_title'] = job.get('title', '') matched_skills, _, matched_categories = analyzer.extract_skills_from_description(description) if not matched_skills: # If still no skills, add some generic ones based on job title job_title = job.get('title', '').lower() logger.warning(f"No skills found for job: {job_title}") # Add some generic skills matched_categories = { "Soft Skills": ["communication", "teamwork", "problem solving", "time management", "adaptability"] } # Try to add some technical skills based on job title keywords if any(kw in job_title for kw in ["developer", "engineer", "programmer"]): matched_categories["Programming Languages"] = ["python", "javascript", "java"] elif any(kw in job_title for kw in ["data", "analyst", "analytics"]): matched_categories["Business & Analytics"] = ["excel", "sql", "data analysis"] elif any(kw in job_title for kw in ["manager", "lead", "director"]): matched_categories["Methodologies"] = ["agile", "scrum", "project management"] elif any(kw in job_title for kw in ["designer", "ux", "ui"]): matched_categories["Tools & Platforms"] = ["figma", "adobe", "sketch"] flash('No specific skills found in job description. Adding generic skills based on job title.', 'warning') # Update skills section if modifier.update_skills_section(matched_categories): if os.environ.get('CV_ENABLE_SUMMARY_TAILORING', '1') == '1': summary_text = _build_professional_summary(job, matched_categories) modifier.update_profile_summary(summary_text) # Save the modified CV if modifier.save_modified_cv(output_path): # Store the path for download session['current_cv'] = output_path session['current_cv_filename'] = output_filename # Update the job with the matched categories if they were generated here job['matched_categories'] = matched_categories _update_processed_job(job_id, job) flash('CV generated successfully', 'success') return render_template('cv_success.html', job=job, cv_filename=output_filename, matched_categories=matched_categories) flash('Failed to generate CV. Please check if your CV template has a skills section.', 'error') return redirect(url_for('job_list')) except Exception as e: logger.error(f"Error generating CV: {str(e)}") flash(f'An error occurred: {str(e)}', 'error') return redirect(url_for('job_list')) @app.route('/download_cv') def download_cv(): """Download the current CV.""" cv_path = session.get('current_cv') if cv_path and not os.path.isabs(cv_path): cv_path = os.path.abspath(cv_path) if not cv_path or not os.path.exists(cv_path): flash('CV file not found', 'error') return redirect(url_for('job_list')) return send_file(cv_path, as_attachment=True) @app.route('/make_all_cvs') def make_all_cvs(): """Generate CVs for all jobs.""" processed_jobs = _load_processed_jobs() cv_template = session.get('cv_template') if not processed_jobs: flash('No jobs found. Please search for jobs first.', 'error') return redirect(url_for('index')) if not cv_template: flash('Please upload a CV template first', 'error') return redirect(url_for('upload_cv')) try: tailoring_mode = _get_tailoring_mode() # Generate CVs for all jobs successful_jobs = [] failed_jobs = [] generated_cvs = [] # Create analyzer for re-extracting skills if needed analyzer = CVAnalyzer() for job in processed_jobs: matched_categories = job.get('matched_categories', {}) # If no skills matched, try to extract them again if not matched_categories: description = job.get('description', '') if description: # Add job title to the document's user_data for better skill extraction doc = analyzer.nlp(description) doc.user_data['job_title'] = job.get('title', '') matched_skills, _, matched_categories = analyzer.extract_skills_from_description(description) job['matched_categories'] = matched_categories today_date = datetime.today().strftime("%Y-%m-%d") safe_company = job['company'].replace(' ', '_') safe_title = job['title'].replace(' ', '_') output_filename = f"CV_{today_date}_{safe_company}_{safe_title}.docx" output_path = os.path.join(app.config['CV_OUTPUT_DIR'], output_filename) if tailoring_mode == 'api': try: _generate_cv_with_api_tailoring(job, cv_template, output_path) generated_cvs.append(output_path) successful_jobs.append(job) except Exception as api_error: logger.error(f"API mode failed for job '{job.get('title', '')}': {api_error}") failed_jobs.append(job) continue # Local mode modifier = CVModifier(cv_template) if modifier.update_skills_section(matched_categories): if os.environ.get('CV_ENABLE_SUMMARY_TAILORING', '1') == '1': summary_text = _build_professional_summary(job, matched_categories) modifier.update_profile_summary(summary_text) if modifier.save_modified_cv(output_path): generated_cvs.append(output_path) successful_jobs.append(job) else: failed_jobs.append(job) else: failed_jobs.append(job) if generated_cvs: session['generated_cvs'] = generated_cvs session['successful_jobs'] = successful_jobs session['failed_jobs'] = failed_jobs flash(f'Successfully generated {len(generated_cvs)} CVs', 'success') if failed_jobs: flash(f'Failed to generate {len(failed_jobs)} CVs', 'warning') return render_template('all_cvs_success.html', successful_jobs=successful_jobs, failed_jobs=failed_jobs) else: flash('No CVs were generated. Make sure your CV template has a skills section.', 'warning') return redirect(url_for('job_list')) except Exception as e: logger.error(f"Error generating CVs: {str(e)}") flash(f'An error occurred: {str(e)}', 'error') return redirect(url_for('job_list')) @app.route('/download_all_cvs') def download_all_cvs(): """Download all generated CVs as a zip file.""" generated_cvs = session.get('generated_cvs', []) generated_cvs = [os.path.abspath(p) if p and not os.path.isabs(p) else p for p in generated_cvs] if not generated_cvs: flash('No generated CVs available', 'error') return redirect(url_for('job_list')) # Create a zip file in memory memory_file = io.BytesIO() with zipfile.ZipFile(memory_file, 'w') as zf: for cv_path in generated_cvs: if os.path.exists(cv_path): # Add file to zip with just the filename (not the full path) zf.write(cv_path, os.path.basename(cv_path)) # Reset file pointer memory_file.seek(0) # Create a date-stamped filename for the zip today_date = datetime.today().strftime("%Y-%m-%d") zip_filename = f"All_CVs_{today_date}.zip" return send_file( memory_file, mimetype='application/zip', as_attachment=True, download_name=zip_filename ) @app.errorhandler(404) def page_not_found(e): """Handle 404 errors.""" return render_template('404.html'), 404 @app.errorhandler(500) def server_error(e): """Handle 500 errors.""" logger.error(f"Server error: {str(e)}") return render_template('500.html'), 500 def main(): """Run the Flask application.""" # Create templates directory if it doesn't exist templates_dir = os.path.join(os.path.dirname(__file__), 'templates') ensure_directory_exists(templates_dir) # Create basic templates if they don't exist create_basic_templates(templates_dir) # Run the app app.run(debug=True, host='0.0.0.0', port=5050) def create_basic_templates(templates_dir): """Create basic HTML templates if they don't exist.""" templates = { 'index.html': ''' Job Application AI Agent

Job Application AI Agent

Search for jobs and generate tailored CVs


{% with messages = get_flashed_messages(with_categories=true) %} {% if messages %} {% for category, message in messages %}
{{ message }}
{% endfor %} {% endif %} {% endwith %}
Step 1: Upload CV Template

First, upload your CV template (.docx format)

Upload CV Template
Step 2: Search for Jobs
''', 'upload_cv.html': ''' Upload CV Template

Upload CV Template

{% with messages = get_flashed_messages(with_categories=true) %} {% if messages %} {% for category, message in messages %}
{{ message }}
{% endfor %} {% endif %} {% endwith %} Back to Home
Upload Your CV Template (.docx)
Please upload a Microsoft Word (.docx) file.
''', 'job_list.html': ''' Job Listings

Job Listings

{% with messages = get_flashed_messages(with_categories=true) %} {% if messages %} {% for category, message in messages %}
{{ message }}
{% endfor %} {% endif %} {% endwith %}
Back to Home
{% if excel_file %} Download Excel {% endif %} {% if jobs %} Generate All CVs {% endif %}

Found {{ jobs|length }} Jobs

Click "Make CV" to generate a tailored CV for a specific job.

{% for job in jobs %}
{{ job.title }}
{{ job.company }}
{% if job.matched_skills %}
Matched Skills:
{% for skill in job.matched_skills %} {{ skill }} {% endfor %}
{% endif %}
{% endfor %}
''', 'cv_success.html': ''' CV Generated Successfully

Success!

Your CV has been tailored for the position of {{ job.title }} at {{ job.company }}.

{% with messages = get_flashed_messages(with_categories=true) %} {% if messages %} {% for category, message in messages %}
{{ message }}
{% endfor %} {% endif %} {% endwith %}
Skills Added to Your CV
{% if matched_categories %} {% for category, skills in matched_categories.items() %}
{{ category }}

{{ skills|join(', ') }}

{% endfor %} {% else %}

No specific skills matched.

{% endif %}
''', 'all_cvs_success.html': ''' All CVs Generated

Success!

Successfully generated {{ cv_count }} tailored CVs.

{% with messages = get_flashed_messages(with_categories=true) %} {% if messages %} {% for category, message in messages %}
{{ message }}
{% endfor %} {% endif %} {% endwith %}
Download Options

You can download all generated CVs as a ZIP file.

Download All CVs (ZIP)
Back to Job List
''', '404.html': ''' Page Not Found

404

Page Not Found

The page you are looking for does not exist.

Go Home
''', '500.html': ''' Server Error

500

Server Error

Something went wrong on our end. Please try again later.

Go Home
''' } for filename, content in templates.items(): filepath = os.path.join(templates_dir, filename) if not os.path.exists(filepath): with open(filepath, 'w') as f: f.write(content) logger.info(f"Created template: {filename}") # Add template filter to get basename from path @app.template_filter('basename') def basename_filter(path): return os.path.basename(path) if __name__ == '__main__': main()