feat: add employee news links parsing and storage

app/services/admin_data.py

@@ -8,7 +8,7 @@ from zoneinfo import ZoneInfo
 from sqlalchemy import Select, Text, and_, desc, func, or_, select
 from sqlalchemy.orm import Session
 
-from app.models import CrawlError, CrawlRun, CrawlRunEmployeeChange, Employee
+from app.models import CrawlError, CrawlRun, CrawlRunEmployeeChange, Employee, EmployeeNewsLink
 
 EMPLOYEE_SORTS = {
     "full_name": Employee.full_name,
@@ -24,6 +24,7 @@ def employee_display_payload(employee: Employee) -> dict[str, Any]:
     data = _as_dict(employee.current_data)
     contacts = _as_dict(data.get("contacts"))
     sections = _as_list(data.get("sections"))
+    stored_news_links = _stored_news_links(employee)
     positions = _clean_list(data.get("positions"))
     emails = _clean_list(contacts.get("emails"))
     phones = _clean_list(contacts.get("phones"))
@@ -43,6 +44,7 @@ def employee_display_payload(employee: Employee) -> dict[str, Any]:
         "address": contacts.get("address"),
         "publications_count": _count_section_items(sections, "publications"),
         "courses_count": _count_section_items(sections, "courses_by_year"),
+        "news_count": len(stored_news_links) or _count_section_items(sections, "news"),
         "first_seen_at": employee.first_seen_at.isoformat() if employee.first_seen_at else None,
         "last_seen_at": employee.last_seen_at.isoformat() if employee.last_seen_at else None,
         "dismissed_at": employee.dismissed_at.isoformat() if employee.dismissed_at else None,
@@ -67,6 +69,7 @@ def employee_detail_payload(employee: Employee) -> dict[str, Any]:
             "contact_items": _normalize_contact_items(contacts.get("items")),
         },
         "external_ids": _normalize_external_ids(data.get("external_ids")),
+        "news_links": _detail_news_links(employee, data),
         "sections": [_normalize_section(section) for section in _as_list(data.get("sections"))],
     }
 
@@ -276,6 +279,8 @@ def _count_section_items(sections: list[dict[str, Any]], section_type: str) -> i
             total += len(section.get("publications") or section.get("items") or [])
         elif section_type == "courses_by_year":
             total += len(section.get("courses") or [])
+        elif section_type == "news":
+            total += len(section.get("news_links") or section.get("items") or [])
     return total
 
 
@@ -348,6 +353,8 @@ def _normalize_section(section: Any) -> dict[str, Any]:
         "year_entries": _normalize_year_entries(section.get("year_entries")),
         "publications": _normalize_publications(section.get("publications")),
         "publications_count": section.get("publications_count"),
+        "news_links": _normalize_news_links(section.get("news_links")),
+        "news_count": section.get("news_count"),
         "theses": _normalize_theses(section.get("theses")),
         "theses_count": section.get("theses_count"),
         "academic_year": section.get("academic_year"),
@@ -370,6 +377,77 @@ def _normalize_links(items: Any) -> list[dict[str, str | None]]:
     return normalized
 
 
+def _stored_news_links(employee: Employee) -> list[dict[str, Any]]:
+    return [_stored_news_link_payload(item) for item in sorted(employee.news_links, key=_news_link_sort_key)]
+
+
+def _news_link_sort_key(item: EmployeeNewsLink) -> tuple:
+    timestamp = item.published_at.timestamp() if item.published_at else 0
+    return (-timestamp, item.title or "", item.id)
+
+
+def _stored_news_link_payload(item: EmployeeNewsLink) -> dict[str, Any]:
+    return {
+        "title": item.title,
+        "url": item.url,
+        "summary": item.summary,
+        "published_at": item.published_at.isoformat() if item.published_at else None,
+        "published_year": item.published_year,
+        "published_display": format_admin_date(item.published_at) if item.published_at else str(item.published_year or ""),
+    }
+
+
+def _detail_news_links(employee: Employee, data: dict[str, Any]) -> list[dict[str, Any]]:
+    stored = _stored_news_links(employee)
+    if stored:
+        return stored
+    for section in _as_list(data.get("sections")):
+        if isinstance(section, dict) and section.get("type") == "news":
+            return _normalize_news_links(section.get("news_links"))
+    return []
+
+
+def format_admin_date(value: Any) -> str:
+    if not value:
+        return ""
+    if isinstance(value, str):
+        try:
+            value = datetime.fromisoformat(value.replace("Z", "+00:00"))
+        except ValueError:
+            return value
+    if not isinstance(value, datetime):
+        return str(value)
+    if value.tzinfo:
+        value = value.astimezone(ZoneInfo("Europe/Moscow"))
+    return value.strftime("%d.%m.%Y")
+
+
+def _normalize_news_links(items: Any) -> list[dict[str, Any]]:
+    normalized = []
+    if not isinstance(items, list):
+        return normalized
+    for item in items:
+        if not isinstance(item, dict):
+            continue
+        title = str(item.get("title") or item.get("url") or "").strip()
+        url = str(item.get("url") or "").strip()
+        summary = str(item.get("summary") or "").strip()
+        published_at = str(item.get("published_at") or "").strip()
+        published_year = item.get("published_year")
+        if title or url:
+            normalized.append(
+                {
+                    "title": title or url,
+                    "url": url or None,
+                    "summary": summary or None,
+                    "published_at": published_at or None,
+                    "published_year": published_year,
+                    "published_display": format_admin_date(published_at) if published_at else str(published_year or ""),
+                }
+            )
+    return normalized
+
+
 def _normalize_year_entries(items: Any) -> list[dict[str, Any]]:
     normalized = []
     if not isinstance(items, list):

app/services/crawler.py

@@ -15,6 +15,7 @@ from app.models import (
     CrawlRun,
     CrawlRunEmployeeChange,
     Employee,
+    EmployeeNewsLink,
     EmployeePublication,
     EmployeeSnapshot,
     ParserSource,
@@ -230,6 +231,7 @@ def _upsert_employee(db: Session, run: CrawlRun, parsed: dict) -> tuple[Employee
         )
     db.flush()
     _try_sync_employee_publications(db, run, employee, parsed)
+    _try_sync_employee_news_links(db, run, employee, parsed)
     return employee, changed
 
 
@@ -349,6 +351,108 @@ def _int_or_none(value: object) -> int | None:
         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) -> int:
     dismissed = 0
     active = db.scalars(select(Employee).where(Employee.status == "active")).all()