diff --git a/README.md b/README.md index d9ac005..f55c69f 100644 --- a/README.md +++ b/README.md @@ -116,4 +116,4 @@ docker compose exec postgres pg_dump -U miem miem_workers > backup.sql docker compose down ``` -Версия сервиса: `0.7.6`. Админка всегда показывает версии backend и frontend в footer. +Версия сервиса: `0.7.7`. Админка всегда показывает версии backend и frontend в footer. diff --git a/app/models.py b/app/models.py index 2772f91..df4a67a 100644 --- a/app/models.py +++ b/app/models.py @@ -162,7 +162,6 @@ class CrawlRun(Base): message: Mapped[str | None] = mapped_column(Text) employee_changes: Mapped[list["CrawlRunEmployeeChange"]] = relationship(back_populates="crawl_run") - dataset_versions: Mapped[list["DatasetVersion"]] = relationship(back_populates="crawl_run") class CrawlRunEmployeeChange(Base): @@ -242,42 +241,3 @@ class ParseResourceCache(Base): body_snapshot: Mapped[bytes] = mapped_column(LargeBinary, nullable=False) parser_version: Mapped[str | None] = mapped_column(String(32)) fetched_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow, nullable=False) - - -class DatasetVersion(Base): - __tablename__ = "dataset_versions" - __table_args__ = ( - UniqueConstraint("hash", name="uq_dataset_versions_hash"), - Index("ix_dataset_versions_created_at", "created_at"), - ) - - id: Mapped[int] = mapped_column(Integer, primary_key=True) - hash: Mapped[str] = mapped_column(String(64), nullable=False) - previous_hash: Mapped[str | None] = mapped_column(String(64)) - crawl_run_id: Mapped[int | None] = mapped_column(ForeignKey("crawl_runs.id")) - employee_count: Mapped[int] = mapped_column(Integer, default=0, nullable=False) - active_count: Mapped[int] = mapped_column(Integer, default=0, nullable=False) - dismissed_count: Mapped[int] = mapped_column(Integer, default=0, nullable=False) - created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow, nullable=False) - - crawl_run: Mapped[CrawlRun | None] = relationship(back_populates="dataset_versions") - items: Mapped[list["DatasetVersionItem"]] = relationship(back_populates="dataset_version", cascade="all, delete-orphan") - - -class DatasetVersionItem(Base): - __tablename__ = "dataset_version_items" - __table_args__ = ( - UniqueConstraint("dataset_version_id", "profile_key", name="uq_dataset_version_items_version_profile"), - Index("ix_dataset_version_items_hash", "dataset_version_id"), - Index("ix_dataset_version_items_profile_key", "profile_key"), - ) - - id: Mapped[int] = mapped_column(Integer, primary_key=True) - dataset_version_id: Mapped[int] = mapped_column(ForeignKey("dataset_versions.id"), nullable=False) - profile_key: Mapped[str] = mapped_column(String(255), nullable=False) - employee_id: Mapped[int | None] = mapped_column(ForeignKey("employees.id")) - status: Mapped[str] = mapped_column(String(32), nullable=False) - checksum: Mapped[str] = mapped_column(String(64), nullable=False) - - dataset_version: Mapped[DatasetVersion] = relationship(back_populates="items") - employee: Mapped[Employee | None] = relationship() diff --git a/app/services/crawler.py b/app/services/crawler.py index e458cb7..40074c1 100644 --- a/app/services/crawler.py +++ b/app/services/crawler.py @@ -26,7 +26,6 @@ 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.dataset_versions import get_or_create_current_version from app.services.resource_cache import ResourceCache HEADERS = { @@ -102,7 +101,6 @@ def run_crawl(db: Session, settings: Settings) -> CrawlRun: max_auto_dismissals=settings.max_auto_dismissals_per_run, ) run.status = "completed" - get_or_create_current_version(db, crawl_run_id=run.id) except Exception as exc: run.status = "failed" run.message = str(exc) @@ -146,7 +144,6 @@ def refresh_dismissed_status(db: Session, settings: Settings) -> CrawlRun: ) run.parsed_count += 1 run.status = "completed" - get_or_create_current_version(db, crawl_run_id=run.id) except Exception as exc: run.status = "failed" run.error_count = 1 @@ -194,7 +191,6 @@ def refresh_employee(db: Session, employee: Employee, settings: Settings) -> Cra else: run.skipped_count = 1 run.status = "completed" - get_or_create_current_version(db, crawl_run_id=run.id) except Exception as exc: run.status = "failed" run.error_count = 1 diff --git a/app/services/dataset_versions.py b/app/services/dataset_versions.py deleted file mode 100644 index 74bc59f..0000000 --- a/app/services/dataset_versions.py +++ /dev/null @@ -1,227 +0,0 @@ -import hashlib -import json -from dataclasses import dataclass - -from sqlalchemy import desc, select -from sqlalchemy.orm import Session - -from app.models import DatasetVersion, DatasetVersionItem, Employee - - -@dataclass(frozen=True) -class EmployeeMarker: - profile_key: str - employee_id: int | None - status: str - checksum: str - - -def get_or_create_current_version(db: Session, *, crawl_run_id: int | None = None) -> DatasetVersion: - employees = db.scalars(select(Employee).order_by(Employee.profile_key)).all() - markers = [_employee_marker(employee) for employee in employees] - dataset_hash = _dataset_hash(markers) - latest = get_latest_version(db) - if latest and latest.hash == dataset_hash: - return latest - - active_count = sum(1 for marker in markers if marker.status == "active") - dismissed_count = sum(1 for marker in markers if marker.status == "dismissed") - version = DatasetVersion( - hash=dataset_hash, - previous_hash=latest.hash if latest else None, - crawl_run_id=crawl_run_id, - employee_count=len(markers), - active_count=active_count, - dismissed_count=dismissed_count, - ) - db.add(version) - db.flush() - for marker in markers: - db.add( - DatasetVersionItem( - dataset_version_id=version.id, - profile_key=marker.profile_key, - employee_id=marker.employee_id, - status=marker.status, - checksum=marker.checksum, - ) - ) - db.flush() - return version - - -def get_latest_version(db: Session) -> DatasetVersion | None: - return db.scalar(select(DatasetVersion).order_by(desc(DatasetVersion.created_at), desc(DatasetVersion.id)).limit(1)) - - -def get_version_by_hash(db: Session, dataset_hash: str | None) -> DatasetVersion | None: - if not dataset_hash: - return None - return db.scalar(select(DatasetVersion).where(DatasetVersion.hash == dataset_hash).limit(1)) - - -def service_info_payload(db: Session, *, tools: list[dict], service_name: str, backend_version: str, protocol_version: str) -> dict: - version = get_or_create_current_version(db) - db.commit() - return { - "service_name": service_name, - "backend_version": backend_version, - "protocolVersion": protocol_version, - "tools": tools, - "dataset": _version_payload(version), - } - - -def sync_employees_payload(db: Session, *, client_hash: str | None = None, include_data: bool = True) -> dict: - current = get_or_create_current_version(db) - db.commit() - if not client_hash: - return _full_sync_payload(db, current, include_data=include_data, reason=None) - if client_hash == current.hash: - return { - "mode": "delta", - "from_hash": client_hash, - "to_hash": current.hash, - "dataset": _version_payload(current), - "changes": {"added": [], "updated": [], "dismissed": [], "removed": []}, - } - - previous = get_version_by_hash(db, client_hash) - if not previous: - return _full_sync_payload(db, current, include_data=include_data, reason="unknown_client_hash", from_hash=client_hash) - - return _delta_sync_payload(db, previous, current, include_data=include_data) - - -def _full_sync_payload( - db: Session, - current: DatasetVersion, - *, - include_data: bool, - reason: str | None, - from_hash: str | None = None, -) -> dict: - employees = db.scalars(select(Employee).order_by(Employee.profile_key)).all() - payload = { - "mode": "full", - "from_hash": from_hash, - "to_hash": current.hash, - "dataset": _version_payload(current), - "items": [_employee_payload(employee, include_data=include_data) for employee in employees], - } - if reason: - payload["reason"] = reason - return payload - - -def _delta_sync_payload(db: Session, previous: DatasetVersion, current: DatasetVersion, *, include_data: bool) -> dict: - previous_items = _items_by_profile_key(previous) - current_items = _items_by_profile_key(current) - employees = {employee.profile_key: employee for employee in db.scalars(select(Employee)).all()} - added = [] - updated = [] - dismissed = [] - removed = [] - - for profile_key, current_item in sorted(current_items.items()): - previous_item = previous_items.get(profile_key) - employee = employees.get(profile_key) - if not previous_item: - if employee: - added.append(_employee_payload(employee, include_data=include_data)) - continue - if previous_item.checksum == current_item.checksum and previous_item.status == current_item.status: - continue - if current_item.status == "dismissed": - dismissed.append(_tombstone(profile_key, current_item.status, employee)) - elif employee: - updated.append(_employee_payload(employee, include_data=include_data)) - - for profile_key, previous_item in sorted(previous_items.items()): - if profile_key not in current_items: - removed.append(_tombstone(profile_key, "removed", employees.get(profile_key), checksum=previous_item.checksum)) - - return { - "mode": "delta", - "from_hash": previous.hash, - "to_hash": current.hash, - "dataset": _version_payload(current), - "changes": { - "added": added, - "updated": updated, - "dismissed": dismissed, - "removed": removed, - }, - } - - -def _items_by_profile_key(version: DatasetVersion) -> dict[str, DatasetVersionItem]: - return {item.profile_key: item for item in version.items} - - -def _version_payload(version: DatasetVersion) -> dict: - return { - "hash": version.hash, - "previous_hash": version.previous_hash, - "created_at": version.created_at.isoformat() if version.created_at else None, - "crawl_run_id": version.crawl_run_id, - "employee_count": version.employee_count, - "active_count": version.active_count, - "dismissed_count": version.dismissed_count, - } - - -def _employee_marker(employee: Employee) -> EmployeeMarker: - return EmployeeMarker( - profile_key=employee.profile_key, - employee_id=employee.id, - status=employee.status, - checksum=employee.current_checksum or _payload_hash(employee.current_data or {}), - ) - - -def _dataset_hash(markers: list[EmployeeMarker]) -> str: - payload = [ - {"profile_key": marker.profile_key, "status": marker.status, "checksum": marker.checksum} - for marker in sorted(markers, key=lambda item: item.profile_key) - ] - return _payload_hash(payload) - - -def _payload_hash(value: object) -> str: - payload = json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), default=str) - return hashlib.sha256(payload.encode("utf-8")).hexdigest() - - -def _employee_payload(employee: Employee, *, include_data: bool) -> dict: - payload = { - "profile_key": employee.profile_key, - "profile_id": employee.profile_id, - "full_name": employee.full_name, - "status": employee.status, - "canonical_url": employee.canonical_url, - "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, - "checksum": employee.current_checksum or _payload_hash(employee.current_data or {}), - } - if include_data: - payload["data"] = employee.current_data - return payload - - -def _tombstone(profile_key: str, status: str, employee: Employee | None, *, checksum: str | None = None) -> dict: - payload = { - "profile_key": profile_key, - "status": status, - "checksum": checksum or (employee.current_checksum if employee else None), - } - if employee: - payload.update( - { - "profile_id": employee.profile_id, - "full_name": employee.full_name, - "canonical_url": employee.canonical_url, - "dismissed_at": employee.dismissed_at.isoformat() if employee.dismissed_at else None, - } - ) - return payload diff --git a/tests/test_api.py b/tests/test_api.py index 84b79a0..8465eff 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -20,6 +20,10 @@ def test_health_returns_versions(): assert response.json()["backend_version"] == "0.7.7" +def test_mcp_endpoint_is_removed(): + assert TestClient(app).get("/mcp").status_code == 404 + + def test_api_employees_and_stats_require_admin_session(): engine = create_engine("sqlite:///:memory:", connect_args={"check_same_thread": False}, poolclass=StaticPool) Base.metadata.create_all(engine) diff --git a/tests/test_dataset_versions.py b/tests/test_dataset_versions.py deleted file mode 100644 index f927795..0000000 --- a/tests/test_dataset_versions.py +++ /dev/null @@ -1,88 +0,0 @@ -from datetime import datetime, timezone - -from app.models import Employee -from app.services.dataset_versions import get_or_create_current_version, sync_employees_payload - - -def _employee(profile_key: str, checksum: str, *, status: str = "active") -> Employee: - return Employee( - profile_key=profile_key, - profile_type=profile_key.split(":", 1)[0], - profile_id=profile_key.split(":", 1)[1], - canonical_url=f"https://www.hse.ru/{profile_key}", - full_name=profile_key, - status=status, - first_seen_at=datetime.now(timezone.utc), - last_seen_at=datetime.now(timezone.utc), - current_data={"profile_key": profile_key}, - current_checksum=checksum, - ) - - -def test_dataset_version_hash_is_stable_for_same_employee_state(db_session): - db_session.add(_employee("staff:alpha", "a" * 64)) - db_session.commit() - - first = get_or_create_current_version(db_session) - db_session.commit() - second = get_or_create_current_version(db_session) - - assert second.id == first.id - assert second.hash == first.hash - assert second.employee_count == 1 - - -def test_dataset_version_hash_changes_when_employee_checksum_changes(db_session): - employee = _employee("staff:alpha", "a" * 64) - db_session.add(employee) - db_session.commit() - first = get_or_create_current_version(db_session) - db_session.commit() - - employee.current_checksum = "b" * 64 - db_session.commit() - second = get_or_create_current_version(db_session) - - assert second.hash != first.hash - assert second.previous_hash == first.hash - - -def test_sync_employees_diff_spans_multiple_intermediate_versions(db_session): - alpha = _employee("staff:alpha", "a" * 64) - db_session.add(alpha) - db_session.commit() - first = get_or_create_current_version(db_session) - db_session.commit() - - beta = _employee("staff:beta", "b" * 64) - db_session.add(beta) - db_session.commit() - get_or_create_current_version(db_session) - db_session.commit() - - alpha.current_checksum = "c" * 64 - alpha.current_data = {"profile_key": "staff:alpha", "changed": True} - db_session.commit() - - payload = sync_employees_payload(db_session, client_hash=first.hash, include_data=False) - - assert payload["mode"] == "delta" - assert [item["profile_key"] for item in payload["changes"]["added"]] == ["staff:beta"] - assert [item["profile_key"] for item in payload["changes"]["updated"]] == ["staff:alpha"] - assert payload["changes"]["dismissed"] == [] - assert payload["changes"]["removed"] == [] - - -def test_sync_employees_reports_dismissed_as_tombstone(db_session): - alpha = _employee("staff:alpha", "a" * 64) - db_session.add(alpha) - db_session.commit() - first = get_or_create_current_version(db_session) - db_session.commit() - - alpha.status = "dismissed" - db_session.commit() - payload = sync_employees_payload(db_session, client_hash=first.hash, include_data=False) - - assert payload["changes"]["dismissed"][0]["profile_key"] == "staff:alpha" - assert payload["changes"]["dismissed"][0]["status"] == "dismissed"