Spaces:
Sleeping
Sleeping
| import os | |
| import re | |
| import json | |
| import requests | |
| import gradio as gr | |
| from pypdf import PdfReader | |
| from docx import Document | |
| OPENWEBNINJA_API_KEY = os.getenv("OPENWEBNINJA_API_KEY") | |
| GEMINI_CHAT_URL = "https://api.openwebninja.com/gemini/chat" | |
| JSEARCH_URL = "https://api.openwebninja.com/jsearch/search-v2" | |
| def extract_text(file): | |
| if file is None: | |
| return "" | |
| path = file if isinstance(file, str) else file.name | |
| print("FILE PATH:", path) | |
| try: | |
| # Try PDF first | |
| try: | |
| reader = PdfReader(path) | |
| text = "" | |
| for page in reader.pages: | |
| page_text = page.extract_text() | |
| if page_text: | |
| text += page_text + "\n" | |
| if text.strip(): | |
| print("Detected PDF") | |
| return text | |
| except: | |
| pass | |
| # Try DOCX | |
| try: | |
| doc = Document(path) | |
| text = "\n".join( | |
| p.text for p in doc.paragraphs | |
| ) | |
| if text.strip(): | |
| print("Detected DOCX") | |
| return text | |
| except: | |
| pass | |
| # Try TXT | |
| try: | |
| with open(path, "r", encoding="utf-8", errors="ignore") as f: | |
| text = f.read() | |
| if text.strip(): | |
| print("Detected TXT") | |
| return text | |
| except: | |
| pass | |
| except Exception as e: | |
| print("TEXT EXTRACTION ERROR:", e) | |
| return "" | |
| def clean_text(text): | |
| text = text.replace("\n", " ") | |
| text = re.sub(r"([a-z0-9])([A-Z])", r"\1 \2", text) | |
| text = re.sub(r"\s+", " ", text) | |
| return text.strip() | |
| def call_openwebninja_gemini(prompt): | |
| headers = { | |
| "X-API-Key": OPENWEBNINJA_API_KEY, | |
| "Content-Type": "application/json" | |
| } | |
| payloads = [ | |
| {"message": prompt}, | |
| {"prompt": prompt}, | |
| {"messages": [{"role": "user", "content": prompt}]} | |
| ] | |
| for payload in payloads: | |
| try: | |
| response = requests.post( | |
| GEMINI_CHAT_URL, | |
| headers=headers, | |
| json=payload, | |
| timeout=60 | |
| ) | |
| print("GEMINI STATUS:", response.status_code) | |
| print("GEMINI RESPONSE:", response.text[:1000]) | |
| if response.status_code == 200: | |
| return response.text | |
| except Exception as e: | |
| print("Gemini API error:", e) | |
| return "" | |
| def extract_json(text): | |
| if not text: | |
| return None | |
| try: | |
| data = json.loads(text) | |
| # OpenWebNinja Gemini format | |
| if isinstance(data, dict): | |
| if "data" in data and isinstance(data["data"], dict): | |
| reply_text = data["data"].get("reply_text") | |
| if reply_text: | |
| return extract_json(reply_text) | |
| for key in ["reply_text", "response", "text", "message", "content", "answer"]: | |
| if key in data and isinstance(data[key], str): | |
| return extract_json(data[key]) | |
| return data | |
| except: | |
| pass | |
| match = re.search(r"\{[\s\S]*\}", text) | |
| if match: | |
| try: | |
| return json.loads(match.group(0)) | |
| except: | |
| return None | |
| return None | |
| def analyze_resume_with_ai(resume_text): | |
| prompt = f""" | |
| You are an AI resume analysis assistant. | |
| Analyze this resume/CV from any career field. | |
| Return ONLY valid JSON in this exact structure: | |
| {{ | |
| "candidate_field": "", | |
| "seniority_level": "", | |
| "education": [], | |
| "work_experience": [], | |
| "technical_skills": [], | |
| "soft_skills": [], | |
| "certifications": [], | |
| "suitable_job_titles": [], | |
| "job_search_queries": [] | |
| }} | |
| Rules: | |
| Rules: | |
| - The resume can be from any field. | |
| - Generate 15 to 20 suitable_job_titles. | |
| - Include junior, graduate, entry-level and related roles. | |
| - Include alternative titles and synonyms. | |
| - Generate 10 to 15 diverse job_search_queries. | |
| - job_search_queries must not be empty. | |
| - Return at least 15 suitable_job_titles. | |
| - Do not include explanations outside JSON. | |
| Resume: | |
| {resume_text[:7000]} | |
| """ | |
| raw = call_openwebninja_gemini(prompt) | |
| profile = extract_json(raw) | |
| if not profile: | |
| return None, raw | |
| # Make sure all expected keys exist | |
| profile.setdefault("candidate_field", "") | |
| profile.setdefault("seniority_level", "") | |
| profile.setdefault("education", []) | |
| profile.setdefault("work_experience", []) | |
| profile.setdefault("technical_skills", []) | |
| profile.setdefault("soft_skills", []) | |
| profile.setdefault("certifications", []) | |
| profile.setdefault("suitable_job_titles", []) | |
| expanded = [] | |
| for role in profile["suitable_job_titles"]: | |
| expanded.extend([ | |
| role, | |
| f"Junior {role}", | |
| f"Graduate {role}", | |
| f"Entry Level {role}" | |
| ]) | |
| profile["suitable_job_titles"] = list( | |
| dict.fromkeys(expanded) | |
| ) | |
| profile.setdefault("job_search_queries", []) | |
| # Fallback query generation | |
| if not profile["job_search_queries"]: | |
| fallback_queries = [] | |
| for title in profile["suitable_job_titles"]: | |
| fallback_queries.append(title) | |
| if profile["candidate_field"]: | |
| fallback_queries.append(profile["candidate_field"]) | |
| for skill in profile["technical_skills"][:4]: | |
| fallback_queries.append(skill) | |
| # Remove duplicates | |
| seen = set() | |
| clean_queries = [] | |
| for q in fallback_queries: | |
| q = str(q).strip() | |
| if q and q.lower() not in seen: | |
| seen.add(q.lower()) | |
| clean_queries.append(q) | |
| profile["job_search_queries"] = clean_queries[:8] | |
| return profile, raw | |
| def clean_query(query, location): | |
| query = query.replace("jobs in", "") | |
| query = query.replace("Jobs in", "") | |
| query = query.replace(location, "") | |
| return " ".join(query.split()).strip() | |
| def search_jobs(query, location): | |
| headers = { | |
| "X-API-Key": OPENWEBNINJA_API_KEY | |
| } | |
| cleaned = clean_query(query, location) | |
| full_query = f"{cleaned} jobs in {location}" | |
| response = requests.get( | |
| JSEARCH_URL, | |
| params={"query": full_query}, | |
| headers=headers, | |
| timeout=30 | |
| ) | |
| print("JSEARCH QUERY:", full_query) | |
| print("JSEARCH STATUS:", response.status_code) | |
| print("JSEARCH RESPONSE:", response.text[:1000]) | |
| if response.status_code != 200: | |
| return [] | |
| data = response.json() | |
| jobs = ( | |
| data.get("jobs") | |
| or data.get("data") | |
| or data.get("results") | |
| or data.get("items") | |
| or [] | |
| ) | |
| if isinstance(jobs, dict): | |
| jobs = ( | |
| jobs.get("jobs") | |
| or jobs.get("results") | |
| or jobs.get("items") | |
| or [] | |
| ) | |
| return jobs if isinstance(jobs, list) else [] | |
| def get_job_field(job, *keys): | |
| for key in keys: | |
| if job.get(key): | |
| return job.get(key) | |
| for nested_key in ["job", "company", "employer", "details"]: | |
| nested = job.get(nested_key) | |
| if isinstance(nested, dict): | |
| for key in keys: | |
| if nested.get(key): | |
| return nested.get(key) | |
| return "" | |
| def score_job(profile, job): | |
| skills = profile.get("technical_skills", []) + profile.get("soft_skills", []) | |
| titles = profile.get("suitable_job_titles", []) | |
| job_text = json.dumps(job).lower() | |
| matched = [] | |
| score = 0 | |
| for skill in skills: | |
| if skill.lower() in job_text: | |
| matched.append(skill) | |
| score += 8 | |
| for title in titles: | |
| if title.lower() in job_text: | |
| score += 15 | |
| score = min(score, 100) | |
| return score, matched | |
| def analyze_resume(file, location): | |
| try: | |
| print("FILE OBJECT:", file) | |
| print("FILE TYPE:", type(file)) | |
| if not OPENWEBNINJA_API_KEY: | |
| return "❌ OPENWEBNINJA_API_KEY is missing in Hugging Face Secrets." | |
| resume_text = clean_text(extract_text(file)) | |
| print("EXTRACTED LENGTH:", len(resume_text)) | |
| print("EXTRACTED TEXT:", resume_text[:500]) | |
| if not resume_text: | |
| return f""" | |
| ❌ Could not extract text from the resume. | |
| File: | |
| {file} | |
| Type: | |
| {type(file)} | |
| """ | |
| profile, raw_ai = analyze_resume_with_ai(resume_text) | |
| if not profile: | |
| return f"❌ AI could not analyze the resume.\n\nRaw response:\n\n```text\n{raw_ai}\n```" | |
| queries = profile.get("job_search_queries", []) | |
| if not queries: | |
| return f"""❌ AI did not generate job search queries. | |
| Raw AI response: | |
| ```text | |
| {raw_ai} | |
| {json.dumps(profile, indent=2)} | |
| ```""" | |
| output = "# CareerMatch AI Results\n\n" | |
| output += f"**Field:** {profile.get('candidate_field', 'Not detected')}\n\n" | |
| output += f"**Seniority:** {profile.get('seniority_level', 'Not detected')}\n\n" | |
| output += f"**Skills:** {', '.join(profile.get('technical_skills', [])) or 'Not detected'}\n\n" | |
| output += f"**Suitable Jobs:** {', '.join(profile.get('suitable_job_titles', [])) or 'Not detected'}\n\n" | |
| output += f"**Search Queries:** {', '.join(queries)}\n\n---\n\n" | |
| all_jobs = [] | |
| for query in queries[:15]: | |
| jobs = search_jobs(query, location) | |
| output += f"Search: `{clean_query(query, location)} jobs in {location}` → {len(jobs)} jobs found\n\n" | |
| for job in jobs[:10]: | |
| if isinstance(job, dict): | |
| all_jobs.append(job) | |
| if not all_jobs: | |
| return output + "❌ No jobs found from JSearch." | |
| unique_jobs = [] | |
| seen = set() | |
| for job in all_jobs: | |
| title = get_job_field(job, "title", "job_title", "name") | |
| company = get_job_field(job, "company_name", "company", "employer_name") | |
| key = f"{title}-{company}" | |
| if key not in seen: | |
| seen.add(key) | |
| unique_jobs.append(job) | |
| ranked = [] | |
| for job in unique_jobs[:10]: | |
| score, matched = score_job(profile, job) | |
| ranked.append((score, matched, job)) | |
| ranked.sort(key=lambda x: x[0], reverse=True) | |
| output += "\n---\n\n## Real Job Matches\n\n" | |
| for score, matched, job in ranked: | |
| title = get_job_field(job, "title", "job_title", "name") or "Unknown Job" | |
| company = get_job_field(job, "company_name", "company", "employer_name") or "Unknown Company" | |
| job_location = get_job_field(job, "location", "job_location", "formatted_location") or "Not specified" | |
| url = get_job_field(job, "url", "apply_link", "job_apply_link", "link") or "#" | |
| output += f"### {title} — {score}% Match\n\n" | |
| output += f"**Company:** {company}\n\n" | |
| output += f"**Location:** {job_location}\n\n" | |
| output += f"**Matched Skills:** {', '.join(matched) if matched else 'Not detected'}\n\n" | |
| output += f"[Apply Here]({url})\n\n---\n\n" | |
| return output | |
| except Exception as e: | |
| return f"❌ Error occurred:\n\n```text\n{type(e).__name__}: {str(e)}\n```" | |
| demo = gr.Interface( | |
| fn=analyze_resume, | |
| inputs=[ | |
| gr.File( | |
| label="Upload Resume/CV", | |
| type="filepath" | |
| ), | |
| gr.Textbox(label="Job Location", value="Mauritius") | |
| ], | |
| outputs=gr.Markdown(label="Results"), | |
| title="CareerMatch AI", | |
| description="Upload any resume and get real job matches using OpenWebNinja Gemini + JSearch." | |
| ) | |
| if __name__ == "__main__": | |
| demo.launch() |