from typing import Optional from fastapi import FastAPI, Form, Request,Depends from fastapi.responses import HTMLResponse, RedirectResponse,StreamingResponse from fastapi.templating import Jinja2Templates import mysql.connector from mysql.connector import Error from fastapi.staticfiles import StaticFiles from starlette.middleware.sessions import SessionMiddleware import csv import io from datetime import datetime, timedelta from fastapi.responses import JSONResponse from dotenv import load_dotenv import os from passlib.hash import bcrypt from fastapi import HTTPException from starlette.middleware.sessions import SessionMiddleware from fastapi import Depends load_dotenv() app = FastAPI() app.add_middleware(SessionMiddleware, secret_key=os.getenv("SESSION_SECRET")) templates = Jinja2Templates(directory="templates") DB_CONFIG = { 'host': os.getenv("HOST"), 'user': os.getenv("USER"), 'password': os.getenv("PASSWORD"), 'database': os.getenv("DATABASE") } print(DB_CONFIG) def require_login(request: Request): if not request.session or "user" not in request.session: raise HTTPException(status_code=302, detail="Not Authorized") # or 401 if you prefer return request.session["user"] def insert_query(job_titles: str, locations: str, after_year: Optional[int], before_year: Optional[int], query_text: str): try: connection = mysql.connector.connect(**DB_CONFIG) cursor = connection.cursor() sql = """ INSERT INTO resume_queries (job_titles, locations, after_year, before_year,query_text) VALUES (%s, %s, %s, %s, %s) """ cursor.execute(sql, (job_titles, locations, after_year, before_year,query_text)) connection.commit() cursor.close() connection.close() return True except Error as e: print(f"❌ MySQL Error: {e}") return False @app.get("/", response_class=HTMLResponse) def form_page(request: Request, message: Optional[str] = None, user: dict = Depends(require_login)): return templates.TemplateResponse("form.html", {"request": request, "message": message}) @app.post("/submit") def submit_form( request: Request, job_titles: Optional[str] = Form(None), locations: Optional[str] = Form(None), after_year: Optional[str] = Form(None), before_year: Optional[str] = Form(None), boolean_query: Optional[str] = Form(None) ): after_year_int = int(after_year) if after_year and after_year.isdigit() else None before_year_int = int(before_year) if before_year and before_year.isdigit() else None # Build query text if boolean_query and boolean_query.strip(): query_text = boolean_query.strip() else: base_query = "(intitle:resume OR inurl:resume)" job_part = ' OR '.join([f'"{title.strip()}"' for title in job_titles.split(',')]) location_part = ' OR '.join([f'"{loc.strip()}"' for loc in locations.split(',')]) query_parts = [ base_query, f"({job_part})", f"({location_part})", "-job -jobs -sample -samples -templates" ] if after_year_int: query_parts.append(f"after:{after_year_int}") if before_year_int: query_parts.append(f"before:{before_year_int}") query_text = ' '.join(query_parts) # Insert with query text success = insert_query(job_titles, locations, after_year_int, before_year_int, query_text) if success: return RedirectResponse(url="/?message=Data+added+successfully.+Scraping+will+start+soon.", status_code=302) return {"error": "Failed to insert into database."} def get_connection(): return mysql.connector.connect(**DB_CONFIG) def parse_filter(filter_text, column_name, params): """ Parse the input filter for AND/OR logic and build SQL parts for the column (job_titles, locations, etc.) """ filter_parts = [part.strip() for part in filter_text.split("AND")] filter_filters = [] for part in filter_parts: # Handle the OR condition in each part or_parts = [or_part.strip() for or_part in part.split("OR")] or_filters = [] for or_part in or_parts: or_filters.append(f"LOWER({column_name}) LIKE %s") params.append(f"%{or_part.lower()}%") # Join OR filters for this part filter_filters.append(f"({' OR '.join(or_filters)})") return f"({' AND '.join(filter_filters)})" # Join AND filters for the full input @app.get("/api/results") def get_results_api(request: Request, job_title: str = '', location: str = '', start_date: str = '', end_date: str = '', draw: int = 1, start: int = 0, length: int = 10, user: dict = Depends(require_login)): conn = get_connection() cursor = conn.cursor(dictionary=True) print("Job Title:", job_title) print("Location:", location) filters = [] params = [] # Parse job_title if it's present if job_title: filters.append(parse_filter(job_title, "rq.job_titles", params)) # Parse location if it's present if location: filters.append(parse_filter(location, "rq.locations", params)) # Handle date filters if start_date: filters.append("rq.created_at >= %s") params.append(start_date) if end_date: end_datetime = datetime.strptime(end_date, "%Y-%m-%d") + timedelta(days=1) filters.append("rq.created_at < %s") params.append(end_datetime.strftime("%Y-%m-%d")) where_clause = f"WHERE {' AND '.join(filters)}" if filters else "" # Total count query cursor.execute(f""" SELECT COUNT(*) as total FROM google_search_results gr JOIN resume_queries rq ON gr.resume_query_id = rq.id {where_clause} """, tuple(params)) total = cursor.fetchone()["total"] # Paginated query if length isn't -1 if length == -1: query = f""" SELECT gr.id, gr.url, gr.email, gr.page_number, gr.status, gr.retry, gr.resume_query_id, rq.job_titles, rq.locations, rq.created_at FROM google_search_results gr JOIN resume_queries rq ON gr.resume_query_id = rq.id {where_clause} ORDER BY gr.resume_query_id DESC """ cursor.execute(query, tuple(params)) rows = cursor.fetchall() else: cursor.execute(f""" SELECT gr.id, gr.url, gr.email, gr.page_number, gr.status, gr.retry, gr.resume_query_id, rq.job_titles, rq.locations, rq.created_at FROM google_search_results gr JOIN resume_queries rq ON gr.resume_query_id = rq.id {where_clause} ORDER BY gr.resume_query_id DESC LIMIT %s OFFSET %s """, (*params, length, start)) rows = cursor.fetchall() # Format datetime fields for row in rows: if isinstance(row.get("created_at"), datetime): row["created_at"] = row["created_at"].strftime("%Y-%m-%d") cursor.close() conn.close() return JSONResponse({ "draw": draw, "recordsTotal": total, "recordsFiltered": total, "data": rows, "filters": { "job_title": job_title, "location": location, "start_date": start_date, "end_date": end_date } }) @app.get("/results", response_class=HTMLResponse) def show_results(request: Request, job_title: str = '', location: str = '', resume_query_id: str = '', start_date: str = '', end_date: str = '', user: dict = Depends(require_login)): conn = get_connection() cursor = conn.cursor(dictionary=True) filters = [] params = [] if job_title: filters.append("rq.job_titles LIKE %s") params.append(f"%{job_title}%") if location: filters.append("rq.locations LIKE %s") params.append(f"%{location}%") if resume_query_id: filters.append("gr.resume_query_id = %s") params.append(resume_query_id) if start_date: filters.append("rq.created_at >= %s") params.append(start_date) if end_date: filters.append("rq.created_at <= %s") params.append(end_date) where_clause = f"WHERE {' AND '.join(filters)}" if filters else "" query = f""" SELECT gr.id, gr.url, gr.email, gr.page_number, gr.status, gr.retry, gr.resume_query_id, rq.job_titles, rq.locations, rq.created_at FROM google_search_results gr JOIN resume_queries rq ON gr.resume_query_id = rq.id {where_clause} ORDER BY gr.resume_query_id DESC LIMIT 1000 """ cursor.execute(query, tuple(params)) results = cursor.fetchall() cursor.close() conn.close() return templates.TemplateResponse("results.html", { "request": request, "results": results, "filters": { "job_title": job_title, "location": location, "resume_query_id": resume_query_id, "start_date": start_date, "end_date": end_date } }) @app.get("/export/all") def export_all(job_title: str = '', location: str = '', start_date: str = '', end_date: str = '', user: dict = Depends(require_login)): conn = mysql.connector.connect( host=os.getenv("HOST"), user=os.getenv("USER"), password=os.getenv("PASSWORD"), database=os.getenv("DATABASE") ) cursor = conn.cursor() query = """ SELECT gr.url, gr.email, rq.job_titles, rq.locations, rq.created_at FROM google_search_results gr JOIN resume_queries rq ON gr.resume_query_id = rq.id """ filters = [] params = [] if job_title: filters.append("rq.job_titles LIKE %s") params.append(f"%{job_title}%") if location: filters.append("rq.locations LIKE %s") params.append(f"%{location}%") if start_date: filters.append("rq.created_at >= %s") params.append(start_date) if end_date: filters.append("rq.created_at <= %s") params.append(end_date) if filters: query += " WHERE " + " AND ".join(filters) cursor.execute(query, tuple(params)) rows = cursor.fetchall() headers = [i[0] for i in cursor.description] output = io.StringIO() writer = csv.writer(output) writer.writerow(headers) writer.writerows(rows) output.seek(0) cursor.close() conn.close() return StreamingResponse(output, media_type="text/csv", headers={"Content-Disposition": "attachment; filename=export_all.csv"}) @app.get("/export/emails") def export_emails(job_title: str = '', location: str = '', start_date: str = '', end_date: str = '', user: dict = Depends(require_login)): conn = mysql.connector.connect( host=os.getenv("HOST"), user=os.getenv("USER"), password=os.getenv("PASSWORD"), database=os.getenv("DATABASE") ) cursor = conn.cursor() # Base query for email export query = """ SELECT gr.email FROM google_search_results gr JOIN resume_queries rq ON gr.resume_query_id = rq.id WHERE gr.email IS NOT NULL AND gr.email != '' """ filters = [] params = [] if job_title: filters.append("rq.job_titles LIKE %s") params.append(f"%{job_title}%") if location: filters.append("rq.locations LIKE %s") params.append(f"%{location}%") if start_date: filters.append("rq.created_at >= %s") params.append(start_date) if end_date: end_datetime = datetime.strptime(end_date, "%Y-%m-%d") + timedelta(days=1) filters.append("rq.created_at < %s") params.append(end_datetime.strftime("%Y-%m-%d")) # Only add filters if they exist if filters: query += " AND " + " AND ".join(filters) print(query) cursor.execute(query, tuple(params)) emails = [row[0] for row in cursor.fetchall()] # Prepare CSV output output = io.StringIO("\n".join(emails)) output.seek(0) cursor.close() conn.close() return StreamingResponse(output, media_type="text/csv", headers={"Content-Disposition": "attachment; filename=emails.csv"}) @app.get("/login", response_class=HTMLResponse) def login_form(request: Request): return templates.TemplateResponse("login.html", {"request": request}) @app.post("/login") def login(request: Request, username: str = Form(...), password: str = Form(...)): conn = get_connection() cursor = conn.cursor(dictionary=True) cursor.execute("SELECT * FROM users WHERE username = %s", (username,)) user = cursor.fetchone() cursor.close() conn.close() if user and bcrypt.verify(password, user["password_hash"]): request.session["user"] = {"id": user["id"], "username": user["username"]} return RedirectResponse(url="/", status_code=302) return templates.TemplateResponse("login.html", {"request": request, "error": "Invalid credentials"}) @app.get("/logout") def logout(request: Request): request.session.clear() return RedirectResponse(url="/login", status_code=302)