501 lines
20 KiB
Python
501 lines
20 KiB
Python
|
"""
|
||
|
|
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", "<unknown>"))
|
||
|
|
|
||
|
|
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
|