""" Maps the repaired CSV records onto the relational model. Design notes ------------ * **Idempotent.** The natural key is the LinkedIn id (falling back to the profile URL), so re-running the import updates instead of duplicating. * **Bulk friendly.** Companies, schools and skills are resolved through in-memory caches, and the child rows of a profile are replaced wholesale with ``bulk_create`` - 284 profiles import in a couple of seconds. * **Self healing input.** The top level columns of this export are frequently missing, but the same information usually exists inside the nested ``experience`` blob. ``_derive_current_position`` falls back to the primary experience entry, which raises coverage of ``job_title_role`` from 108 to 183 profiles and of the current employer from 219 to 271. """ from __future__ import annotations import logging import re from dataclasses import dataclass, field from typing import Any from django.contrib.postgres.search import SearchVector from django.db import transaction from apps.profiles.dataset import DatasetReader, ImportReport, parse_literal from apps.profiles.models import ( Company, Education, Experience, Profile, ProfileSkill, School, Skill, ) logger = logging.getLogger(__name__) _YEAR_RE = re.compile(r"(\d{4})") _MAX_SKILLS = 60 @dataclass class ImportStats: created: int = 0 updated: int = 0 skills: int = 0 companies: int = 0 schools: int = 0 experiences: int = 0 educations: int = 0 report: ImportReport = field(default_factory=ImportReport) def as_dict(self) -> dict[str, Any]: return { "profiles_created": self.created, "profiles_updated": self.updated, "skills": self.skills, "companies": self.companies, "schools": self.schools, "experiences": self.experiences, "educations": self.educations, **self.report.as_dict(), } class ProfileImporter: """Loads the dataset into PostgreSQL.""" def __init__(self, path: str, limit: int | None = None) -> None: self.path = path self.limit = limit self._companies: dict[str, Company] = {} self._schools: dict[str, School] = {} self._skills: dict[str, Skill] = {} # -- Entry point -------------------------------------------------------- def run(self) -> ImportStats: records, report = DatasetReader(self.path).read() if self.limit: records = records[: self.limit] stats = ImportStats(report=report) self._warm_caches() for record in records: try: with transaction.atomic(): self._import_record(record, stats) except Exception: # noqa: BLE001 - one bad row must not abort the run logger.exception( "profile_import_failed", extra={"event": "profile_import_failed", "name": record.get("full_name")}, ) report.issues.append(record.get("full_name", "")) self._refresh_skill_counters() self._refresh_search_vectors() stats.skills = len(self._skills) stats.companies = len(self._companies) stats.schools = len(self._schools) return stats # -- Caches ------------------------------------------------------------- def _warm_caches(self) -> None: self._companies = {c.name: c for c in Company.objects.all()} self._schools = {s.name: s for s in School.objects.all()} self._skills = {s.slug: s for s in Skill.objects.all()} # -- Per record --------------------------------------------------------- def _import_record(self, record: dict[str, str], stats: ImportStats) -> None: experiences = parse_literal(record.get("experience"), []) educations = parse_literal(record.get("education"), []) primary = self._primary_experience(experiences) title, role, sub_role, levels = self._derive_current_position(record, primary) company = self._resolve_current_company(record, primary) skills = self._clean_str_list(parse_literal(record.get("skills"), []))[:_MAX_SKILLS] past_companies = [ str(item.get("company", {}).get("name") or "") for item in experiences if isinstance(item, dict) and isinstance(item.get("company"), dict) ] schools = [ str(item.get("school", {}).get("name") or "") for item in educations if isinstance(item, dict) and isinstance(item.get("school"), dict) ] majors = [ major for item in educations if isinstance(item, dict) for major in self._clean_str_list(item.get("majors") or []) ] defaults: dict[str, Any] = { "full_name": record.get("full_name", "")[:255], "first_name": record.get("first_name", "")[:128], "last_name": record.get("last_name", "")[:128], "gender": record.get("gender", "")[:16], "linkedin_url": record.get("linkedin_url", "")[:255], "linkedin_username": record.get("linkedin_username", "")[:128], "github_url": record.get("github_url", "")[:255], "twitter_url": record.get("twitter_url", "")[:255], "job_title": title[:255], "job_title_role": role[:64], "job_title_sub_role": sub_role[:64], "seniority_levels": levels, "current_company": company, "industry": (record.get("industry") or "")[:128], "job_summary": record.get("job_summary", ""), "job_start_date": record.get("job_start_date", "")[:10], "location_name": record.get("location_name", "")[:255], "location_locality": record.get("location_locality", "")[:128], "location_region": record.get("location_region", "")[:128], "location_country": record.get("location_country", "")[:128], "location_continent": record.get("location_continent", "")[:64], "summary": record.get("summary", ""), "years_experience": self._to_int(record.get("inferred_years_experience"), 0, 70), "inferred_salary": record.get("inferred_salary", "")[:32], "connections": self._to_int(record.get("linkedin_connections"), 0, 100_000), "interests": self._clean_str_list(parse_literal(record.get("interests"), []))[:40], "languages": self._language_names(parse_literal(record.get("languages"), [])), "emails": self._emails(parse_literal(record.get("emails"), [])), "phone_numbers": self._clean_str_list(parse_literal(record.get("phone_numbers"), []))[ :10 ], "skills_text": " ".join(skills), "companies_text": " ".join( filter(None, {*past_companies, company.name if company else ""}) ), "education_text": " ".join(filter(None, [*schools, *majors])), } external_id = (record.get("linkedin_id") or "")[:64] lookup = ( {"external_id": external_id} if external_id else {"linkedin_url": defaults["linkedin_url"]} ) profile, created = Profile.objects.update_or_create(**lookup, defaults=defaults) if created: stats.created += 1 else: stats.updated += 1 self._sync_skills(profile, skills) stats.experiences += self._sync_experiences(profile, experiences) stats.educations += self._sync_educations(profile, educations) # -- Current position --------------------------------------------------- @staticmethod def _primary_experience(experiences: list[Any]) -> dict[str, Any]: entries = [item for item in experiences if isinstance(item, dict)] for item in entries: if item.get("is_primary"): return item return entries[0] if entries else {} def _derive_current_position( self, record: dict[str, str], primary: dict[str, Any] ) -> tuple[str, str, str, list[str]]: title_blob = primary.get("title") if isinstance(primary.get("title"), dict) else {} title = record.get("job_title") or str(title_blob.get("name") or "") role = record.get("job_title_role") or str(title_blob.get("role") or "") sub_role = record.get("job_title_sub_role") or str(title_blob.get("sub_role") or "") levels = self._clean_str_list( parse_literal(record.get("job_title_levels"), None) or title_blob.get("levels") or [] ) return title, role, sub_role, levels[:5] def _resolve_current_company( self, record: dict[str, str], primary: dict[str, Any] ) -> Company | None: name = (record.get("job_company_name") or "").strip() payload: dict[str, Any] = { "external_id": record.get("job_company_id", ""), "industry": record.get("job_company_industry", ""), "size": record.get("job_company_size", ""), "founded": self._to_int(record.get("job_company_founded"), 1600, 2100), "website": record.get("job_company_website", ""), "linkedin_url": record.get("job_company_linkedin_url", ""), "location_name": record.get("job_company_location_name", ""), "country": record.get("job_company_location_country", ""), } if not name and isinstance(primary.get("company"), dict): blob = primary["company"] name = str(blob.get("name") or "").strip() location = blob.get("location") if isinstance(blob.get("location"), dict) else {} payload = { "external_id": str(blob.get("id") or ""), "industry": str(blob.get("industry") or ""), "size": str(blob.get("size") or ""), "founded": self._to_int(str(blob.get("founded") or ""), 1600, 2100), "website": str(blob.get("website") or ""), "linkedin_url": str(blob.get("linkedin_url") or ""), "location_name": str(location.get("name") or ""), "country": str(location.get("country") or ""), } return self._get_or_create_company(name, payload) def _get_or_create_company(self, name: str, payload: dict[str, Any]) -> Company | None: name = name.strip()[:255] if not name: return None cached = self._companies.get(name) if cached: return cached company = Company.objects.create( name=name, external_id=str(payload.get("external_id") or "")[:128], industry=str(payload.get("industry") or "")[:128], size=str(payload.get("size") or "")[:32], founded=payload.get("founded"), website=str(payload.get("website") or "")[:255], linkedin_url=str(payload.get("linkedin_url") or "")[:255], location_name=str(payload.get("location_name") or "")[:255], country=str(payload.get("country") or "")[:128], ) self._companies[name] = company return company def _get_or_create_school(self, blob: dict[str, Any]) -> School | None: name = str(blob.get("name") or "").strip()[:255] if not name: return None cached = self._schools.get(name) if cached: return cached location = blob.get("location") if isinstance(blob.get("location"), dict) else {} school = School.objects.create( name=name, type=str(blob.get("type") or "")[:64], website=str(blob.get("website") or "")[:255], linkedin_url=str(blob.get("linkedin_url") or "")[:255], location_name=str(location.get("name") or "")[:255], country=str(location.get("country") or "")[:128], ) self._schools[name] = school return school # -- Children ----------------------------------------------------------- def _sync_skills(self, profile: Profile, skills: list[str]) -> None: ProfileSkill.objects.filter(profile=profile).delete() links: list[ProfileSkill] = [] used: set[int] = set() for position, name in enumerate(skills): slug = Skill.make_slug(name) if not slug: continue skill = self._skills.get(slug) if skill is None: skill, _ = Skill.objects.get_or_create(slug=slug, defaults={"name": name[:128]}) self._skills[slug] = skill if skill.pk in used: continue used.add(skill.pk) links.append(ProfileSkill(profile=profile, skill=skill, position=position)) ProfileSkill.objects.bulk_create(links, ignore_conflicts=True) def _sync_experiences(self, profile: Profile, experiences: list[Any]) -> int: Experience.objects.filter(profile=profile).delete() rows: list[Experience] = [] for item in experiences: if not isinstance(item, dict): continue blob = item.get("company") if isinstance(item.get("company"), dict) else {} title = item.get("title") if isinstance(item.get("title"), dict) else {} company = self._get_or_create_company( str(blob.get("name") or ""), { "external_id": blob.get("id"), "industry": blob.get("industry"), "size": blob.get("size"), "founded": self._to_int(str(blob.get("founded") or ""), 1600, 2100), "website": blob.get("website"), "linkedin_url": blob.get("linkedin_url"), "location_name": ( (blob.get("location") or {}).get("name") if isinstance(blob.get("location"), dict) else "" ), "country": ( (blob.get("location") or {}).get("country") if isinstance(blob.get("location"), dict) else "" ), }, ) rows.append( Experience( profile=profile, company=company, company_name=str(blob.get("name") or "")[:255], title=str(title.get("name") or "")[:255], title_role=str(title.get("role") or "")[:64], title_sub_role=str(title.get("sub_role") or "")[:64], levels=self._clean_str_list(title.get("levels") or [])[:5], start_date=str(item.get("start_date") or "")[:10], end_date=str(item.get("end_date") or "")[:10], start_year=self._year(item.get("start_date")), end_year=self._year(item.get("end_date")), is_primary=bool(item.get("is_primary")), location_name=self._first_location(item.get("location_names")), summary=str(item.get("summary") or ""), ) ) Experience.objects.bulk_create(rows) return len(rows) def _sync_educations(self, profile: Profile, educations: list[Any]) -> int: Education.objects.filter(profile=profile).delete() rows: list[Education] = [] for item in educations: if not isinstance(item, dict): continue blob = item.get("school") if isinstance(item.get("school"), dict) else {} rows.append( Education( profile=profile, school=self._get_or_create_school(blob), school_name=str(blob.get("name") or "")[:255], degrees=self._clean_str_list(item.get("degrees") or [])[:5], majors=self._clean_str_list(item.get("majors") or [])[:5], minors=self._clean_str_list(item.get("minors") or [])[:5], gpa=self._to_decimal(item.get("gpa")), start_date=str(item.get("start_date") or "")[:10], end_date=str(item.get("end_date") or "")[:10], end_year=self._year(item.get("end_date")), ) ) Education.objects.bulk_create(rows) return len(rows) # -- Post processing ---------------------------------------------------- @staticmethod def _refresh_skill_counters() -> None: """Single UPDATE ... FROM keeps the facet ordering cheap at query time.""" from django.db import connection with connection.cursor() as cursor: cursor.execute( """ UPDATE profiles_skill AS s SET profile_count = COALESCE(counts.total, 0) FROM ( SELECT skill_id, COUNT(*) AS total FROM profiles_profileskill GROUP BY skill_id ) AS counts WHERE counts.skill_id = s.id """ ) cursor.execute( """ UPDATE profiles_skill SET profile_count = 0 WHERE id NOT IN (SELECT skill_id FROM profiles_profileskill) """ ) @staticmethod def _refresh_search_vectors() -> None: """ Rebuild the tsvector for every profile in one statement. Weights encode the ranking intent: a hit on the person's name beats a hit on their title or skills, which in turn beats a hit in the free-text summary. """ Profile.objects.update( search_vector=( SearchVector("full_name", weight="A", config="english") + SearchVector("job_title", weight="B", config="english") + SearchVector("skills_text", weight="B", config="english") + SearchVector("companies_text", weight="B", config="english") + SearchVector("industry", weight="C", config="english") + SearchVector("education_text", weight="C", config="english") + SearchVector("location_name", weight="C", config="english") + SearchVector("summary", weight="D", config="english") ) ) # -- Small helpers ------------------------------------------------------ @staticmethod def _clean_str_list(values: Any) -> list[str]: if not isinstance(values, list): return [] seen: set[str] = set() out: list[str] = [] for value in values: if not isinstance(value, str): continue cleaned = value.strip().lower()[:128] if cleaned and cleaned not in seen: seen.add(cleaned) out.append(cleaned) return out @staticmethod def _language_names(values: Any) -> list[str]: if not isinstance(values, list): return [] return [ str(item.get("name")).strip().lower()[:64] for item in values if isinstance(item, dict) and item.get("name") ][:10] @staticmethod def _emails(values: Any) -> list[dict[str, str]]: if not isinstance(values, list): return [] return [ { "address": str(item.get("address"))[:254], "type": str(item.get("type") or "other")[:32], } for item in values if isinstance(item, dict) and item.get("address") ][:10] @staticmethod def _first_location(values: Any) -> str: if isinstance(values, list) and values and isinstance(values[0], str): return values[0][:255] return "" @staticmethod def _to_int(value: str | None, minimum: int, maximum: int) -> int | None: if not value: return None try: number = int(float(str(value).replace(",", ""))) except (TypeError, ValueError): return None return number if minimum <= number <= maximum else None @staticmethod def _to_decimal(value: Any) -> Any: try: number = float(value) except (TypeError, ValueError): return None return round(number, 2) if 0 <= number <= 10 else None @staticmethod def _year(value: Any) -> int | None: if not value: return None match = _YEAR_RE.search(str(value)) if not match: return None year = int(match.group(1)) return year if 1900 <= year <= 2100 else None