cyberyan_linkedin_search/backend/apps/profiles/dataset.py

"""
Resilient reader for the supplied LinkedIn CSV export.

Why this module exists
----------------------
The provided file is a real-world dirty export, not a clean CSV.  Feeding it to
``csv.DictReader`` silently produces garbage.  Four separate defects are present:

1. **Injected watermark** - many rows start with an extra, unheadered column
   containing ``H:\\New folder\\leak\\...@BreachedData1 ...``.  Everything after
   it is shifted one position to the right.
2. **Missing columns** - most records omit one or more optional fields
   (``job_title_role``, ``job_title_sub_role``, ...) instead of writing an empty
   value, so the remaining values slide left relative to the header.
3. **Broken quoting** - a handful of ``summary`` values contain an unescaped
   double quote, which splits a single logical record across several physical
   rows.
4. **Concatenated exports** - the header line appears a second time in the
   middle of the file and ~40 people are duplicated.

Repair strategy
---------------
Every column has a *signature*: ``linkedin_url`` contains ``linkedin.com/in/``,
``skills`` is a Python list literal, ``location_geo`` looks like ``lat,lon`` and
so on.  We score each (column, value) pair against those signatures and then run
a Needleman-Wunsch style alignment between the row's values and the header
columns, allowing two edit operations:

* *skip column* - the record simply does not carry that field;
* *drop value*  - a stray fragment that belongs to no column.

The alignment that maximises the total score is the reconstructed record.  The
result is validated a second time and any value that still fails its signature
is discarded rather than persisted as junk - it is better to store nothing than
to store a phone number in the e-mail column.

``ast.literal_eval`` is used (never ``eval``) so the nested Python-repr columns
can be parsed without executing anything from the file.
"""

from __future__ import annotations

import ast
import csv
import re
import sys
from collections.abc import Callable, Iterator
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any

# The nested columns are large; the stdlib default (128 KiB) is not enough.
csv.field_size_limit(min(sys.maxsize, 2**31 - 1))

WATERMARK_MARKERS = ("@BreachedData", "H:\\New folder")
CONTINENTS = frozenset(
    {"north america", "south america", "europe", "asia", "africa", "oceania", "antarctica"}
)
GENDERS = frozenset({"male", "female"})

Validator = Callable[[str], bool | None]


# ---------------------------------------------------------------------------
# Column signatures
# ---------------------------------------------------------------------------
def _list_of_str(value: str) -> bool | None:
    parsed = _safe_literal(value)
    if parsed is None:
        return None
    return isinstance(parsed, list) and all(isinstance(item, str) for item in parsed)


def _list_of_dicts(required_key: str) -> Validator:
    def check(value: str) -> bool | None:
        parsed = _safe_literal(value)
        if parsed is None:
            return None
        if not isinstance(parsed, list):
            return None
        if not parsed:
            return True
        return all(isinstance(item, dict) for item in parsed) and required_key in parsed[0]

    return check


def _dict_with(required_key: str) -> Validator:
    def check(value: str) -> bool | None:
        parsed = _safe_literal(value)
        if parsed is None:
            return None
        return isinstance(parsed, dict) and required_key in parsed

    return check


def _matches(pattern: str) -> Validator:
    compiled = re.compile(pattern)
    return lambda value: bool(compiled.fullmatch(value))


def _contains(needle: str) -> Validator:
    return lambda value: needle in value


def _one_of(options: frozenset[str]) -> Validator:
    return lambda value: value.lower() in options


def _safe_literal(value: str) -> Any | None:
    """``ast.literal_eval`` that returns ``None`` instead of raising."""
    try:
        return ast.literal_eval(value)
    except (ValueError, SyntaxError, MemoryError, RecursionError):
        return None


#: Columns without an entry here accept any text (name, title, industry, ...).
FIELD_VALIDATORS: dict[str, Validator] = {
    "gender": _one_of(GENDERS),
    "linkedin_url": _contains("linkedin.com/in/"),
    "linkedin_id": _matches(r"\d{3,20}"),
    "facebook_url": _contains("facebook.com/"),
    "facebook_id": _matches(r"\d{5,25}"),
    "job_title_role": _matches(r"[a-z_]{3,40}"),
    "job_title_sub_role": _matches(r"[a-z_]{3,40}"),
    "job_title_levels": _list_of_str,
    "job_company_website": _matches(r"[\w.-]+\.[a-z]{2,10}"),
    "job_company_size": _matches(
        r"(1-10|11-50|51-200|201-500|501-1000|1001-5000|5001-10000|10001\+)"
    ),
    "job_company_founded": _matches(r"(1[6-9]|20)\d\d(\.0)?"),
    "job_company_linkedin_url": _contains("linkedin.com/company/"),
    "job_company_linkedin_id": _matches(r"\d{3,20}"),
    "job_company_facebook_url": _contains("facebook.com/"),
    "job_company_twitter_url": _contains("twitter.com/"),
    "job_company_location_geo": _matches(r"-?\d+\.\d+,-?\d+\.\d+"),
    "job_company_location_continent": _one_of(CONTINENTS),
    "job_last_updated": _matches(r"\d{4}-\d{2}-\d{2}"),
    "job_start_date": _matches(r"\d{4}(-\d{2})?"),
    "location_geo": _matches(r"-?\d+\.\d+,-?\d+\.\d+"),
    "location_continent": _one_of(CONTINENTS),
    "location_last_updated": _matches(r"\d{4}-\d{2}-\d{2}"),
    "linkedin_connections": _matches(r"\d+(\.\d+)?"),
    "inferred_salary": _matches(r"[\d,]+-[\d,]+|>?\d[\d,]*\+?"),
    "inferred_years_experience": _matches(r"\d{1,2}(\.\d+)?"),
    "phone_numbers": _list_of_str,
    "emails": _list_of_dicts("address"),
    "interests": _list_of_str,
    "skills": _list_of_str,
    "location_names": _list_of_str,
    "regions": _list_of_str,
    "countries": _list_of_str,
    "street_addresses": _list_of_dicts("street_address"),
    "experience": _list_of_dicts("company"),
    "education": _list_of_dicts("school"),
    "profiles": _list_of_dicts("network"),
    "certifications": _list_of_dicts("name"),
    "languages": _list_of_dicts("name"),
    "version_status": _dict_with("status"),
    "work_email": _contains("@"),
    "middle_initial": _matches(r"[a-z]"),
    "birth_year": _matches(r"(19|20)\d\d"),
    "birth_date": _matches(r"\d{4}-\d{2}(-\d{2})?"),
    "twitter_url": _contains("twitter.com/"),
    "github_url": _contains("github.com/"),
    "mobile_phone": _matches(r"\+?\d[\d -]{6,20}"),
}

# Alignment weights.  Tuned so that a confirmed signature match outweighs any
# combination of edits, while losing a non-empty value is expensive.
SCORE_MATCH = 3.0
SCORE_MISMATCH = -4.0
SCORE_FREE_TEXT = 0.3
SCORE_EMPTY = 0.15
PENALTY_MISSING_COLUMN = -0.5
PENALTY_DROP_EMPTY = -0.3
PENALTY_DROP_VALUE = -3.5


@dataclass
class ImportReport:
    """Counters surfaced by the management command so imports are auditable."""

    physical_rows: int = 0
    duplicate_headers: int = 0
    repaired_rows: int = 0
    orphan_fragments: int = 0
    duplicate_records: int = 0
    discarded_values: int = 0
    records: int = 0
    issues: list[str] = field(default_factory=list)

    def as_dict(self) -> dict[str, Any]:
        return {
            "physical_rows": self.physical_rows,
            "duplicate_headers": self.duplicate_headers,
            "repaired_rows": self.repaired_rows,
            "orphan_fragments": self.orphan_fragments,
            "duplicate_records": self.duplicate_records,
            "discarded_values": self.discarded_values,
            "records": self.records,
        }


class DatasetReader:
    """Yields clean, de-duplicated records from the raw CSV export."""

    def __init__(self, path: str | Path) -> None:
        self.path = Path(path)

    def read(self) -> tuple[list[dict[str, str]], ImportReport]:
        report = ImportReport()
        with self.path.open(encoding="utf-8", newline="") as handle:
            rows = list(csv.reader(handle))

        if not rows:
            return [], report

        header = rows[0]
        report.physical_rows = len(rows) - 1

        logical_rows = self._reassemble(rows[1:], header, report)

        seen: set[str] = set()
        records: list[dict[str, str]] = []
        for values in logical_rows:
            record = self._align(values, header)
            self._sanitize(record, report)
            key = self._identity(record)
            if key in seen:
                report.duplicate_records += 1
                continue
            seen.add(key)
            records.append(record)

        report.records = len(records)
        return records, report

    # -- Step 1: turn physical rows into logical records --------------------
    def _reassemble(
        self, rows: list[list[str]], header: list[str], report: ImportReport
    ) -> list[list[str]]:
        expected = len(header)
        logical: list[list[str]] = []
        for raw_row in rows:
            row = self._strip_watermark(list(raw_row))
            if not row:
                continue
            if self._is_repeated_header(row, header):
                report.duplicate_headers += 1
                continue
            if self._starts_record(row) or not logical:
                if len(row) != expected:
                    report.repaired_rows += 1
                logical.append(row)
            elif len(logical[-1]) < expected - 7:
                # The previous record is short: this fragment is its tail,
                # split by a stray quote inside a free-text column.
                logical[-1].extend(row)
                report.repaired_rows += 1
            else:
                # Cannot be attributed to any record without guessing.
                report.orphan_fragments += 1
        return logical

    @staticmethod
    def _strip_watermark(row: list[str]) -> list[str]:
        while row and any(marker in row[0] for marker in WATERMARK_MARKERS):
            row = row[1:]
        return row

    @staticmethod
    def _is_repeated_header(row: list[str], header: list[str]) -> bool:
        return len(row) > 3 and row[:3] == header[:3]

    @staticmethod
    def _starts_record(row: list[str]) -> bool:
        """
        A record always begins with ``full_name`` followed by ``first_name``,
        and the full name starts with the first name.  That is a strong,
        cheap anchor for telling a new record from a continuation fragment.
        """
        if len(row) < 9:
            return False
        full_name, first_name = row[0].strip(), row[1].strip()
        return bool(
            full_name
            and first_name
            and len(full_name) < 80
            and "@" not in full_name
            and full_name.startswith(first_name)
        )

    # -- Step 2: align values to columns ------------------------------------
    @staticmethod
    def _score(column: str, value: str) -> float:
        if value == "":
            return SCORE_EMPTY
        validator = FIELD_VALIDATORS.get(column)
        if validator is None:
            return SCORE_FREE_TEXT
        try:
            verdict = validator(value)
        except Exception:  # noqa: BLE001 - a broken value is just a mismatch
            verdict = False
        if verdict is None:
            return SCORE_MISMATCH
        return SCORE_MATCH if verdict else SCORE_MISMATCH

    def _align(self, values: list[str], header: list[str]) -> dict[str, str]:
        """Needleman-Wunsch alignment of row values against header columns."""
        n, m = len(values), len(header)
        neg_inf = float("-inf")
        best = [[neg_inf] * (m + 1) for _ in range(n + 1)]
        move = [[0] * (m + 1) for _ in range(n + 1)]
        best[0][0] = 0.0

        for j in range(1, m + 1):
            best[0][j] = best[0][j - 1] + PENALTY_MISSING_COLUMN
            move[0][j] = 2
        for i in range(1, n + 1):
            best[i][0] = best[i - 1][0] + self._drop_penalty(values[i - 1])
            move[i][0] = 3

        for i in range(1, n + 1):
            value = values[i - 1]
            drop = self._drop_penalty(value)
            for j in range(1, m + 1):
                score = best[i - 1][j - 1] + self._score(header[j - 1], value)
                choice = 1
                skip_column = best[i][j - 1] + PENALTY_MISSING_COLUMN
                if skip_column > score:
                    score, choice = skip_column, 2
                drop_value = best[i - 1][j] + drop
                if drop_value > score:
                    score, choice = drop_value, 3
                best[i][j], move[i][j] = score, choice

        record: dict[str, str] = {}
        i, j = n, m
        while i > 0 or j > 0:
            choice = move[i][j]
            if choice == 1:
                record[header[j - 1]] = values[i - 1].strip()
                i, j = i - 1, j - 1
            elif choice == 2:
                record.setdefault(header[j - 1], "")
                j -= 1
            else:
                i -= 1
        return record

    @staticmethod
    def _drop_penalty(value: str) -> float:
        return PENALTY_DROP_EMPTY if value == "" else PENALTY_DROP_VALUE

    # -- Step 3: final validation ------------------------------------------
    @staticmethod
    def _sanitize(record: dict[str, str], report: ImportReport) -> None:
        for column, value in list(record.items()):
            if not value:
                continue
            validator = FIELD_VALIDATORS.get(column)
            if validator is None:
                continue
            try:
                verdict = validator(value)
            except Exception:  # noqa: BLE001
                verdict = False
            if not verdict:
                record[column] = ""
                report.discarded_values += 1

    @staticmethod
    def _identity(record: dict[str, str]) -> str:
        return (
            record.get("linkedin_id") or record.get("linkedin_url") or record.get("full_name") or ""
        ).lower()


def parse_literal(value: str | None, default: Any) -> Any:
    """Public helper for the importer: parse a nested column defensively."""
    if not value:
        return default
    parsed = _safe_literal(value)
    return default if parsed is None else parsed


def read_dataset(path: str | Path) -> Iterator[dict[str, str]]:
    """Convenience wrapper for callers that do not need the report."""
    records, _ = DatasetReader(path).read()
    yield from records