Spaces:
Paused
Paused
import os | |
# Ensure Hugging Face writes cache to a safe writable location on Spaces | |
os.environ["HF_HOME"] = "/tmp/huggingface" | |
os.environ["TRANSFORMERS_CACHE"] = "/tmp/huggingface/transformers" | |
os.environ["HUGGINGFACE_HUB_CACHE"] = "/tmp/huggingface/hub" | |
from flask import Flask, render_template, redirect, url_for, flash, request | |
from flask_login import LoginManager, login_required, current_user | |
from werkzeug.utils import secure_filename | |
import os | |
import sys | |
import json | |
from datetime import datetime | |
# Adjust sys.path for import flexibility | |
current_dir = os.path.dirname(os.path.abspath(__file__)) | |
parent_dir = os.path.dirname(current_dir) | |
sys.path.append(parent_dir) | |
sys.path.append(current_dir) | |
# Import and initialize DB | |
from backend.models.database import db, Job, Application, init_db | |
from backend.models.user import User | |
from backend.routes.auth import auth_bp | |
from backend.routes.interview_api import interview_api | |
from backend.models.resume_parser.resume_to_features import extract_resume_features | |
# Initialize Flask app | |
app = Flask(__name__) | |
app.config['SECRET_KEY'] = 'your-secret-key' | |
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///codingo.db' | |
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False | |
# Initialize DB with app | |
init_db(app) | |
# Flask-Login setup | |
login_manager = LoginManager() | |
login_manager.login_view = 'auth.login' | |
login_manager.init_app(app) | |
def load_user(user_id): | |
return db.session.get(User, int(user_id)) | |
# Register blueprints | |
app.register_blueprint(auth_bp) | |
app.register_blueprint(interview_api, url_prefix="/api") | |
def handle_resume_upload(file): | |
""" | |
Save uploaded file temporarily, extract features, then clean up. | |
Returns (features_dict, error_message, filename) | |
""" | |
if not file or file.filename == '': | |
return None, "No file uploaded", None | |
try: | |
filename = secure_filename(file.filename) | |
filepath = os.path.join(current_dir, 'temp', filename) | |
os.makedirs(os.path.dirname(filepath), exist_ok=True) | |
file.save(filepath) | |
features = extract_resume_features(filepath) | |
os.remove(filepath) # Clean up after parsing | |
return features, None, filename | |
except Exception as e: | |
print(f"Error in handle_resume_upload: {e}") | |
return None, str(e), None | |
# Routes | |
def index(): | |
return render_template('index.html') | |
def jobs(): | |
all_jobs = Job.query.order_by(Job.date_posted.desc()).all() | |
return render_template('jobs.html', jobs=all_jobs) | |
def job_detail(job_id): | |
job = Job.query.get_or_404(job_id) | |
return render_template('job_detail.html', job=job) | |
def apply(job_id): | |
job = Job.query.get_or_404(job_id) | |
if request.method == 'POST': | |
file = request.files.get('resume') | |
features, error, _ = handle_resume_upload(file) | |
if error or not features: | |
flash("Resume parsing failed. Please try again.", "danger") | |
return render_template('apply.html', job=job) | |
application = Application( | |
job_id=job_id, | |
user_id=current_user.id, | |
name=current_user.username, | |
email=current_user.email, | |
extracted_features=json.dumps(features) | |
) | |
db.session.add(application) | |
db.session.commit() | |
flash('Your application has been submitted successfully!', 'success') | |
return redirect(url_for('jobs')) | |
return render_template('apply.html', job=job) | |
def my_applications(): | |
"""View user's applications""" | |
applications = Application.query.filter_by(user_id=current_user.id).order_by(Application.date_applied.desc()).all() | |
return render_template('my_applications.html', applications=applications) | |
def parse_resume(): | |
"""API endpoint for parsing resume and returning features""" | |
file = request.files.get('resume') | |
features, error, _ = handle_resume_upload(file) | |
if error: | |
print(f"[Resume Error] {error}") | |
return {"error": "Error parsing resume. Please try again."}, 400 | |
if not features: | |
print("[Resume Error] No features extracted.") | |
return {"error": "Failed to extract resume details."}, 400 | |
response = { | |
"name": features.get('name', ''), | |
"email": features.get('email', ''), | |
"mobile_number": features.get('mobile_number', ''), | |
"skills": features.get('skills', []), | |
"experience": features.get('experience', []), | |
"education": features.get('education', []), | |
"summary": features.get('summary', '') | |
} | |
return response, 200 | |
def interview_page(job_id): | |
job = Job.query.get_or_404(job_id) | |
application = Application.query.filter_by(user_id=current_user.id, job_id=job_id).first() | |
if not application or not application.extracted_features: | |
flash("Please apply for this job and upload your resume first.", "warning") | |
return redirect(url_for('job_detail', job_id=job_id)) | |
cv_data = json.loads(application.extracted_features) | |
return render_template("interview.html", job=job, cv=cv_data) | |
if __name__ == '__main__': | |
print("Starting Codingo application...") | |
with app.app_context(): | |
db.create_all() | |
app.run(debug=True, port=7860) |