749 lines
27 KiB
Python
749 lines
27 KiB
Python
import gzip
|
||
import hashlib
|
||
import json
|
||
import re
|
||
import time
|
||
from datetime import datetime, timezone
|
||
|
||
import requests
|
||
from sqlalchemy import inspect, select
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.config import Settings
|
||
from app.models import (
|
||
CrawlError,
|
||
CrawlRun,
|
||
CrawlRunEmployeeChange,
|
||
Employee,
|
||
EmployeeNewsLink,
|
||
EmployeePublication,
|
||
EmployeeProfileUrl,
|
||
EmployeeSnapshot,
|
||
ParserSource,
|
||
ProfileTab,
|
||
)
|
||
from app.parser.collector import collect_profile_links
|
||
from app.parser.profile import parse_person_profile
|
||
from app.parser.profile_url import profile_key
|
||
from app.services.academic_degrees import academic_degrees
|
||
from app.services.resource_cache import ResourceCache
|
||
|
||
HEADERS = {
|
||
"User-Agent": "Mozilla/5.0 (compatible; MIEMEmployeesBot/0.1.0; +https://miem.hse.ru/)"
|
||
}
|
||
|
||
|
||
def run_crawl(db: Session, settings: Settings) -> CrawlRun:
|
||
source = _ensure_source(db, settings.source_url)
|
||
run = CrawlRun(source_url=source.source_url, status="running")
|
||
db.add(run)
|
||
db.commit()
|
||
db.refresh(run)
|
||
|
||
found_keys: set[str] = set()
|
||
parsed_count = 0
|
||
skipped_count = 0
|
||
try:
|
||
with requests.Session() as session:
|
||
resource_cache = ResourceCache(db)
|
||
urls = collect_profile_links(session, source.source_url, HEADERS, settings.request_timeout)
|
||
if settings.crawl_limit:
|
||
urls = urls[: settings.crawl_limit]
|
||
run.found_count = len(urls)
|
||
db.commit()
|
||
|
||
for url in urls:
|
||
key = profile_key(url)
|
||
if key:
|
||
found_keys.add(key)
|
||
try:
|
||
parsed = parse_person_profile(
|
||
session,
|
||
url,
|
||
HEADERS,
|
||
settings.request_timeout,
|
||
settings.parser_use_playwright,
|
||
resource_cache=resource_cache,
|
||
)
|
||
if not parsed:
|
||
continue
|
||
employee, changed = _upsert_employee(db, run, parsed)
|
||
if employee.profile_key:
|
||
found_keys.add(employee.profile_key)
|
||
if changed:
|
||
parsed_count += 1
|
||
else:
|
||
skipped_count += 1
|
||
run.parsed_count = parsed_count
|
||
run.skipped_count = skipped_count
|
||
db.commit()
|
||
except Exception as exc:
|
||
run.error_count += 1
|
||
db.add(
|
||
CrawlError(
|
||
crawl_run_id=run.id,
|
||
profile_url=url,
|
||
error_type=type(exc).__name__,
|
||
message=str(exc),
|
||
)
|
||
)
|
||
db.commit()
|
||
finally:
|
||
time.sleep(settings.request_delay_seconds)
|
||
|
||
run.dismissed_count = _mark_dismissed(
|
||
db,
|
||
run,
|
||
found_keys,
|
||
session,
|
||
settings.request_timeout,
|
||
confirmation_runs=settings.dismissal_confirmation_runs,
|
||
max_auto_dismissals=settings.max_auto_dismissals_per_run,
|
||
)
|
||
run.status = "completed"
|
||
except Exception as exc:
|
||
run.status = "failed"
|
||
run.message = str(exc)
|
||
finally:
|
||
run.finished_at = datetime.now(timezone.utc)
|
||
db.commit()
|
||
db.refresh(run)
|
||
return run
|
||
|
||
|
||
def refresh_dismissed_status(db: Session, settings: Settings) -> CrawlRun:
|
||
source = _ensure_source(db, settings.source_url)
|
||
run = CrawlRun(source_url=source.source_url, status="running")
|
||
db.add(run)
|
||
db.commit()
|
||
db.refresh(run)
|
||
|
||
try:
|
||
employees = db.scalars(select(Employee).where(Employee.status == "dismissed")).all()
|
||
run.found_count = len(employees)
|
||
with requests.Session() as session:
|
||
urls = collect_profile_links(session, source.source_url, HEADERS, settings.request_timeout)
|
||
source_keys = {key for url in urls if (key := profile_key(url))}
|
||
now = datetime.now(timezone.utc)
|
||
for employee in employees:
|
||
if employee.profile_key not in source_keys:
|
||
run.skipped_count += 1
|
||
continue
|
||
employee.status = "active"
|
||
employee.dismissed_at = None
|
||
employee.last_seen_at = now
|
||
employee.profile_unavailable_streak = 0
|
||
employee.last_profile_check_at = now
|
||
_record_employee_change(
|
||
db,
|
||
run,
|
||
employee,
|
||
"reactivated",
|
||
profile_available=True,
|
||
message="Сотрудник снова найден в исходном списке.",
|
||
)
|
||
run.parsed_count += 1
|
||
run.status = "completed"
|
||
except Exception as exc:
|
||
run.status = "failed"
|
||
run.error_count = 1
|
||
run.message = str(exc)
|
||
db.add(
|
||
CrawlError(
|
||
crawl_run_id=run.id,
|
||
profile_url=source.source_url,
|
||
error_type=type(exc).__name__,
|
||
message=str(exc),
|
||
)
|
||
)
|
||
finally:
|
||
run.finished_at = datetime.now(timezone.utc)
|
||
db.commit()
|
||
db.refresh(run)
|
||
return run
|
||
|
||
|
||
def refresh_employee(db: Session, employee: Employee, settings: Settings) -> CrawlRun:
|
||
run = CrawlRun(source_url=employee.canonical_url, status="running", found_count=1)
|
||
db.add(run)
|
||
db.commit()
|
||
db.refresh(run)
|
||
|
||
try:
|
||
with requests.Session() as session:
|
||
resource_cache = ResourceCache(db)
|
||
parsed = parse_person_profile(
|
||
session,
|
||
employee.canonical_url,
|
||
HEADERS,
|
||
settings.request_timeout,
|
||
settings.parser_use_playwright,
|
||
resource_cache=resource_cache,
|
||
)
|
||
if not parsed:
|
||
raise ValueError("Профиль не удалось распарсить.")
|
||
if _parsed_profile_key(parsed) != employee.profile_key:
|
||
raise ValueError("Распарсенный профиль не совпадает с обновляемым сотрудником.")
|
||
|
||
_, changed = _upsert_employee(db, run, parsed)
|
||
if changed:
|
||
run.parsed_count = 1
|
||
else:
|
||
run.skipped_count = 1
|
||
run.status = "completed"
|
||
except Exception as exc:
|
||
run.status = "failed"
|
||
run.error_count = 1
|
||
run.message = str(exc)
|
||
db.add(
|
||
CrawlError(
|
||
crawl_run_id=run.id,
|
||
profile_url=employee.canonical_url,
|
||
error_type=type(exc).__name__,
|
||
message=str(exc),
|
||
)
|
||
)
|
||
finally:
|
||
run.finished_at = datetime.now(timezone.utc)
|
||
db.commit()
|
||
db.refresh(run)
|
||
return run
|
||
|
||
|
||
def _ensure_source(db: Session, source_url: str) -> ParserSource:
|
||
source = db.scalar(select(ParserSource).where(ParserSource.source_url == source_url))
|
||
if source:
|
||
return source
|
||
source = ParserSource(source_url=source_url, enabled=True)
|
||
db.add(source)
|
||
db.commit()
|
||
db.refresh(source)
|
||
return source
|
||
|
||
|
||
def _parsed_profile_key(parsed: dict) -> str:
|
||
return f"{parsed.get('profile_type')}:{parsed.get('profile_id')}"
|
||
|
||
|
||
def _upsert_employee(db: Session, run: CrawlRun, parsed: dict) -> tuple[Employee, bool]:
|
||
html = parsed.pop("_html", None)
|
||
parsed.pop("_resource_manifest", None)
|
||
checksum = _checksum(parsed)
|
||
key = _parsed_profile_key(parsed)
|
||
employee = db.scalar(select(Employee).where(Employee.profile_key == key))
|
||
if not employee:
|
||
employee = _find_employee_with_moved_profile(db, parsed)
|
||
now = datetime.now(timezone.utc)
|
||
if not employee:
|
||
employee = Employee(
|
||
profile_key=key,
|
||
profile_type=parsed.get("profile_type"),
|
||
profile_id=parsed.get("profile_id"),
|
||
canonical_url=parsed["source_url"],
|
||
first_seen_at=now,
|
||
)
|
||
db.add(employee)
|
||
run.new_count += 1
|
||
is_new = True
|
||
else:
|
||
is_new = False
|
||
|
||
parser_version = parsed.get("parser_version")
|
||
changed = is_new or employee.current_checksum != checksum or employee.parser_version != parser_version
|
||
previous_url = employee.canonical_url if employee.canonical_url != parsed["source_url"] else None
|
||
employee.profile_key = key
|
||
employee.profile_type = parsed.get("profile_type")
|
||
employee.profile_id = parsed.get("profile_id")
|
||
employee.canonical_url = parsed["source_url"]
|
||
employee.full_name = parsed.get("full_name")
|
||
employee.status = "active"
|
||
employee.last_seen_at = now
|
||
employee.dismissed_at = None
|
||
employee.profile_unavailable_streak = 0
|
||
employee.last_profile_check_at = now
|
||
employee.parser_version = parser_version
|
||
employee.has_academic_degree = bool(academic_degrees(parsed))
|
||
if changed:
|
||
employee.current_data = parsed
|
||
employee.current_checksum = checksum
|
||
db.flush()
|
||
_sync_profile_url_history(db, employee, previous_url, employee.canonical_url, now)
|
||
|
||
if is_new:
|
||
_record_employee_change(
|
||
db,
|
||
run,
|
||
employee,
|
||
"new",
|
||
profile_available=True,
|
||
message="Сотрудник впервые найден в источнике.",
|
||
)
|
||
|
||
if changed:
|
||
db.query(ProfileTab).filter(ProfileTab.employee_id == employee.id).delete()
|
||
for tab in parsed.get("tabs") or []:
|
||
db.add(
|
||
ProfileTab(
|
||
employee_id=employee.id,
|
||
title=tab.get("title") or "",
|
||
href=tab.get("href") or "",
|
||
data_index=tab.get("data_index"),
|
||
)
|
||
)
|
||
|
||
db.add(
|
||
EmployeeSnapshot(
|
||
employee_id=employee.id,
|
||
crawl_run_id=run.id,
|
||
parsed_data=parsed,
|
||
html_snapshot=gzip.compress(html.encode("utf-8")) if html else None,
|
||
checksum=checksum,
|
||
parser_version=parser_version,
|
||
)
|
||
)
|
||
db.flush()
|
||
_try_sync_employee_publications(db, run, employee, parsed)
|
||
_try_sync_employee_news_links(db, run, employee, parsed)
|
||
return employee, changed
|
||
|
||
|
||
def _find_employee_with_moved_profile(db: Session, parsed: dict) -> Employee | None:
|
||
full_name = parsed.get("full_name")
|
||
if not full_name:
|
||
return None
|
||
candidates = db.scalars(select(Employee).where(Employee.full_name == full_name)).all()
|
||
if len(candidates) == 1:
|
||
return candidates[0]
|
||
|
||
parsed_identity = _profile_identity_values(parsed)
|
||
if not parsed_identity:
|
||
return None
|
||
matches = [candidate for candidate in candidates if parsed_identity & _employee_identity_values(candidate)]
|
||
return matches[0] if len(matches) == 1 else None
|
||
|
||
|
||
def _profile_identity_values(profile: dict) -> set[str]:
|
||
if not isinstance(profile, dict):
|
||
return set()
|
||
values = set()
|
||
contacts = profile.get("contacts") or {}
|
||
if not isinstance(contacts, dict):
|
||
contacts = {}
|
||
for email in contacts.get("emails") or []:
|
||
normalized = str(email).strip().lower()
|
||
if normalized:
|
||
values.add(f"email:{normalized}")
|
||
for item in profile.get("external_ids") or []:
|
||
if not isinstance(item, dict):
|
||
continue
|
||
system = str(item.get("system") or "").strip().lower()
|
||
value = str(item.get("value") or "").strip().lower()
|
||
if system and value:
|
||
values.add(f"external:{system}:{value}")
|
||
return values
|
||
|
||
|
||
def _employee_identity_values(employee: Employee) -> set[str]:
|
||
return _profile_identity_values(employee.current_data or {})
|
||
|
||
|
||
def _sync_profile_url_history(
|
||
db: Session,
|
||
employee: Employee,
|
||
previous_url: str | None,
|
||
current_url: str,
|
||
seen_at: datetime,
|
||
) -> None:
|
||
urls = {url for url in (previous_url, current_url) if url}
|
||
for url in urls:
|
||
history = db.scalar(
|
||
select(EmployeeProfileUrl).where(
|
||
EmployeeProfileUrl.employee_id == employee.id,
|
||
EmployeeProfileUrl.url == url,
|
||
)
|
||
)
|
||
if history:
|
||
history.last_seen_at = seen_at
|
||
else:
|
||
db.add(
|
||
EmployeeProfileUrl(
|
||
employee_id=employee.id,
|
||
url=url,
|
||
first_seen_at=seen_at,
|
||
last_seen_at=seen_at,
|
||
)
|
||
)
|
||
|
||
|
||
def _try_sync_employee_publications(db: Session, run: CrawlRun, employee: Employee, parsed: dict) -> None:
|
||
try:
|
||
if not _publication_payloads(parsed):
|
||
return
|
||
if not _employee_publications_table_exists(db):
|
||
return
|
||
with db.begin_nested():
|
||
_sync_employee_publications(db, employee, parsed)
|
||
except Exception as exc:
|
||
db.add(
|
||
CrawlError(
|
||
crawl_run_id=run.id,
|
||
profile_url=employee.canonical_url,
|
||
error_type=type(exc).__name__,
|
||
message=f"Не удалось сохранить публикации сотрудника: {exc}",
|
||
)
|
||
)
|
||
|
||
|
||
def _employee_publications_table_exists(db: Session) -> bool:
|
||
return inspect(db.connection()).has_table(EmployeePublication.__tablename__)
|
||
|
||
|
||
def _sync_employee_publications(db: Session, employee: Employee, parsed: dict) -> None:
|
||
publications = _publication_payloads(parsed)
|
||
seen_hashes = set()
|
||
for publication in publications:
|
||
source_hash = _publication_hash(publication)
|
||
seen_hashes.add(source_hash)
|
||
publication_id = _clean_optional(publication.get("publication_id") or publication.get("id"))
|
||
existing = None
|
||
if publication_id:
|
||
existing = db.scalar(
|
||
select(EmployeePublication).where(
|
||
EmployeePublication.employee_id == employee.id,
|
||
EmployeePublication.publication_id == publication_id,
|
||
)
|
||
)
|
||
if not existing:
|
||
existing = db.scalar(
|
||
select(EmployeePublication).where(
|
||
EmployeePublication.employee_id == employee.id,
|
||
EmployeePublication.source_hash == source_hash,
|
||
)
|
||
)
|
||
if not existing:
|
||
existing = EmployeePublication(employee_id=employee.id, source_hash=source_hash, title=_publication_title(publication))
|
||
db.add(existing)
|
||
_apply_publication(existing, publication, source_hash)
|
||
|
||
if seen_hashes:
|
||
stale = db.scalars(
|
||
select(EmployeePublication).where(
|
||
EmployeePublication.employee_id == employee.id,
|
||
EmployeePublication.source_hash.not_in(seen_hashes),
|
||
)
|
||
).all()
|
||
for item in stale:
|
||
db.delete(item)
|
||
|
||
|
||
def _publication_payloads(parsed: dict) -> list[dict]:
|
||
publications = []
|
||
for section in parsed.get("sections") or []:
|
||
if not isinstance(section, dict) or section.get("type") != "publications":
|
||
continue
|
||
for publication in section.get("publications") or []:
|
||
if isinstance(publication, dict):
|
||
publications.append(publication)
|
||
return publications
|
||
|
||
|
||
def _apply_publication(target: EmployeePublication, publication: dict, source_hash: str) -> None:
|
||
target.publication_id = _clean_optional(publication.get("publication_id") or publication.get("id"))
|
||
target.title = _publication_title(publication)
|
||
target.year = _int_or_none(publication.get("year"))
|
||
target.publication_type = _clean_optional(publication.get("publication_type") or publication.get("type"))
|
||
target.language = _clean_optional(publication.get("language"))
|
||
target.status = _int_or_none(publication.get("status"))
|
||
target.url = _clean_optional(publication.get("url"))
|
||
target.doi_url = _clean_optional(publication.get("doi_url"))
|
||
target.other_url = _clean_optional(publication.get("other_url"))
|
||
target.document_url = _clean_optional(publication.get("document_url"))
|
||
target.citation_text = _clean_optional(publication.get("citation_text") or publication.get("text"))
|
||
target.annotation = publication.get("annotation") if isinstance(publication.get("annotation"), dict) else None
|
||
target.description = publication.get("description") if isinstance(publication.get("description"), dict) else None
|
||
target.authors = publication.get("authors") if isinstance(publication.get("authors"), list) else None
|
||
target.raw_data = publication.get("raw_data") if isinstance(publication.get("raw_data"), dict) else publication
|
||
target.source_hash = source_hash
|
||
|
||
|
||
def _publication_hash(publication: dict) -> str:
|
||
return _payload_hash(publication.get("raw_data") if isinstance(publication.get("raw_data"), dict) else publication)
|
||
|
||
|
||
def _payload_hash(value: object) -> str:
|
||
payload = json.dumps(_stable_checksum_payload(value), ensure_ascii=False, sort_keys=True, separators=(",", ":"), default=str)
|
||
return hashlib.sha256(payload.encode("utf-8")).hexdigest()
|
||
|
||
|
||
def _publication_title(publication: dict) -> str:
|
||
return _clean_optional(publication.get("title") or publication.get("text") or publication.get("id")) or "Untitled publication"
|
||
|
||
|
||
def _clean_optional(value: object) -> str | None:
|
||
text = str(value or "").strip()
|
||
return text or None
|
||
|
||
|
||
def _int_or_none(value: object) -> int | None:
|
||
try:
|
||
return int(value)
|
||
except (TypeError, ValueError):
|
||
return None
|
||
|
||
|
||
def _try_sync_employee_news_links(db: Session, run: CrawlRun, employee: Employee, parsed: dict) -> None:
|
||
try:
|
||
if not _news_link_payloads(parsed):
|
||
return
|
||
if not _employee_news_links_table_exists(db):
|
||
return
|
||
with db.begin_nested():
|
||
_sync_employee_news_links(db, employee, parsed)
|
||
except Exception as exc:
|
||
db.add(
|
||
CrawlError(
|
||
crawl_run_id=run.id,
|
||
profile_url=employee.canonical_url,
|
||
error_type=type(exc).__name__,
|
||
message=f"Не удалось сохранить новости сотрудника: {exc}",
|
||
)
|
||
)
|
||
|
||
|
||
def _employee_news_links_table_exists(db: Session) -> bool:
|
||
return inspect(db.connection()).has_table(EmployeeNewsLink.__tablename__)
|
||
|
||
|
||
def _sync_employee_news_links(db: Session, employee: Employee, parsed: dict) -> None:
|
||
news_links = _news_link_payloads(parsed)
|
||
seen_hashes = set()
|
||
for news_link in news_links:
|
||
source_hash = _news_link_hash(news_link)
|
||
seen_hashes.add(source_hash)
|
||
url = _clean_optional(news_link.get("url"))
|
||
existing = None
|
||
if url:
|
||
existing = db.scalar(
|
||
select(EmployeeNewsLink).where(
|
||
EmployeeNewsLink.employee_id == employee.id,
|
||
EmployeeNewsLink.url == url,
|
||
)
|
||
)
|
||
if not existing:
|
||
existing = db.scalar(
|
||
select(EmployeeNewsLink).where(
|
||
EmployeeNewsLink.employee_id == employee.id,
|
||
EmployeeNewsLink.source_hash == source_hash,
|
||
)
|
||
)
|
||
if not existing:
|
||
existing = EmployeeNewsLink(employee_id=employee.id, source_hash=source_hash, title=_news_link_title(news_link))
|
||
db.add(existing)
|
||
_apply_news_link(existing, news_link, source_hash)
|
||
|
||
if seen_hashes:
|
||
stale = db.scalars(
|
||
select(EmployeeNewsLink).where(
|
||
EmployeeNewsLink.employee_id == employee.id,
|
||
EmployeeNewsLink.source_hash.not_in(seen_hashes),
|
||
)
|
||
).all()
|
||
for item in stale:
|
||
db.delete(item)
|
||
|
||
|
||
def _news_link_payloads(parsed: dict) -> list[dict]:
|
||
news_links = []
|
||
for section in parsed.get("sections") or []:
|
||
if not isinstance(section, dict) or section.get("type") != "news":
|
||
continue
|
||
for item in section.get("news_links") or []:
|
||
if isinstance(item, dict):
|
||
news_links.append(item)
|
||
return news_links
|
||
|
||
|
||
def _apply_news_link(target: EmployeeNewsLink, news_link: dict, source_hash: str) -> None:
|
||
target.title = _news_link_title(news_link)
|
||
target.url = _clean_optional(news_link.get("url"))
|
||
target.summary = _clean_optional(news_link.get("summary"))
|
||
target.published_at = _datetime_or_none(news_link.get("published_at"))
|
||
target.published_year = _int_or_none(news_link.get("published_year"))
|
||
target.raw_data = news_link.get("raw_data") if isinstance(news_link.get("raw_data"), dict) else news_link
|
||
target.source_hash = source_hash
|
||
|
||
|
||
def _news_link_hash(news_link: dict) -> str:
|
||
return _payload_hash(news_link.get("raw_data") if isinstance(news_link.get("raw_data"), dict) else news_link)
|
||
|
||
|
||
def _news_link_title(news_link: dict) -> str:
|
||
return _clean_optional(news_link.get("title") or news_link.get("url")) or "Untitled news"
|
||
|
||
|
||
def _datetime_or_none(value: object) -> datetime | None:
|
||
if isinstance(value, datetime):
|
||
return value
|
||
if not value:
|
||
return None
|
||
try:
|
||
parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00"))
|
||
except ValueError:
|
||
return None
|
||
return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc)
|
||
|
||
|
||
def _mark_dismissed(
|
||
db: Session,
|
||
run: CrawlRun,
|
||
found_keys: set[str],
|
||
session: requests.Session,
|
||
timeout: int,
|
||
*,
|
||
confirmation_runs: int = 3,
|
||
max_auto_dismissals: int | None = 25,
|
||
) -> int:
|
||
dismissed = 0
|
||
candidates = db.scalars(
|
||
select(Employee).where(Employee.status.in_(("active", "verification_required")))
|
||
).all()
|
||
now = datetime.now(timezone.utc)
|
||
unavailable = []
|
||
for employee in candidates:
|
||
if employee.profile_key in found_keys:
|
||
continue
|
||
profile_available = _profile_check(session, employee.canonical_url, timeout)
|
||
employee.last_profile_check_at = now
|
||
if profile_available is None:
|
||
db.add(
|
||
CrawlError(
|
||
crawl_run_id=run.id,
|
||
profile_url=employee.canonical_url,
|
||
error_type="ProfileAvailabilityCheckError",
|
||
message="Не удалось надёжно проверить доступность профиля; статус сотрудника не изменён.",
|
||
)
|
||
)
|
||
continue
|
||
if profile_available:
|
||
employee.profile_unavailable_streak = 0
|
||
if employee.status == "verification_required":
|
||
employee.status = "active"
|
||
_record_employee_change(
|
||
db,
|
||
run,
|
||
employee,
|
||
"missing_from_source",
|
||
profile_available=True,
|
||
message="Профиль доступен, но ссылка отсутствует в исходном списке.",
|
||
)
|
||
continue
|
||
next_streak = employee.profile_unavailable_streak + 1
|
||
unavailable.append((employee, next_streak))
|
||
|
||
dismissal_blocked = bool(
|
||
max_auto_dismissals is not None and len(unavailable) > max_auto_dismissals
|
||
)
|
||
if dismissal_blocked:
|
||
run.message = (
|
||
f"Автоматическое увольнение приостановлено: {len(unavailable)} профилей "
|
||
f"одновременно не подтвердились (лимит {max_auto_dismissals})."
|
||
)
|
||
|
||
for employee, next_streak in unavailable:
|
||
employee.profile_unavailable_streak = next_streak
|
||
if next_streak < confirmation_runs or dismissal_blocked:
|
||
employee.status = "verification_required"
|
||
_record_employee_change(
|
||
db,
|
||
run,
|
||
employee,
|
||
"verification_required",
|
||
profile_available=False,
|
||
message=(
|
||
"Профиль не подтвердился. Автоматическое увольнение отложено до "
|
||
f"{confirmation_runs} последовательных проверок."
|
||
if not dismissal_blocked
|
||
else "Автоматическое увольнение отложено из-за массовой ошибки проверки профилей."
|
||
),
|
||
)
|
||
continue
|
||
|
||
employee.status = "dismissed"
|
||
employee.dismissed_at = now
|
||
_record_employee_change(
|
||
db,
|
||
run,
|
||
employee,
|
||
"dismissed",
|
||
profile_available=False,
|
||
message=(
|
||
"Сотрудник отсутствует в исходном списке, профиль не подтвердился "
|
||
f"{confirmation_runs} раза подряд."
|
||
),
|
||
)
|
||
dismissed += 1
|
||
db.commit()
|
||
return dismissed
|
||
|
||
|
||
def _profile_is_available(session: requests.Session, url: str, timeout: int) -> bool:
|
||
return _profile_check(session, url, timeout) is True
|
||
|
||
|
||
def _profile_check(session: requests.Session, url: str, timeout: int) -> bool | None:
|
||
try:
|
||
response = session.get(url, headers=HEADERS, timeout=timeout, allow_redirects=True)
|
||
if response.status_code < 400:
|
||
return True
|
||
if response.status_code in {404, 410}:
|
||
return False
|
||
return None
|
||
except requests.RequestException:
|
||
return None
|
||
|
||
|
||
def _record_employee_change(
|
||
db: Session,
|
||
run: CrawlRun,
|
||
employee: Employee,
|
||
change_type: str,
|
||
*,
|
||
profile_available: bool | None,
|
||
message: str,
|
||
) -> None:
|
||
db.add(
|
||
CrawlRunEmployeeChange(
|
||
crawl_run_id=run.id,
|
||
employee_id=employee.id,
|
||
profile_key=employee.profile_key,
|
||
profile_url=employee.canonical_url,
|
||
full_name=employee.full_name,
|
||
change_type=change_type,
|
||
profile_available=profile_available,
|
||
message=message,
|
||
)
|
||
)
|
||
|
||
|
||
def _checksum(data: dict) -> str:
|
||
payload = json.dumps(_stable_checksum_payload(data), ensure_ascii=False, sort_keys=True, separators=(",", ":"))
|
||
return hashlib.sha256(payload.encode("utf-8")).hexdigest()
|
||
|
||
|
||
def _stable_checksum_payload(value):
|
||
if isinstance(value, dict):
|
||
return {key: _stable_checksum_payload(item) for key, item in value.items()}
|
||
if isinstance(value, list):
|
||
return [_stable_checksum_payload(item) for item in value]
|
||
if isinstance(value, str):
|
||
return _normalize_date_dependent_experience(value)
|
||
return value
|
||
|
||
|
||
def _normalize_date_dependent_experience(value: str) -> str:
|
||
return re.sub(
|
||
r"(?i)(стаж(?:\s+работы)?(?:\s+в\s+ниу\s+вшэ|\s+в\s+вшэ)?\s*:?\s*)\d+\s*(?:год(?:а|ов)?|лет)",
|
||
r"\1<experience-years>",
|
||
value,
|
||
)
|