""" resume_parser.py ---------------- Parses a resume (text or PDF) into a structured JSON profile used by the matching engine. Run once, outputs profile.json. """ import json import re import sys import os import logging logger = logging.getLogger(__name__) # ---- Known skill taxonomy ---- SKILL_TAXONOMY = { "bgp": ["bgp", "border gateway protocol", "bgp peering", "route maps"], "ospf": ["ospf", "open shortest path first"], "eigrp": ["eigrp"], "mpls": ["mpls"], "vxlan": ["vxlan"], "vrrp": ["vrrp"], "vlans": ["vlan", "vlans"], "stp": ["stp", "spanning tree"], "tcp/ip": ["tcp/ip", "tcp", "ip", "tcpip"], "dns": ["dns", "domain name"], "dhcp": ["dhcp"], "qos": ["qos", "quality of service"], "ipv6": ["ipv6"], "multicast": ["multicast", "pim", "igmp"], "nat": ["nat", "network address translation"], "acls": ["acl", "acls", "access control list"], "load balancing": ["load balanc"], "cisco": ["cisco", "catalyst", "nexus"], "arista": ["arista"], "juniper": ["juniper", "junos", "srx"], "mikrotik": ["mikrotik"], "brocade": ["brocade"], "fortinet": ["fortinet", "fortigate"], "pfsense": ["pfsense"], "barracuda": ["barracuda"], "vyos": ["vyos"], "firewall": ["firewall", "ngfw"], "vpn": ["vpn", "ipsec", "site-to-site"], "802.1x": ["802.1x", "nac", "network access control"], "2fa": ["2fa", "mfa", "two-factor", "multi-factor"], "network segmentation": ["network segmentation", "microsegmentation"], "aws": ["aws", "amazon web services", "vpc"], "azure": ["azure"], "gcp": ["gcp", "google cloud"], "ibm cloud": ["ibm cloud"], "vmware": ["vmware", "esxi", "vsphere"], "proxmox": ["proxmox"], "lxc": ["lxc", "linux container"], "docker": ["docker", "container"], "kubernetes": ["kubernetes", "k8s"], "terraform": ["terraform"], "hybrid cloud": ["hybrid cloud"], "python": ["python"], "bash": ["bash", "shell script"], "ansible": ["ansible"], "git": ["git", "github", "gitlab"], "iac": ["infrastructure as code", "infrastructure-as-code", "iac"], "ci/cd": ["ci/cd", "cicd", "jenkins", "pipeline"], "zabbix": ["zabbix"], "librenms": ["librenms"], "nagios": ["nagios"], "snmp": ["snmp"], "syslog": ["syslog"], "datadog": ["datadog"], "grafana": ["grafana"], "prometheus": ["prometheus"], "splunk": ["splunk"], "solarwinds": ["solarwinds"], "active directory": ["active directory", "ad"], "linux": ["linux", "unix", "rhel", "ubuntu", "centos", "debian"], "windows server": ["windows server", "windows ad"], "voip": ["voip", "sip"], "sql": ["sql", "mysql", "postgres", "mssql"], "certificate management": ["certificate", "ssl", "tls", "pki"], } CERT_PATTERNS = [ (r"ccnp", "CCNP"), (r"ccna", "CCNA"), (r"ccie", "CCIE"), (r"aws\s+certified\s+cloud\s+practitioner", "AWS Cloud Practitioner"), (r"aws\s+certified\s+solutions?\s+architect", "AWS Solutions Architect"), (r"aws\s+certified", "AWS Certified"), (r"comptia\s+security\+", "CompTIA Security+"), (r"comptia\s+network\+", "CompTIA Network+"), (r"jncia", "JNCIA"), (r"jncis", "JNCIS"), (r"fortinet\s+nse", "Fortinet NSE"), (r"palo\s+alto\s+pcnse", "PCNSE"), ] TITLE_VARIANTS = { "network engineer": ["network engineer", "network eng", "noc engineer"], "infrastructure engineer": ["infrastructure engineer", "infra engineer"], "network administrator": ["network administrator", "network admin"], "system administrator": ["system administrator", "sysadmin", "systems administrator"], "cloud engineer": ["cloud engineer", "cloud network engineer"], "devops engineer": ["devops", "site reliability", "sre"], "security engineer": ["security engineer", "network security"], } def extract_skills(text): text_lower = text.lower() found = {} for skill_name, patterns in SKILL_TAXONOMY.items(): for pattern in patterns: if pattern.lower() in text_lower: count = text_lower.count(pattern.lower()) found[skill_name] = min(count, 5) break return found def extract_titles(text): titles = [] title_patterns = [ r"(?:^|\n)\s*([\w\s&/]+(?:Engineer|Administrator|Analyst|Manager|Architect|Specialist|Technician|Director))\s*(?:\(|–|-|\d|$)", ] for pattern in title_patterns: matches = re.findall(pattern, text, re.IGNORECASE | re.MULTILINE) for m in matches: clean = m.strip() if 5 < len(clean) < 60: titles.append(clean) return list(set(titles)) def extract_experience_years(text): explicit = re.search(r"(\d+)\+?\s*years?\s*(?:of\s+)?experience", text, re.IGNORECASE) if explicit: return int(explicit.group(1)) date_ranges = re.findall(r"(\d{2}/\d{4})\s*[–-]\s*(\d{2}/\d{4}|present|current)", text, re.IGNORECASE) if date_ranges: from datetime import datetime total_months = 0 for start_str, end_str in date_ranges: try: start = datetime.strptime(start_str, "%m/%Y") if end_str.lower() in ("present", "current"): end = datetime.now() else: end = datetime.strptime(end_str, "%m/%Y") months = (end.year - start.year) * 12 + (end.month - start.month) total_months += max(months, 0) except ValueError: continue if total_months > 0: return total_months // 12 return 0 def extract_certifications(text): text_lower = text.lower() found = [] for pattern, cert_name in CERT_PATTERNS: if re.search(pattern, text_lower): found.append(cert_name) return list(set(found)) def extract_education(text): education = [] edu_patterns = [ r"(bachelor|master|diploma|degree|b\.?s\.?|m\.?s\.?|ph\.?d|associate)[\w\s,]*(?:in\s+)?([\w\s&]+?)(?:\d{4}|\n|$)", ] for pattern in edu_patterns: matches = re.findall(pattern, text, re.IGNORECASE) for level, field in matches: education.append({"level": level.strip(), "field": field.strip()[:50]}) return education def parse_resume_text(text): skills = extract_skills(text) titles = extract_titles(text) years = extract_experience_years(text) certs = extract_certifications(text) education = extract_education(text) normalized_titles = set() for title in titles: title_lower = title.lower() for norm, variants in TITLE_VARIANTS.items(): if any(v in title_lower for v in variants): normalized_titles.add(norm) break else: normalized_titles.add(title_lower) return { "skills": skills, "titles": list(normalized_titles), "raw_titles": titles, "years_experience": years, "certifications": certs, "education": education, "full_text": text, } def parse_resume_file(filepath): ext = os.path.splitext(filepath)[1].lower() if ext == ".txt": with open(filepath, "r", encoding="utf-8") as f: text = f.read() elif ext == ".pdf": try: import pdfplumber with pdfplumber.open(filepath) as pdf: text = "\n".join(page.extract_text() or "" for page in pdf.pages) except ImportError: logger.error("Install pdfplumber: pip install pdfplumber") sys.exit(1) else: raise ValueError(f"Unsupported file type: {ext}") return parse_resume_text(text) def save_profile(profile, output_path="profile.json"): save_data = {k: v for k, v in profile.items() if k != "full_text"} with open(output_path, "w") as f: json.dump(save_data, f, indent=2) logger.info(f"Profile saved to {output_path}") return save_data if __name__ == "__main__": logging.basicConfig(level=logging.INFO) if len(sys.argv) < 2: print("Usage: python resume_parser.py ") sys.exit(1) filepath = sys.argv[1] profile = parse_resume_file(filepath) saved = save_profile(profile) print(json.dumps(saved, indent=2))