395 lines
14 KiB
Python
395 lines
14 KiB
Python
|
"""
|
||
|
|
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
|