import os import fitz # PyMuPDF from flask import Flask, render_template, request, redirect, url_for, flash, session from werkzeug.utils import secure_filename from flask_mail import Mail, Message from itsdangerous import URLSafeTimedSerializer from openai import OpenAI import uuid import requests import emaillogger from dotenv import load_dotenv import chromadb from chromadb.utils import embedding_functions from chromadb.config import Settings from datetime import datetime, timedelta load_dotenv() app = Flask(__name__) app.config['SECRET_KEY'] = os.getenv('SECRET_KEY') app.config['SECURITY_PASSWORD_SALT'] = os.getenv('SECURITY_PASSWORD_SALT') app.config['UPLOAD_FOLDER'] = 'static/uploads' app.config['ALLOWED_EXTENSIONS'] = {'pdf'} # Email configuration app.config['MAIL_SERVER'] = 'mail.digigee.com' app.config['MAIL_PORT'] = 587 app.config['MAIL_USE_TLS'] = True app.config['MAIL_USERNAME'] = os.getenv('MAIL_USERNAME') app.config['MAIL_PASSWORD'] = os.getenv('MAIL_PASSWORD') app.config['OPENAI_API_KEY'] = os.getenv('OPENAI_API_KEY') mail = Mail(app) email_logger = emaillogger.EmailLogger() def allowed_file(filename): return '.' in filename and filename.rsplit('.', 1)[1].lower() in app.config['ALLOWED_EXTENSIONS'] def generate_confirmation_token(email): serializer = URLSafeTimedSerializer(app.config['SECRET_KEY']) return serializer.dumps(email, salt=app.config['SECURITY_PASSWORD_SALT']) def confirm_token(token, expiration=3600): serializer = URLSafeTimedSerializer(app.config['SECRET_KEY']) try: email = serializer.loads( token, salt=app.config['SECURITY_PASSWORD_SALT'], max_age=expiration ) except: return False return email def send_confirmation_email(email, token): confirm_url = url_for('confirm_email', token=token, _external=True) html = render_template('confirm.html', confirm_url=confirm_url) msg = Message('Confirm Your CV Upload', sender=app.config['MAIL_USERNAME'], recipients=[email]) msg.html = html mail.send(msg) def summarize_pdf(file_path): # Create OpenAI client client = OpenAI(api_key=app.config['OPENAI_API_KEY']) # Extract text from PDF using fitz text = "" try: doc = fitz.open(file_path) for page in doc: text += page.get_text() doc.close() except Exception as e: print(f"Error reading PDF: {e}") return None # Call OpenAI API to summarize try: response = client.chat.completions.create( model="gpt-3.5-turbo", messages=[ {"role": "system", "content": "Summarize the key details of this CV in a concise paragraph. Focus on the candidate's work experience, highlighting any leadership roles and industry-specific skills. Include educational background and any relevant certifications or awards. Present the information in a clear and structured manner."}, {"role": "user", "content": f"Please summarize the following text:\n\n{text}"} ], max_tokens=500, temperature=0.7 ) return response.choices[0].message.content.strip() except Exception as e: print(f"Error calling OpenAI API: {e}") return None def get_matching_jobs(summary, max_distance=2.0): """ Find matching jobs from ChromaDB based on CV summary Returns top 3 matches with the lowest distance for jobs posted within last 5 days """ try: # Initialize ChromaDB client chroma_client = chromadb.Client( Settings( persist_directory="../chroma_db", is_persistent=True ) ) # Get the collection openai_ef = embedding_functions.OpenAIEmbeddingFunction( api_key=os.getenv('OPENAI_API_KEY'), model_name="text-embedding-3-small" ) collection = chroma_client.get_collection( name="job_postings", embedding_function=openai_ef ) # Get current date and date from days ahead current_date = datetime.now() days_ahead = current_date + timedelta(days=1) # Query the collection results = collection.query( query_texts=[summary], n_results=100 # Query more results initially to filter by date ) # Filter results by distance and date relevant_jobs = [] for i, (doc, metadata, distance) in enumerate(zip( results['documents'][0], results['metadatas'][0], results['distances'][0] )): # Try to convert metadata date string to datetime try: job_date = datetime.strptime(metadata['date'], '%d.%m.%Y.') except ValueError: print(f"Unable to parse date: {metadata['date']}") continue if job_date >= days_ahead: relevant_jobs.append((doc, metadata, distance)) print(f"Job {i+1}:") #print(f" Document: {doc}") print(f" Metadata: {metadata}") print(f" Distance: {distance}") print("==========================================") # Sort the relevant jobs by distance and select the top 3 relevant_jobs.sort(key=lambda x: x[2]) #relevant_jobs.sort(key=lambda x: abs(x[2])) top_matches = [ { 'title': job[1]['title'], 'company': job[1]['company'], 'date': job[1]['date'], 'skills': job[1]['tags'], 'link': job[1]['link'], 'distance': round(job[2], 4) } for job in relevant_jobs[:3] ] return top_matches except Exception as e: print(f"Error querying ChromaDB: {e}") return [] @app.route('/') def index(): return render_template('index.html') @app.route('/upload', methods=['GET', 'POST']) def upload_file(): if request.method == 'POST': if 'file' not in request.files: flash('No file part') return redirect(request.url) file = request.files['file'] if file.filename == '': flash('No selected file') return redirect(request.url) if file and allowed_file(file.filename): filename = secure_filename(file.filename) file_id = str(uuid.uuid4()) file_path = os.path.join(app.config['UPLOAD_FOLDER'], f"{file_id}_{filename}") file.save(file_path) email = request.form['email'] # Generate validation token token = generate_confirmation_token(request.form['email']) # Store file information in session session['pending_file'] = { 'id': file_id, 'name': filename, 'path': file_path, 'email': email } try: # Send validation email send_confirmation_email(email, token) # Log the sent authentication email email_logger.log_auth_email( email=email, auth_code=token, # You might want to store a different identifier sent_successfully=True ) return render_template('email_sent.html') except Exception as e: # Log failed attempt email_logger.log_auth_email( email=email, auth_code=token, sent_successfully=False ) flash('Error sending confirmation email. Please try again.', 'danger') return redirect(request.url) return render_template('upload.html') @app.route('/confirm/') def confirm_email(token): email = confirm_token(token) if not email: flash('The confirmation link is invalid or has expired.', 'danger') return redirect(url_for('index')) pending_file = session.get('pending_file') if not pending_file or pending_file['email'] != email: flash('No pending file found or email mismatch.', 'danger') return redirect(url_for('index')) try: # Mark the email as verified in our logs email_logger.mark_as_verified(email, token) # Get CV summary summary = summarize_pdf(pending_file['path']) # Get matching jobs matching_jobs = get_matching_jobs(summary) # Clear the pending file from session session.pop('pending_file', None) return render_template('summary.html', summary=summary, filename=pending_file['name'], matching_jobs=matching_jobs) except Exception as e: flash('Error processing file after confirmation.', 'danger') return redirect(url_for('index')) if __name__ == '__main__': app.run(debug=True)